Initial chekin post-discontinuity.
This commit is contained in:
0
pyoo/__init__.py
Normal file
0
pyoo/__init__.py
Normal file
122
pyoo/base.py
Normal file
122
pyoo/base.py
Normal file
@ -0,0 +1,122 @@
|
||||
"""
|
||||
Base data and types for supporting Pyoo operations.
|
||||
"""
|
||||
|
||||
from collections import namedtuple
|
||||
from typing import Callable, List, Tuple, Type, Sequence, Union, Iterable, Protocol
|
||||
|
||||
|
||||
PREPOSITIONS = (
|
||||
"with/using",
|
||||
"at/to",
|
||||
# 'in front of',
|
||||
"in/inside/into",
|
||||
# 'on top of/on/onto/upon',
|
||||
"on/onto/upon",
|
||||
# 'out of/from inside/from',
|
||||
"from",
|
||||
"over",
|
||||
"through",
|
||||
"under/underneath/beneath",
|
||||
"behind",
|
||||
"beside",
|
||||
"for/about",
|
||||
"is",
|
||||
"as",
|
||||
"off",
|
||||
)
|
||||
"""`tuple[str]`: A list of preposition strings (equivilancies are / separated)"""
|
||||
|
||||
NORMALIZED_PREPS = tuple([x.split("/") for x in PREPOSITIONS])
|
||||
"""A list of prepositions with equiviliants split."""
|
||||
|
||||
|
||||
class PyooError(Exception):
|
||||
"""Generic Pyoo error."""
|
||||
|
||||
|
||||
class PyooVerbNotFound(PyooError):
|
||||
"""Pyoo Error which indicates a verb was not found."""
|
||||
|
||||
|
||||
class PyooObjectNotFound(PyooError):
|
||||
"""Pyoo Error which indicates an object was not found."""
|
||||
|
||||
|
||||
VerbCallFrame = namedtuple("VerbCallFrame", "environment,player,verbname,dobj,dobjstr,prepstr,iobj,iobjstr,argstr")
|
||||
"""Namedtuple which contains the call frame (all arguments) for a verb call."""
|
||||
|
||||
|
||||
class Verb:
|
||||
"""This is a wrapper for a callable to give it the properties required for verbs."""
|
||||
def __init__(self,
|
||||
function: Callable,
|
||||
names: List[str],
|
||||
callspec: Tuple[str, List[str], str]):
|
||||
self.function = function
|
||||
self.name = names[0]
|
||||
self.names = names
|
||||
self.callspec = callspec
|
||||
|
||||
def __call__(self, *args, **kwargs) -> None:
|
||||
return self.function(*args, **kwargs)
|
||||
|
||||
def __get__(self, obj, objtype: Type):
|
||||
class VerbFunction:
|
||||
def __init__(self, func: Callable, obj, typ: Type, verb: "Verb"):
|
||||
self._func = func
|
||||
self._obj = obj
|
||||
self._typ = typ
|
||||
self.is_verb = True
|
||||
self.name = verb.name
|
||||
self.names = verb.names
|
||||
self.callspec = verb.callspec
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
return self._func.__get__(self._obj, self._typ)(*args, **kwargs)
|
||||
|
||||
return VerbFunction(self.function, obj, objtype, self)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Verb {self.name} {self.callspec}>"
|
||||
|
||||
|
||||
def make_verb(verbname: str, dobjspec: str, prepspec: str, iobjspec: str) -> Callable[..., Verb]:
|
||||
"""This simple decorator adds verb metadata to a method or function.
|
||||
|
||||
Arguments:
|
||||
verbname (str): List of verb names, comma separated, with wildcards.
|
||||
dobjspec (str): 'this' or 'that' or 'none' or 'any'
|
||||
(this = the object which defines the verb, that = an object in the soup, any = any string,
|
||||
none = blank)
|
||||
iobjspec (str): 'this' or 'that' or 'none' or 'any' (as above)
|
||||
prepspec (str): one of prepositions strings (as in the PREPOSITIONS list)
|
||||
"""
|
||||
def verb_decorate(verbfunc: Callable[[VerbCallFrame, ], None]) -> Verb:
|
||||
names = [x.strip() for x in verbname.split(",")]
|
||||
ps: List[str] = []
|
||||
if prepspec.find("/") > 0:
|
||||
ps = prepspec.split("/")
|
||||
else:
|
||||
for prep in NORMALIZED_PREPS:
|
||||
if prepspec in prep:
|
||||
ps = prep
|
||||
break
|
||||
ps = [prepspec]
|
||||
return Verb(verbfunc, names, (dobjspec, ps, iobjspec))
|
||||
|
||||
return verb_decorate
|
||||
|
||||
|
||||
#
|
||||
# type things
|
||||
#
|
||||
|
||||
|
||||
class InterpreterProtocol(Protocol):
|
||||
"""Defines minimal protocol for interpreter objects."""
|
||||
def update(self) -> None:
|
||||
...
|
||||
|
||||
|
||||
SingleMultiString = Union[Sequence[str], str, Iterable[str]]
|
||||
139
pyoo/interpret.py
Normal file
139
pyoo/interpret.py
Normal file
@ -0,0 +1,139 @@
|
||||
"""This module defines the basic interpreter system."""
|
||||
|
||||
import fnmatch
|
||||
|
||||
from typing import Optional, List, Set, Tuple, cast
|
||||
|
||||
from .things import Thing, Container, Place, Player
|
||||
from .base import VerbCallFrame, PyooVerbNotFound, PyooObjectNotFound
|
||||
|
||||
|
||||
class Interpreter:
|
||||
"""Base interpreter class.
|
||||
|
||||
Manages a collection of objects, and invoking verbs on them.
|
||||
"""
|
||||
def __init__(self, contents: Optional[List[Thing]] = None):
|
||||
"""Initialize an interpreter.
|
||||
|
||||
Arguments:
|
||||
contents (optional list of things): The initial contents to populate the interpreter with.
|
||||
|
||||
"""
|
||||
if contents is None:
|
||||
contents = []
|
||||
self.contents: Set[Thing] = set(contents)
|
||||
for cont in self.contents:
|
||||
cont.interpreter = self
|
||||
self.content_cache: List[Tuple[str, Thing]] = []
|
||||
self.update()
|
||||
|
||||
def add_player(self, player_object: Player) -> None:
|
||||
"""Add a player to the interpreter."""
|
||||
self.contents.add(player_object)
|
||||
player_object.interpreter = self
|
||||
self.update()
|
||||
|
||||
def remove_player(self, player_object: Player) -> None:
|
||||
"""Remove a player from the interpreter."""
|
||||
self.contents.remove(player_object)
|
||||
self.update()
|
||||
|
||||
def update(self) -> None:
|
||||
"""Do any updates required when objects are added or removed."""
|
||||
self.update_caches()
|
||||
|
||||
def update_caches(self) -> None:
|
||||
"""Update the caches for objects contained in the interpreter, as well as updating all of the caches
|
||||
on contained containers.
|
||||
"""
|
||||
self.content_cache = []
|
||||
for obj in self.contents:
|
||||
for name in obj.names:
|
||||
self.content_cache.append((name, obj))
|
||||
try:
|
||||
cast(Container, obj).update_caches()
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
def handle_move(self, newroom: Place, player: Player) -> None:
|
||||
"""Move a player to a new contained room."""
|
||||
if player.location:
|
||||
player.location.handle_exit(player)
|
||||
newroom.handle_enter(player)
|
||||
|
||||
def lookup_global_object(self, objstr: str) -> List[Tuple[str, Thing]]:
|
||||
"""Find an object within the soup by objstr."""
|
||||
return [x for x in self.content_cache if fnmatch.fnmatch(objstr, x[0])]
|
||||
|
||||
def lookup_object(self, player: Player, objstr: str) -> Optional[Thing]:
|
||||
"""Find an object relative to a player
|
||||
|
||||
This first checks the player's inventory, and then the container which contains the
|
||||
player.
|
||||
"""
|
||||
m: Optional[List[Tuple[str, Thing]]] = None
|
||||
try:
|
||||
m = player.get_name_matches(objstr)
|
||||
except PyooObjectNotFound:
|
||||
m = None
|
||||
if not m and player.location:
|
||||
m = player.location.get_name_matches(objstr)
|
||||
if m:
|
||||
return m[0][1]
|
||||
else:
|
||||
return None
|
||||
|
||||
def interpret(self, command: str, player: Player) -> None:
|
||||
"""Interpret a player's command by splitting it into components and finding matching verbs."""
|
||||
# FIXME make this better to support mulitple word objstrs and prepstr
|
||||
if not command:
|
||||
return
|
||||
|
||||
cmd_comps = command.split()
|
||||
verbstr = cmd_comps[0]
|
||||
dobjstr = ""
|
||||
prepstr = ""
|
||||
iobjstr = ""
|
||||
argstr = ""
|
||||
|
||||
try:
|
||||
argstr = " ".join(cmd_comps[1:])
|
||||
except IndexError:
|
||||
pass
|
||||
|
||||
try:
|
||||
dobjstr = cmd_comps[1]
|
||||
prepstr = cmd_comps[2]
|
||||
iobjstr = cmd_comps[3]
|
||||
except IndexError:
|
||||
pass
|
||||
|
||||
try:
|
||||
cmdmatches = player.get_command_matches(command)
|
||||
except PyooVerbNotFound:
|
||||
if player.location:
|
||||
cmdmatches = player.location.get_command_matches(command)
|
||||
else:
|
||||
raise PyooVerbNotFound
|
||||
cmd = cmdmatches[0]
|
||||
# glob = cmd[0]
|
||||
# comps = cmd[1]
|
||||
verb = cmd[2]
|
||||
this = cmd[3]
|
||||
|
||||
dobj = None
|
||||
if verb.callspec[0] == "this":
|
||||
dobj = this
|
||||
elif verb.callspec[0] == "that":
|
||||
# lookup object
|
||||
dobj = self.lookup_object(player, dobjstr)
|
||||
|
||||
iobj = None
|
||||
if verb.callspec[2] == "this":
|
||||
iobj = this
|
||||
elif verb.callspec[2] == "that":
|
||||
# lookp object
|
||||
iobj = self.lookup_object(player, iobjstr)
|
||||
|
||||
return verb(VerbCallFrame(self, player, verbstr, dobj, dobjstr, prepstr, iobj, iobjstr, argstr))
|
||||
87
pyoo/placeloader.py
Normal file
87
pyoo/placeloader.py
Normal file
@ -0,0 +1,87 @@
|
||||
"""PlaceLoader is a system to load simple plain text data to build a room-based space."""
|
||||
|
||||
from enum import Enum
|
||||
from typing import Dict, List, TextIO, Type
|
||||
|
||||
from .things import Place
|
||||
from .interpret import Interpreter
|
||||
|
||||
|
||||
class PlaceLoaderStates(Enum):
|
||||
START = 0
|
||||
DESC = 1
|
||||
EXITS = 2
|
||||
|
||||
|
||||
class PlaceLoader:
|
||||
"""
|
||||
A factory which loads a simple data format containing a description of rooms, and how
|
||||
they are connected, and produces room objects for each one.
|
||||
|
||||
The format is a plain text file containing a series of room definitions:
|
||||
|
||||
ROOMNAME
|
||||
DESCRIPTION (multiple lines)
|
||||
.
|
||||
direction,alias ROOMNAME2
|
||||
.
|
||||
ROOMNAME2 ...
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, in_file: TextIO, baseplace: Type[Place] = Place):
|
||||
"""Initialize the factory by loading a description file.
|
||||
|
||||
Arguments:
|
||||
in_file: A file to load the places from.
|
||||
baseplace: A type to initialize the rooms from.
|
||||
|
||||
"""
|
||||
self.base = baseplace
|
||||
self.places: Dict[str, Place] = {} # indexed by name
|
||||
|
||||
if in_file:
|
||||
self.process_file(in_file)
|
||||
|
||||
def process_file(self, in_file: TextIO) -> None:
|
||||
"""Load any room information from the passed in file.
|
||||
|
||||
Arguments:
|
||||
in_file: The file to load the rooms from.
|
||||
|
||||
"""
|
||||
state = PlaceLoaderStates.START
|
||||
desc: List[str] = []
|
||||
temp_exits: Dict[Place, List[str]] = {}
|
||||
for line in in_file:
|
||||
line = line.strip()
|
||||
if state == PlaceLoaderStates.START:
|
||||
rm = self.base(line)
|
||||
temp_exits[rm] = []
|
||||
state = PlaceLoaderStates.DESC
|
||||
desc = []
|
||||
elif state == PlaceLoaderStates.DESC:
|
||||
if line == ".":
|
||||
state = PlaceLoaderStates.EXITS
|
||||
rm.description = desc
|
||||
else:
|
||||
desc.append(line)
|
||||
elif state == PlaceLoaderStates.EXITS:
|
||||
if line == ".":
|
||||
state = PlaceLoaderStates.START
|
||||
self.places[rm.name] = rm
|
||||
else:
|
||||
temp_exits[rm].append(line)
|
||||
|
||||
# assemble the exits
|
||||
for place in list(self.places.values()):
|
||||
for ext in temp_exits[place]:
|
||||
names, destination = ext.split(" ", 1)
|
||||
for nm in names.split(","):
|
||||
place.ways[nm] = self.places[destination]
|
||||
place.update_go()
|
||||
|
||||
|
||||
def interpreter_from_placeloader(placeloader: PlaceLoader) -> Interpreter:
|
||||
"""Return an interpreter intialized with the contents of a placeloader."""
|
||||
return Interpreter(list(placeloader.places.values()))
|
||||
241
pyoo/things.py
Normal file
241
pyoo/things.py
Normal file
@ -0,0 +1,241 @@
|
||||
"""Base (Pyoo) classes which implement basic protocols."""
|
||||
|
||||
import fnmatch
|
||||
import itertools
|
||||
|
||||
from typing import cast, List, Optional, Tuple, Set, Dict
|
||||
|
||||
from .base import make_verb, PyooVerbNotFound, InterpreterProtocol, SingleMultiString, Verb, VerbCallFrame
|
||||
|
||||
|
||||
class Thing:
|
||||
"""The base of all Pyoo objects."""
|
||||
|
||||
def __init__(self, name: str, description: SingleMultiString = "A nondescript object."):
|
||||
"""Create a Thing
|
||||
|
||||
Arguments:
|
||||
name (str): A string of comma-separated names for this object.
|
||||
description (str): A string that describes this object.
|
||||
"""
|
||||
names = [x.strip() for x in name.split(",")]
|
||||
self.name = names[0]
|
||||
self.names: Tuple[str, ...] = tuple(names)
|
||||
self.description = description
|
||||
self.location: Optional["Container"] = None
|
||||
self.interpreter: Optional[InterpreterProtocol] = None
|
||||
|
||||
def tell(self, message: SingleMultiString) -> None:
|
||||
print("<tell stub>", self, message)
|
||||
|
||||
def verbs(self) -> List[Verb]:
|
||||
"""Return a list of bound methods which denote themselves as verbs."""
|
||||
verbs: List[Verb] = []
|
||||
for item in dir(self):
|
||||
try:
|
||||
v = self.__getattribute__(item)
|
||||
if v.is_verb:
|
||||
verbs.append(v)
|
||||
except AttributeError:
|
||||
continue
|
||||
return verbs
|
||||
|
||||
def verb_globs(self) -> List[Tuple[str, Tuple, Verb, "Thing"]]:
|
||||
"""Return a list of (globstr, bound method) where commands matching globstr should call method (given that
|
||||
'that' matches an object in the soup).
|
||||
"""
|
||||
verbglobs: List[Tuple[str, Tuple, Verb, "Thing"]] = []
|
||||
for vrb in self.verbs():
|
||||
vvars: List[Tuple] = [tuple(vrb.names)]
|
||||
if vrb.callspec[0] == "this":
|
||||
vvars.append(self.names)
|
||||
elif vrb.callspec[0] in ("that", "any"):
|
||||
vvars.append(("*",))
|
||||
|
||||
if vrb.callspec[1] != ["none"]:
|
||||
vvars.append(tuple(vrb.callspec[1]))
|
||||
|
||||
if vrb.callspec[2] == "this":
|
||||
vvars.append(self.names)
|
||||
elif vrb.callspec[2] in ("that", "any"):
|
||||
vvars.append(("*",))
|
||||
|
||||
for combo in itertools.product(*vvars):
|
||||
globstr = " ".join(combo)
|
||||
verbglobs.append((globstr, tuple(combo), vrb, self))
|
||||
|
||||
return verbglobs
|
||||
|
||||
def handle_move(self, newlocation: "Container") -> None:
|
||||
"""Handle moving this object to a new location.
|
||||
|
||||
Acts as a way for children to hook this occurance too.
|
||||
"""
|
||||
self.location = newlocation
|
||||
|
||||
def handle_remove(self, oldlocation: "Container") -> None:
|
||||
"""Handle removing this object from a container.
|
||||
|
||||
Acts as a way for children to hook this occurance too.
|
||||
"""
|
||||
self.location = None
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<Thing '%s' object at 0x%x>" % (self.name, self.__hash__())
|
||||
|
||||
|
||||
class Container(Thing):
|
||||
"""A Pyoo object which contains other Pyoo objects."""
|
||||
|
||||
def __init__(self, names: str, description: SingleMultiString = ""):
|
||||
"""Create a Container.
|
||||
|
||||
Arguments:
|
||||
names (str): A comma-separated list of names.
|
||||
description (str): A description for this object.
|
||||
|
||||
"""
|
||||
super().__init__(names, description)
|
||||
self.contents: Set[Thing] = set()
|
||||
self.name_cache: List[Tuple[str, Thing]] = []
|
||||
self.command_cache: List[Tuple[str, Tuple, Verb, Thing]] = []
|
||||
|
||||
def update_caches(self) -> None:
|
||||
"""Update the internal cache of contained object's verbs and names."""
|
||||
self.name_cache = []
|
||||
self.command_cache = []
|
||||
for obj in self.contents:
|
||||
for name in obj.names:
|
||||
self.name_cache.append((name, obj))
|
||||
for verbglob in obj.verb_globs():
|
||||
if verbglob[0][0] == "#":
|
||||
continue
|
||||
self.command_cache.append(verbglob)
|
||||
for verbglob in self.verb_globs():
|
||||
if verbglob[0][0] == "#":
|
||||
continue
|
||||
self.command_cache.append(verbglob)
|
||||
|
||||
def get_command_matches(self, command_spec: str) -> List[Tuple[str, Tuple, Verb, Thing]]:
|
||||
"""Return a list of commands which match a command_spec ordered by specificity.
|
||||
|
||||
Arguments:
|
||||
command_spec (str): A command
|
||||
|
||||
Returns:
|
||||
array[tuple]: A list of command specifiers.
|
||||
|
||||
"""
|
||||
res = [x for x in self.command_cache if fnmatch.fnmatch(command_spec, x[0])]
|
||||
# sort by ambiguity (percentage of *)
|
||||
res.sort(key=lambda a: a[0].count("*") / float(len(a[0])))
|
||||
if (len(res) < 1):
|
||||
raise PyooVerbNotFound
|
||||
return res
|
||||
|
||||
def get_name_matches(self, name: str) -> List[Tuple[str, Thing]]:
|
||||
"""Return a list of objects which match a name spec.
|
||||
|
||||
Arguments:
|
||||
name (str): A name of an object
|
||||
|
||||
Returns:
|
||||
array[tuple]: A list of name,thing tuples.
|
||||
|
||||
"""
|
||||
return [x for x in self.name_cache if fnmatch.fnmatch(name, x[0])]
|
||||
|
||||
def handle_exit(self, oldobj: Thing) -> None:
|
||||
"""Handle an object leaving the container.
|
||||
|
||||
This allows children to hook this occurence also.
|
||||
|
||||
Arguments:
|
||||
oldobj (`pyoo.things.Thing`): The object which is exiting.
|
||||
|
||||
"""
|
||||
self.contents.remove(oldobj)
|
||||
oldobj.handle_remove(self)
|
||||
|
||||
def handle_enter(self, newobj: Thing) -> None:
|
||||
"""Handle an object entering the container.
|
||||
|
||||
This allows children to hook this occurence also.
|
||||
|
||||
Arguments:
|
||||
newobj (`pyoo.things.Thing`): The object which is entering.
|
||||
|
||||
"""
|
||||
self.contents.add(newobj)
|
||||
newobj.handle_move(self)
|
||||
try:
|
||||
cast("Container", newobj).update_caches()
|
||||
except AttributeError:
|
||||
pass
|
||||
self.update_caches()
|
||||
|
||||
def handle_tell(self, msg: SingleMultiString, who: Set[Thing]) -> None:
|
||||
"""Handle processing a tell to a list of objects.
|
||||
|
||||
"""
|
||||
for obj in who:
|
||||
obj.tell(msg)
|
||||
|
||||
def tell(self, msg: SingleMultiString) -> None:
|
||||
"""Handle telling all content objects."""
|
||||
self.handle_tell(msg, self.contents)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<Container '%s' object at 0x%x>" % (self.name, self.__hash__())
|
||||
|
||||
|
||||
class Place(Container):
|
||||
"""A specialized container which models a place."""
|
||||
|
||||
def __init__(self, names: str, description: str = "") -> None:
|
||||
"""Create a place."""
|
||||
super().__init__(names, description)
|
||||
self.ways: Dict[str, Container] = dict()
|
||||
|
||||
def tell_only(self, message: SingleMultiString, verb_callframe: VerbCallFrame) -> None:
|
||||
self.handle_tell(message, self.contents - {verb_callframe.player})
|
||||
|
||||
# this verb expects to be annotated from update_go. We never want it ot be at the top of a match list by its deault
|
||||
# name either
|
||||
@make_verb("#go", "none", "none", "none")
|
||||
def go(self, verb_callframe: VerbCallFrame) -> None:
|
||||
self.do_go(verb_callframe.verbname, verb_callframe)
|
||||
|
||||
@make_verb("go,move,walk,run", "any", "none", "none")
|
||||
def go_dir(self, verb_callframe: VerbCallFrame) -> None:
|
||||
self.do_go(verb_callframe.dobjstr, verb_callframe)
|
||||
|
||||
def do_go(self, direction: str, verb_callframe: VerbCallFrame) -> None:
|
||||
if direction in self.ways:
|
||||
self.tell_only("{} moves {}".format(verb_callframe.player.name, direction), verb_callframe)
|
||||
verb_callframe.player.tell("You move {}".format(direction))
|
||||
verb_callframe.environment.handle_move(self.ways[direction], verb_callframe.player)
|
||||
|
||||
def handle_enter(self, newobj: Thing) -> None:
|
||||
super().handle_enter(newobj)
|
||||
self.handle_tell("{} arrives".format(newobj.name), self.contents - {newobj})
|
||||
|
||||
def update_go(self) -> None:
|
||||
# note does no actually remove items from the go verb in case the descender is overloading.
|
||||
# also note, the interpreter needs to have update() called after this is called.
|
||||
for direction in self.ways:
|
||||
if direction not in self.go.names:
|
||||
self.go.names.append(direction)
|
||||
if self.interpreter:
|
||||
self.interpreter.update()
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<Place '%s' object at 0x%x>" % (self.name, self.__hash__())
|
||||
|
||||
|
||||
class Player(Container):
|
||||
def __init__(self, names: str, description: str = ""):
|
||||
super().__init__(names, description)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<Player '%s' object at 0x%x>" % (self.name, self.__hash__())
|
||||
Reference in New Issue
Block a user
