Core Classes
This is the code reference for the Open Mafia Engine's core classes.
open_mafia_engine.core.game
Game
Defines the state of an entire game, including the execution context.
Parameters
gen_phases : Callable[[Game], AbstractPhaseSystem]
A function that creates a phase system, given a Game.
Use AbstractPhaseSystem.gen(*args, **kwargs) to create this function.
The default is SimplePhaseCycle.gen(), which creates a day-night cycle.
Attributes
event_engine : EventEngine The event engine, which handles all subscription and broadcasting. action_queue : ActionQueue The base action queue, which holds all core-level actions & all history. phase_system : AbstractPhaseSystem System that defines phases and their transitions. actors : List[Actor] All the actors (players) of the game. factions : List[Faction] All the factions (teams) of the game. aux : AuxHelper Helper object for auxiliary game objects.
add(self, obj: GameObject)
Adds the object to this game.
This is automatically called during obj.__init__()
Source code in core/game.py
def add(self, obj: GameObject):
"""Adds the object to this game.
This is automatically called during `obj.__init__()`
"""
if isinstance(obj, Actor):
if obj not in self._actors:
self._actors.append(obj)
elif isinstance(obj, Faction):
if obj not in self._factions:
self._factions.append(obj)
elif isinstance(obj, AuxObject):
self._aux.add(obj)
change_phase(self, new_phase: Optional[Phase] = None)
Changes the phase to the given one (or bumps it). This causes events.
Source code in core/game.py
def change_phase(self, new_phase: Optional[Phase] = None):
"""Changes the phase to the given one (or bumps it). This causes events."""
# NOTE: You can pass strings here, because ETryPhaseChange.__init__ converts
self.process_event(ETryPhaseChange(self, new_phase=new_phase))
load(file: Union[Path, str, IO[bytes]]) -> Game
classmethod
Cloudpickle-based loading
Source code in core/game.py
@classmethod
def load(cls, file: Union[Path, str, IO[bytes]]) -> Game:
"""Cloudpickle-based loading"""
if isinstance(file, Path):
with file.open(mode="rb") as f:
return cls.load(f)
elif isinstance(file, str):
return cls.load(Path(file))
else:
res = cloudpickle.load(file)
if not isinstance(res, cls):
raise TypeError(f"Wrong object was pickled: {file!r}")
return res
process_event(self, event: Event, *, process_now: bool = False)
Processes the action.
Source code in core/game.py
def process_event(self, event: Event, *, process_now: bool = False):
"""Processes the action."""
responses: List[Action] = self.event_engine.broadcast(event)
for resp in responses:
self.action_queue.enqueue(resp)
process_now = (
process_now
or self.current_phase.action_resolution == ActionResolutionType.instant
or isinstance(event, ETryPhaseChange)
)
if process_now:
self.action_queue.process_all()
save(self, file: Union[Path, str, IO[bytes]])
Cloudpickle-based saving
Source code in core/game.py
def save(self, file: Union[Path, str, IO[bytes]]):
"""Cloudpickle-based saving"""
if isinstance(file, Path):
with file.open(mode="wb") as f:
self.save(f)
elif isinstance(file, str):
return self.save(Path(file))
else:
cloudpickle.dump(self, file)
open_mafia_engine.core.game_object
GameObject
Base class for game objects.
GameObjectMeta
Metaclass for game objects.
inject_converters(func: Callable) -> Callable
Decorator that adds converters for all possible types.
Source code in core/game_object.py
def inject_converters(func: Callable) -> Callable:
"""Decorator that adds converters for all possible types."""
if hasattr(func, "__is_converting__"):
return func
@wraps(func)
def wrapper(*args, **kwargs):
# Work with signature
sig = inspect.signature(func)
sb = sig.bind(*args, **kwargs)
sb.apply_defaults() # we want to convert the default,s too!
# Get type hints
# FIXME: Unsure whether this will work for external subclasses.
type_hints = get_type_hints(func, localns=_get_ns())
game_param = sig.parameters.get("game")
if game_param is None:
self_param = sig.parameters.get("self")
if self_param is None:
raise TypeError(f"`game` or `self` are required, got {sig!r}")
self: GameObject = sb.arguments["self"]
if not isinstance(self, GameObject):
raise TypeError(f"`self` must be GameObject, got {self!r}")
game: Game = self.game
else:
game: Game = sb.arguments["game"]
params = list(sig.parameters.values())
nargs = []
nkw = {}
for p in params:
val = sb.arguments[p.name]
if p.kind == inspect.Parameter.VAR_POSITIONAL:
# TODO: special behavior for sequence of params
# e.g. f(game, *actors: List[Actor])
nargs.extend(val)
elif p.kind == inspect.Parameter.VAR_KEYWORD:
# TODO: special behavior for mapping of params
# e.g. f(game, **maps: Dict[str, Actor])
nkw.update(val)
else:
th = type_hints.get(p.name, _BAD_HINT)
if converter.can_convert_to(th):
val = converter.convert(game, th, val)
# except MafiaConverterError:
# # TODO: Pre-check instead?
# pass
if p.kind == inspect.Parameter.POSITIONAL_ONLY:
nargs.append(val)
elif p.name in sb.kwargs:
nkw[p.name] = val
else:
nargs.append(val)
res = func(*nargs, **nkw)
return res
setattr(wrapper, "__is_converting__", True) # FIXME
return wrapper
open_mafia_engine.core.event_system
Action
Core action object.
Attributes
game : Game source : GameObject The object that generated this action. priority : float canceled : bool
Abstract Methods
doit(self) -> None Performs the action.
Class Attributes
Pre : Type[EPreAction]
Post : Type[EPostAction]
Pre- and post-action event classes to use with action.pre and action.post.
You may override these with your own when subclassing.
Post
Post-action event.
Pre
Pre-action event.
doit(self)
Performs the action.
Source code in core/event_system.py
@abstractmethod
def doit(self):
"""Performs the action."""
generate(func: Callable, name: str = None, doc: str = None) -> Type[Action]
classmethod
Create an Action subtype from a function.
Parameters
func : Callable Function that does something. :) name : str What name to use. By default, will auto-generate a name from the func name. doc : str Docstring. By default, will use func doc, with prepended Action info.
Source code in core/event_system.py
@classmethod
def generate(
cls, func: Callable, name: str = None, doc: str = None
) -> Type[Action]:
"""Create an Action subtype from a function.
Parameters
----------
func : Callable
Function that does something. :)
name : str
What name to use. By default, will auto-generate a name from the func name.
doc : str
Docstring. By default, will use func doc, with prepended Action info.
"""
if name is None:
# NOTE: We can add random bits at the end to avoid conflicts
# But this might mess up serialization?
# from uuid import uuid4
# rand_ = str(uuid4()).replace("-", "")[-8:]
name = f"{cls.__name__}_{func.__name__}"
if doc is None:
doc = "(GENERATED ACTION) " + (func.__doc__ or "")
# Get all signatures, parameters to merge
sig_core = inspect.signature(cls.__init__)
params_core = list(sig_core.parameters.values())
sig_func = inspect.signature(func)
params_func = list(sig_func.parameters.values())
# Make sure the function has "self" as 0th arg
if (len(params_func) < 1) or (params_func[0].name != "self"):
raise TypeError(f"Function requires `self` argument to be first.")
# Give a hint via warnings :)
if params_func[0].annotation is None:
warnings.warn(
"We suggest you annotate like this to improve code editor experience:\n"
f" {func.__name__}(self: Action, ...)"
)
# Generate the list of params by merging
DEFAULTS = {k: v.default for k, v in sig_core.parameters.items()}
DEFAULTS.update({k: v.default for k, v in sig_func.parameters.items()})
params_res: List[inspect.Parameter] = []
attr_names = []
kinds = [
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
inspect.Parameter.VAR_POSITIONAL,
inspect.Parameter.KEYWORD_ONLY,
inspect.Parameter.VAR_KEYWORD,
]
i_core = 0
i_func = 0
for kind in kinds:
while (i_core < len(params_core)) and (params_core[i_core].kind == kind):
params_res.append(params_core[i_core])
i_core += 1
while (i_func < len(params_func)) and (params_func[i_func].kind == kind):
pfi = params_func[i_func]
conflict_params = [p for p in params_res if p.name == pfi.name]
if len(conflict_params) == 0:
params_res.append(pfi)
attr_names.append(pfi.name)
elif len(conflict_params) == 1:
# conflict_params[0].default = pfi.default, except can't set param!
conf = conflict_params[0]
replacement = conf.replace(default=pfi.default)
params_res[params_res.index(conf)] = replacement
# don't add it - will conflict
i_func += 1
# Note: we make sure that we don't have duplicate *args, **kwargs
sig_res = inspect.Signature(params_res)
@wraps(cls.__init__, new_sig=sig_res)
def __init__(
self,
game,
/,
*args,
priority: float = DEFAULTS.get("priority", 0.0),
canceled: bool = DEFAULTS.get("canceled", False),
**kwargs,
):
super(type(self), self).__init__(game, priority=priority, canceled=canceled)
# Set attributes
bs = sig_res.bind(
self, game, *args, priority=priority, canceled=canceled, **kwargs
) # FIXME: Should we pass in `game`?
bs.apply_defaults()
for attr_name in attr_names:
setattr(self, attr_name, bs.arguments[attr_name])
# NOTE: We can't keep the signature, because someone might change
# the arguments on the class instance; we have to "re-parse" instead
# Housekeeping
self._func = func
self._sig = sig_res
self._attr_names = tuple(attr_names)
def doit(self):
"""Performs the action (generated from function)."""
func = self._func
sig = self._sig
attr_names = self._attr_names
# "re-parse" the signature
args = []
kwargs = {}
for k, p in sig.parameters.items():
if k not in attr_names:
continue
v = getattr(self, k)
if p.kind in [
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
]:
args += v
elif p.kind == inspect.Parameter.VAR_POSITIONAL:
args.extend(v)
elif p.kind == inspect.Parameter.KEYWORD_ONLY:
kwargs[k] = v
elif p.kind == inspect.Parameter.VAR_KEYWORD:
kwargs.update(v)
func(self, *args, **kwargs)
GeneratedAction = type(
name, (cls,), {"__init__": __init__, "doit": doit, "__doc__": doc}
)
return GeneratedAction
post(self) -> EPostAction
Create a post-event for this action.
Source code in core/event_system.py
def post(self) -> EPostAction:
"""Create a post-event for this action."""
return self.Post(self.game, self)
pre(self) -> EPreAction
Create a pre-event for this action.
Source code in core/event_system.py
def pre(self) -> EPreAction:
"""Create a pre-event for this action."""
return self.Pre(self.game, self)
ActionInspector
Helper to inspect Action objects.
ignored_args: List[str]
property
readonly
Arguments that are ignored
param_names: List[str]
property
readonly
Parameter names.
type_hints: Dict[str, Type]
property
readonly
Type hints, without ignored arguments.
extract_value(self, param: str) -> Any
Gets value for the parameter.
Source code in core/event_system.py
def extract_value(self, param: str) -> Any:
"""Gets value for the parameter."""
# TODO: Make this smarter? :)
return getattr(self.action, param)
params_of_type(self, T: Type) -> List[str]
Returns parameter names that have the given type.
Source code in core/event_system.py
def params_of_type(self, T: Type) -> List[str]:
"""Returns parameter names that have the given type."""
def chk(v):
try:
return issubclass(v, T)
except Exception:
return False
return [k for k, v in self.type_hints.items() if chk(v)]
set_value(self, param: str, obj: Any)
Sets value for the parameter.
Source code in core/event_system.py
def set_value(self, param: str, obj: Any):
"""Sets value for the parameter."""
# TODO: Make this smarter? :)
if not hasattr(self.action, param):
warnings.warn(f"Action has no parameter {param!r}, setting anyways.")
return setattr(self.action, param, obj)
ActionQueue
Action queue.
queue : List[Action] Actions are sorted by decreasing priority. Ties are broken by insertion order. history : List[Action] Actions are stored in the order they were processed in, including sub-queues.
add_history(self, actions: List[Action])
Adds the actions to history.
Source code in core/event_system.py
def add_history(self, actions: List[Action]):
"""Adds the actions to history."""
self._history.extend(actions)
enqueue(self, action: Action)
Add an action to the queue.
Source code in core/event_system.py
def enqueue(self, action: Action):
"""Add an action to the queue."""
if not isinstance(action, Action):
raise TypeError(f"Expected Action, got {action!r}")
self._queue.add(action)
peek_batch(self) -> List[Action]
Peek at the next batch of actions, without removing them.
If there are no more actions, this returns an empty list.
Source code in core/event_system.py
def peek_batch(self) -> List[Action]:
"""Peek at the next batch of actions, without removing them.
If there are no more actions, this returns an empty list.
"""
if len(self._queue) == 0:
return []
res = []
i = 0
priority = self._queue[0].priority
while (i < len(self._queue)) and (self._queue[i].priority == priority):
res.append(self._queue[i])
return res
pop_batch(self) -> List[Action]
Gets the next batch of actions, removing them from the queue.
If there are no more actions, this returns an empty list.
Source code in core/event_system.py
def pop_batch(self) -> List[Action]:
"""Gets the next batch of actions, removing them from the queue.
If there are no more actions, this returns an empty list.
"""
if len(self._queue) == 0:
return []
res = []
priority = self._queue[0].priority
while (len(self._queue) > 0) and (self._queue[0].priority == priority):
action: Action = self._queue.pop(0)
res.append(action)
return res
CancelAction
Action that cancels other actions.
doit(self)
Performs the action.
Source code in core/event_system.py
def doit(self):
self.target.canceled = True
ConditionalCancelAction
Cancels the action, but checks condition just in case again.
If condition(action), actually does cancel the action.
doit(self)
Performs the action.
Source code in core/event_system.py
def doit(self):
if self.condition(self.target):
super().doit()
Constraint
Base class for constraints.
Override check and return a self.Violation("Description") or None.
Note that each Constraint is itself a Subscriber, so it technically can have its own constraints, but by default they're not used.
prefix_tags: List[str]
property
readonly
These are tags used for descriptions, as a prefix.
Violation
Constraint was violated.
check(self, action: Action) -> Optional[Constraint.Violation]
Checks whether the action is allowed.
If allowed, return None.
If not allowed, return a Constraint.Violation
Source code in core/event_system.py
@abstractmethod
def check(self, action: Action) -> Optional[Constraint.Violation]:
"""Checks whether the `action` is allowed.
If allowed, return None.
If not allowed, return a `Constraint.Violation`
"""
handler_post(event: EPostAction) -> Optional[List[Action]]
handler_pre(event: EPreAction) -> Optional[List[Action]]
hook_post_action(self, action: Action) -> Optional[List[Action]]
Hook called when parent successfully actioned.
Source code in core/event_system.py
def hook_post_action(self, action: Action) -> Optional[List[Action]]:
"""Hook called when parent successfully actioned."""
hook_pre_action(self, action: Action) -> Optional[List[Action]]
Hook called when parent is trying to action & no violation for self.
Source code in core/event_system.py
def hook_pre_action(self, action: Action) -> Optional[List[Action]]:
"""Hook called when parent is trying to action & no violation for self."""
EPostAction
Post-action event.
EPreAction
Pre-action event.
Event
Core event object.
EventEngine
Subscription and broadcasting engine.
add_handler(self, handler: EventHandler, parent: Subscriber) -> _HandlerFunc
Adds the handler, with given parent, to own subscribers.
Source code in core/event_system.py
def add_handler(self, handler: EventHandler, parent: Subscriber) -> _HandlerFunc:
"""Adds the handler, with given parent, to own subscribers."""
f = partial(handler.func, parent)
for etype in handler.etypes:
self._handlers[etype].append(f)
if parent not in self._handlers[etype]:
self._subscribers[etype].append(parent)
return f
broadcast(self, event: Event) -> List[Action]
Broadcasts event to all handlers.
Source code in core/event_system.py
def broadcast(self, event: Event) -> List[Action]:
"""Broadcasts event to all handlers."""
# Loop over superclasses, but make sure you don't repeat handlers
funcs = [] # NOTE: not using a set, because we want deterministic sorting
ET = type(event)
for T in ET.mro():
if issubclass(T, Event):
funcs += [h for h in self._handlers[T] if h not in funcs]
# Call each of the functions
res = []
for f in funcs:
x = f(event)
if x is None:
x = []
res.extend(x)
return res
remove_subscriber(self, sub: Subscriber)
Removes all subscriptions from the subscriber.
FIXME: This operation is very hackish and iterates over EVERYTHING. This can probably be fixed by adding back-references, somehow.
Source code in core/event_system.py
def remove_subscriber(self, sub: Subscriber):
"""Removes all subscriptions from the subscriber.
FIXME: This operation is very hackish and iterates over EVERYTHING.
This can probably be fixed by adding back-references, somehow.
"""
hfs = sub.handler_funcs
for etype in self._subscribers.keys():
try:
self._subscribers[etype].remove(sub)
for hf in hfs:
try:
self._handlers[etype].remove(hf)
except ValueError:
pass
except ValueError:
pass
sub._handler_funcs = []
EventHandler
Descriptor that implements event handling logic.
Subscriber
Base class for objects that listen to events.
Creating Event Handlers
Subclass from this and use a handler or handles decorator.
The following result in the same calls behavior:
1 2 3 4 5 6 7 8 | |
Adding Constraints
Constraints are added after the Subscriber is created.
use_default_constraints: bool
property
readonly
Whether this object is using default constraints for this class.
add_default_constraints(self)
Adds default constraints for this type. Override for your own types.
Source code in core/event_system.py
def add_default_constraints(self):
"""Adds default constraints for this type. Override for your own types."""
check_constraints(self, action: Action) -> List[Constraint.Violation]
Checks all constraints. All violations are returned.
Source code in core/event_system.py
def check_constraints(self, action: Action) -> List[Constraint.Violation]:
"""Checks all constraints. All violations are returned."""
res = []
for con in self._constraints:
r = con.check(action)
if r is not None:
res.append(r)
return res
get_handlers() -> List[EventHandler]
classmethod
Returns all event handlers for this class.
Source code in core/event_system.py
@classmethod
def get_handlers(cls) -> List[EventHandler]:
"""Returns all event handlers for this class."""
res = []
for T in cls.mro():
for x in T.__dict__.values():
if isinstance(x, EventHandler):
res.append(x)
return res
handler(func: _HandlerFunc) -> EventHandler
Decorator to automatically infer event handler.
Usage
Use this as a decorator with a mandatory type hint for event:
1 2 3 4 | |
Source code in core/event_system.py
def handler(func: _HandlerFunc) -> EventHandler:
"""Decorator to automatically infer event handler.
Usage
-----
Use this as a decorator with a mandatory type hint for `event`:
class A(Subscriber):
@handler
def f(self, event: Union[EPreAction, EPostAction]):
return []
"""
type_hints = get_type_hints(func)
th = type_hints.get("event")
if th is None:
raise TypeError("Type hint for 'event' is required; otherwise, use `handles()`")
if get_origin(th) is None:
if issubclass(th, Event):
return EventHandler(func, th)
elif get_origin(th) is Union:
etypes = get_args(th)
for a in etypes:
if not issubclass(a, Event):
raise TypeError(f"One of Union types is not Event: {th!r}")
return EventHandler(func, *etypes)
raise NotImplementedError(f"Unsupported type hint: {th!r}")
handles(*etypes: List[Event]) -> Callable[[_HandlerFunc], EventHandler]
Decorator factory, to handle events.
Usage
Use this as a decorator factory with the event types as arguments:
1 2 3 4 | |
Source code in core/event_system.py
def handles(*etypes: List[Event]) -> Callable[[_HandlerFunc], EventHandler]:
"""Decorator factory, to handle events.
Usage
-----
Use this as a decorator factory with the event types as arguments:
class A(Subscriber):
@handles(EPreAction, EPostAction)
def f(self, event) -> Optional[List[Action]]:
return None
"""
def _inner(func: _HandlerFunc) -> EventHandler:
return EventHandler(func, *etypes)
return _inner
open_mafia_engine.core.state
Ability
Basic Ability object.
Attributes
game owner : Actor name : str The Ability's name. desc : str Description. Default is "".
argument_names: List[str]
property
readonly
Names of arguments. User-facing.
activate(self, *args, **kwargs) -> Optional[List[Action]]
Activate this ability with some arguments.
Make the signature be the same as the Action's. If generated, it should match already.
Source code in core/state.py
@abstractmethod
def activate(self, *args, **kwargs) -> Optional[List[Action]]:
"""Activate this ability with some arguments.
Make the signature be the same as the Action's.
If generated, it should match already.
"""
generate(action_or_func: Union[Type[Action], Callable], params: List[str] = None, name: str = None, doc: str = None, desc: str = None) -> Type[Ability]
classmethod
Create an Ability subtype from an Action or function.
Parameters
action_or_func : Type[Action] or Callable If an Action subclass, uses it directly. If a callable (function), generates an action type and uses it. params : List[str] The names of parameters to leave as activation parameters. The rest will be taken as arguments for the Ability itself.
Source code in core/state.py
@classmethod
def generate(
cls,
action_or_func: Union[Type[Action], Callable],
params: List[str] = None,
name: str = None,
doc: str = None,
desc: str = None,
) -> Type[Ability]:
"""Create an Ability subtype from an Action or function.
Parameters
----------
action_or_func : Type[Action] or Callable
If an Action subclass, uses it directly.
If a callable (function), generates an action type and uses it.
params : List[str]
The names of parameters to leave as activation parameters.
The rest will be taken as arguments for the Ability itself.
"""
# Create an action type
if isinstance(action_or_func, type) and issubclass(action_or_func, Action):
TAction = action_or_func
elif callable(action_or_func):
gen_name: str = name if name else f"{action_or_func.__name__}_Action"
TAction = Action.generate(action_or_func, name=gen_name, doc=doc)
else:
raise TypeError(f"Expected Action or function, got {action_or_func!r}")
TAction: Type[Action]
# Fix input arguments
if params is None:
params = []
if name is None:
name = f"{cls.__name__}_{TAction.__name__}"
# TODO - add random bits to avoid conflict?
if doc is None:
doc = "(GENERATED ABILITY) " + (TAction.__doc__ or "")
# Split the signature into __init__() and activate() args
sig_action = inspect.signature(TAction.__init__)
par_action = list(sig_action.parameters.values())
par_activate = [
inspect.Parameter("self", kind=inspect.Parameter.POSITIONAL_OR_KEYWORD)
] + [p for p in par_action if p.name in params]
sig_activate = inspect.Signature(
par_activate, return_annotation=Optional[List[Action]]
)
sig_init = "TODO" # TODO: Actually create a proper init signature!
# TODO: Set sig_init
def __init__(self, game: Game, /, owner: Actor, name: str, desc: str = desc):
super().__init__(game, owner, name, desc=desc)
@with_signature(sig_activate)
def activate(self, *args, **kwargs) -> Optional[List[Action]]:
"""Activate this ability with some arguments."""
try:
return [self.TAction(self.game, self, *args, **kwargs)]
except Exception as e:
logger.exception("Error executing action:")
if False: # Set for debugging errors :)
raise
return None
GeneratedAbility = type(
name, (cls,), {"__init__": __init__, "activate": activate, "__doc__": doc}
)
GeneratedAbility.TAction = TAction
return GeneratedAbility
handle_activate(event: EActivate) -> Optional[List[Action]]
Actor
Actor object.
add(self, obj: Union[Ability, Trigger, Faction])
Adds
Source code in core/state.py
def add(self, obj: Union[Ability, Trigger, Faction]):
"""Adds """
if isinstance(obj, Ability):
self.add_ability(obj)
elif isinstance(obj, Trigger):
self.add_trigger(obj)
elif isinstance(obj, Faction):
warnings.warn("Better to do `Faction.add_actor(me)` directly.")
obj.add_actor(self)
else:
raise TypeError(f"Expected Ability or Trigger, got {obj!r}")
add_ability(self, ability: Ability)
Adds this ability to self, possibly removing the old owner.
Source code in core/state.py
def add_ability(self, ability: Ability):
"""Adds this ability to self, possibly removing the old owner."""
if not isinstance(ability, Ability):
raise TypeError(f"Expected Ability, got {ability!r}")
if ability in self._abilities:
return
self._abilities.append(ability)
if ability._owner is not self:
ability._owner._abilities.remove(ability)
ability._owner = self
add_trigger(self, trigger: Trigger)
Adds this trigger to self, possibly removing the old owner.
Source code in core/state.py
def add_trigger(self, trigger: Trigger):
"""Adds this trigger to self, possibly removing the old owner."""
if not isinstance(trigger, Trigger):
raise TypeError(f"Expected Trigger, got {trigger!r}")
if trigger in self._triggers:
return
self._triggers.append(trigger)
if trigger._owner is not self:
trigger._owner._triggers.remove(trigger)
trigger._owner = self
ATBase
Base object for abilities and triggers.
Attributes
game owner : Actor name : str The name of this object. desc : str Description. Default is "".
argument_names: List[str]
property
readonly
Names of arguments used when calling. Default is empty.
prefix_tags: List[str]
property
readonly
These are used for descriptions.
add_default_constraints(self)
Adds default constraints for this type. Override for your own types.
Source code in core/state.py
def add_default_constraints(self):
super().add_default_constraints()
# Default constraints for abilities and triggers:
ConstraintOwnerAlive(self.game, self)
ConstraintActorTargetsAlive(self.game, self)
full_description(self, full_constraints: bool = False, highlight_method: Callable = None) -> str
Gets the full description of the Action/Trigger.
If full_constraints, also adds the full descriptions of all constraints.
By default, though, it skips these, adding only the prefix tags.
Source code in core/state.py
def full_description(
self, full_constraints: bool = False, highlight_method: Callable = None
) -> str:
"""Gets the full description of the Action/Trigger.
If `full_constraints`, also adds the full descriptions of all constraints.
By default, though, it skips these, adding only the prefix tags.
"""
prestr = ", ".join(self.prefix_tags)
if prestr == "":
prestr = "Any"
arg_names = self.argument_names
abil_targs = ", ".join(arg_names)
if abil_targs != "":
abil_targs = f" ({abil_targs})"
if highlight_method is None:
name = self.name
else:
name = highlight_method(self.name)
res = f"[{prestr}] {name}{abil_targs}: {self.desc}"
if full_constraints:
fcd = []
# TODO: Long descriptions for constraints
fcd = [indent(x, ". ") for x in fcd]
res = "\n".join([res] + fcd)
return res
ATConstraint
Constraint for Actions and Triggers. Will raise if used elsewhere.
owner: Actor
property
readonly
The owner of this constraint's parent.
ConstraintActorTargetsAlive
Any targets for the action, if they are Actors, must be alive.
check(self, action: Action) -> Optional[Constraint.Violation]
Checks whether the action is allowed.
If allowed, return None.
If not allowed, return a Constraint.Violation
Source code in core/state.py
def check(self, action: Action) -> Optional[Constraint.Violation]:
ai = ActionInspector(action)
p2a: Dict[str, Actor] = ai.values_of_type(Actor)
bads = []
for p, a in p2a.items():
if a.status["dead"]: # default is None, which is falsy
bads.append(f"{p!r} ({a.name!r})")
if len(bads) > 0:
return self.Violation("Targets are dead: " + ", ".join(bads))
ConstraintOwnerAlive
The ability's (trigger's) owner must be alive.
check(self, action: Action) -> Optional[Constraint.Violation]
Checks whether the action is allowed.
If allowed, return None.
If not allowed, return a Constraint.Violation
Source code in core/state.py
def check(self, action: Action) -> Optional[Constraint.Violation]:
if self.owner.status["dead"]: # default is None, which is falsey
return self.Violation(f"{self.owner.name!r} (owner) is dead.")
EActivate
Event of ability activation.
That is, this event is triggered by a player trying to activate their Ability.
EStatusChange
The Status has changed for some Actor.
Faction
Faction, a.k.a. Alignment.
OutcomeChecker
Checks for Faction Outcome. Base class.
Status
dict-like representation of an actor's status.
Access of empty attribs gives None. Changing an attribute emits an EStatusChange event.
Attributes
parent: Actor attribs : dict Raw keyword arguments for the status.
Trigger
Basic Trigger object.
Attributes
game : Game owner : Actor name : str The Trigger's name. desc : str Description. Default is "".
prefix_tags: List[str]
property
readonly
These are used for descriptions.
open_mafia_engine.core.phase_cycle
AbstractPhaseSystem
Interface for a phase system.
It's possible to see all phases by using game.phase_system.possible_phases
current_phase: Phase
property
writable
Returns the current phase.
possible_phases: Iterable[Phase]
property
readonly
Returns all possible phases (as a new iterable).
If it is infinite, override getitem as well!
bump_phase(self) -> Phase
Updates the phase to use the next one, then returns the current one.
Source code in core/phase_cycle.py
@abstractmethod
def bump_phase(self) -> Phase:
"""Updates the phase to use the next one, then returns the current one."""
gen(*args, **kwargs) -> Callable[[Game], AbstractPhaseSystem]
classmethod
Create a callable that generates a phase cycle.
Source code in core/phase_cycle.py
@classmethod
def gen(cls, *args, **kwargs) -> Callable[[Game], AbstractPhaseSystem]:
"""Create a callable that generates a phase cycle."""
def func_gen(game: Game) -> AbstractPhaseSystem:
return cls(game, *args, **kwargs)
return func_gen
system_phase_change(event: ETryPhaseChange) -> Optional[List[PhaseChangeAction]]
EPostPhaseChange
Phase has changed.
EPrePhaseChange
Phase is about to change.
ETryPhaseChange
Try to change the phase.
Phase
Represents a monolithic "phase" of action.
Attributes
name : str The current phase name. action_resolution : ActionResolutionType One of {"instant", "end_of_phase"}
PhaseChangeAction
Action to change the phase.
Parameters
new_phase : None or Phase
The resulting phase. By default, None uses the next phase.
old_phase : Phase
The phase that this action was created in.
Post
Phase has changed.
Pre
Phase is about to change.
doit(self)
Performs the action.
Source code in core/phase_cycle.py
def doit(self):
if self.new_phase is None:
self.game.phase_system.bump_phase()
else:
self.game.phase_system.current_phase = self.new_phase
SimplePhaseCycle
Simple phase cycle definition.
Parameters
game : Game
cycle : None or List[Tuple[str, ActionResolutionType]]
The cycle definition. Submit pairs of (name, resolution_type).
By default, uses [("day", "instant"), ("night", "end_of_phase")]
current_phase: Phase
property
writable
Returns the current phase.
possible_phases: List[Phase]
property
readonly
Returns all possible phases (as a new iterable).
If it is infinite, override getitem as well!
bump_phase(self) -> Phase
Updates the phase to use the next one, then returns the current one.
Trying to bump on shutdown phase will be ignored.
Source code in core/phase_cycle.py
def bump_phase(self) -> Phase:
"""Updates the phase to use the next one, then returns the current one.
Trying to bump on `shutdown` phase will be ignored.
"""
if self._i == self._STARTUP:
self._i = 0
return self.current_phase
elif self._i == self._SHUTDOWN:
return self.current_phase
# raise ValueError(f"Cannot bump shutdown phase: {self.shutdown}")
self._i += 1
return self.current_phase
gen(cycle: List[Tuple[str, ActionResolutionType]] = None, current_phase: Optional[str] = None) -> Callable[[Game], SimplePhaseCycle]
classmethod
Generator for a simple phase cycle.
Source code in core/phase_cycle.py
@classmethod
def gen(
cls,
cycle: List[Tuple[str, ActionResolutionType]] = None,
current_phase: Optional[str] = None,
) -> Callable[[Game], SimplePhaseCycle]:
"""Generator for a simple phase cycle."""
return super().gen(cycle=cycle, current_phase=current_phase)
open_mafia_engine.core.auxiliary
AuxHelper
Auxiliary object helper.
add(self, obj: AuxObject)
Adds the aux object to self.
Source code in core/auxiliary.py
def add(self, obj: AuxObject):
"""Adds the aux object to self."""
if not isinstance(obj, AuxObject):
raise TypeError(f"Expected AuxObject, got {obj!r}")
if self._key_map.get(obj.key) == obj:
return
elif obj.key in self._key_map.keys():
# FIXME: different object with the same key!
return
elif obj in self._key_map.values():
# FIXME: same object with different key? Key should be able to change.
return
# OK, let's add it
if len(self) > self.max_objects:
raise ValueError(f"Reached {self.max_objects} (max) aux objects!")
self._key_map[obj.key] = obj
filter_by_type(self, T: Type[AuxObject]) -> List[AuxObject]
Returns all AuxObjects with the given type.
You can pass in Union types as well.
Source code in core/auxiliary.py
def filter_by_type(self, T: Type[AuxObject]) -> List[AuxObject]:
"""Returns all `AuxObject`s with the given type.
You can pass in Union types as well.
"""
if get_origin(T) == Union:
T_raw = get_args(T)
T = tuple(x for x in T_raw if x is not None)
# T is a tuple of types
def chk(x):
try:
return isinstance(x, T)
except Exception:
return False
return [x for x in self._key_map.values() if chk(x)]
remove(self, obj: AuxObject)
Removes obj from self.
Source code in core/auxiliary.py
def remove(self, obj: AuxObject):
"""Removes `obj` from self."""
if not isinstance(obj, AuxObject):
raise TypeError(f"Expected AuxObject, got {obj!r}")
found = self._key_map.get(obj.key)
if found is None:
return
elif found == obj:
del self._key_map[obj.key]
obj._unsub()
else:
# FIXME: Object has different key, or another object has same key
return
AuxObject
Base class for auxiliary objects.
Aux objects have a key they are retrievable by.
Attributes
game : Game key : str The key to retrieve the object by. If None (default), generates one.
generate_key() -> str
classmethod
Generates a key for this class (used if None is passed in init).
Source code in core/auxiliary.py
@classmethod
def generate_key(cls) -> str:
"""Generates a key for this class (used if None is passed in __init__)."""
return cls.__qualname__ + "_" + str(uuid4()).replace("-", "")
get_or_create(/, game: Game, key: str = None, *, use_default_constraints: bool = True, **kwargs) -> AuxObject
classmethod
Finds the AuxObject by key or, if there is no object, creates it.
Source code in core/auxiliary.py
@classmethod
def get_or_create(
cls,
game: Game,
/,
key: str = None,
*,
use_default_constraints: bool = True,
**kwargs,
) -> AuxObject:
"""Finds the AuxObject by key or, if there is no object, creates it."""
if key is None:
key = cls.generate_key()
res = game.aux.get(key)
if res is None:
res = cls(
game, key=key, use_default_constraints=use_default_constraints, **kwargs
)
if not isinstance(res, cls):
raise TypeError(f"Wrong type for AuxObject: expected {cls!r}, got {res!r}")
return res
RemoveAuxAction
Removes the aux object, at the end of the stack.
doit(self)
Performs the action.
Source code in core/auxiliary.py
def doit(self):
self.game.aux.remove(self.target)
open_mafia_engine.core.outcome
EOutcomeAchieved
An outcome has been achieved.
OutcomeAction
A Faction achieves victory or defeat.
Post
An outcome has been achieved.
Pre
A Faction is about to achieve an Outcome.
doit(self)
Sets the "outcome" status for each Actor in Faction?
Source code in core/outcome.py
def doit(self):
"""Sets the "outcome" status for each Actor in Faction?"""
for actor in self.faction.actors:
actor.status["outcome"] = self.outcome
open_mafia_engine.core.ender
EGameEnded
The game has ended.
outcome_for(self, faction: Faction) -> Outcome
Returns the outcome for a given faction.
Source code in core/ender.py
@inject_converters
def outcome_for(self, faction: Faction) -> Outcome:
"""Returns the outcome for a given faction."""
return self.outcomes[faction.name]
EndTheGame
Action that ends the game.
Maybe this should inherit from PhaseChangeAction? I'm skeptical. Ending the game is significantly different.
Post
The game has ended.
outcome_for(self, faction: Faction) -> Outcome
Returns the outcome for a given faction.
Source code in core/ender.py
@inject_converters
def outcome_for(self, faction: Faction) -> Outcome:
"""Returns the outcome for a given faction."""
return self.outcomes[faction.name]
Pre
The game is about to end.
doit(self)
Ends the game, setting the phase to 'shutdown'.
Source code in core/ender.py
def doit(self):
"""Ends the game, setting the phase to 'shutdown'."""
self.game.change_phase(self.game.phase_system.shutdown)
GameEnder
Ends the game when all factions get an Outcome.
handle_outcome(event: EOutcomeAchieved) -> Optional[List[EndTheGame]]
open_mafia_engine.core.converters
get_ability_by_path(game: Game, obj: str) -> Ability
Gets the ability by 'path' made of names.
Assuming PATH_SEP is "/", this will parse as:
1 | |
This will do fuzzy matching on Actor and Ability separately. TODO: Maybe do fuzzy matching on total string?
Source code in core/converters.py
@converter.register
def get_ability_by_path(game: Game, obj: str) -> Ability:
"""Gets the ability by 'path' made of names.
Assuming PATH_SEP is "/", this will parse as:
"{actor_name}/ability/{ability_name}"
This will do fuzzy matching on Actor and Ability separately.
TODO: Maybe do fuzzy matching on total string?
"""
try:
owner_name, _abil, abil_name = get_parts(obj)
assert _abil == ABILITY
except Exception as e:
raise ValueError(f"Bad/non-existing path for Ability: {obj}") from e
owner: Actor = get_actor_by_name(game, owner_name)
matcher = FuzzyMatcher({ab.name: ab for ab in owner.abilities}, score_cutoff=10)
try:
return matcher[abil_name]
except Exception as e:
raise ValueError(f"Could not find Ability by path: {obj!r}") from e
get_actor_by_name(game: Game, obj: str) -> Actor
Gets the Actor by exact or fuzzy name match.
Source code in core/converters.py
@converter.register
def get_actor_by_name(game: Game, obj: str) -> Actor:
"""Gets the Actor by exact or fuzzy name match."""
matcher = FuzzyMatcher({a.name: a for a in game.actors}, score_cutoff=10)
try:
return matcher[obj]
except Exception as e:
raise ValueError(f"Could not find Actor by name: {obj!r}") from e
get_faction_by_name(game: Game, obj: str) -> Faction
Gets the Faction by exact or fuzzy name match.
Source code in core/converters.py
@converter.register
def get_faction_by_name(game: Game, obj: str) -> Faction:
"""Gets the Faction by exact or fuzzy name match."""
matcher = FuzzyMatcher({f.name: f for f in game.factions}, score_cutoff=20)
try:
return matcher[obj]
except Exception as e:
raise ValueError(f"Could not find Faction by name: {obj!r}") from e
get_phase_by_name(game: Game, obj: str) -> Phase
Gets the phase by name from the cycle. Can raise KeyError.
Source code in core/converters.py
@converter.register
def get_phase_by_name(game: Game, obj: str) -> Phase:
"""Gets the phase by name from the cycle. Can raise KeyError."""
try:
return game.phase_system[obj]
except KeyError:
pass
# Make sure we don't infinitely loop through phases
N_MAX = 20
options = {}
for i, p in enumerate(game.phase_system.possible_phases):
if i > N_MAX:
warnings.warn(f"Found over {N_MAX} phases; maybe infinite?")
break
options[p.name] = p
matcher = FuzzyMatcher(options, score_cutoff=50)
return matcher[obj]
get_trigger_by_path(game: Game, obj: str) -> Trigger
Gets the trigger by 'path' made of names.
Assuming PATH_SEP is "/", this will parse as:
1 | |
This will do fuzzy matching on Actor and Trigger separately. TODO: Maybe do fuzzy matching on total string?
Source code in core/converters.py
@converter.register
def get_trigger_by_path(game: Game, obj: str) -> Trigger:
"""Gets the trigger by 'path' made of names.
Assuming PATH_SEP is "/", this will parse as:
"{actor_name}/trigger/{trigger_name}"
This will do fuzzy matching on Actor and Trigger separately.
TODO: Maybe do fuzzy matching on total string?
"""
try:
owner_name, _trig, trig_name = get_parts(obj)
assert _trig == TRIGGER
except Exception as e:
raise ValueError(f"Bad/non-existing path for Trigger: {obj}") from e
owner: Actor = get_actor_by_name(game, owner_name)
matcher = FuzzyMatcher({tr.name: tr for tr in owner.triggers}, score_cutoff=10)
try:
return matcher[trig_name]
except Exception as e:
raise ValueError(f"Could not find Trigger by path: {obj!r}") from e
open_mafia_engine.core.enums
ActionResolutionType
How actions are resolved.
Outcome
Outcome (victory or defeat).