75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
from abc import abstractmethod, ABCMeta
|
|
class Unavailable(Exception): pass
|
|
class NotAllowed(Exception): pass
|
|
|
|
class Action(Exception, metaclass=ABCMeta):
|
|
def __init__(self, *args):
|
|
super().__init__(*args)
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def prepare(cls) -> bool:
|
|
pass
|
|
|
|
@classmethod
|
|
def perform(cls):
|
|
raise cls()
|
|
|
|
class ActionFlow:
|
|
def __init__(self):
|
|
self.indices: dict[str, int] = {}
|
|
self.actions: list[type[Action]] = []
|
|
self.on: list[bool] = []
|
|
|
|
def queue(self, action: type[Action]):
|
|
name = action.__name__
|
|
index = self.indices.get(name)
|
|
assert index is not None, "Action '%s' is not registered" % name
|
|
|
|
if self.on[index]: raise Unavailable(name)
|
|
if self.on[index] is None: raise NotAllowed(name)
|
|
self.on[index] = bool(self.actions[index].prepare())
|
|
|
|
def stage(self, *actions: type[Action]):
|
|
for Actor in actions:
|
|
self.allow(Actor)
|
|
self.queue(Actor)
|
|
self.react()
|
|
|
|
def react(self):
|
|
while any(self.on):
|
|
for index in self.indices.values():
|
|
if (self.on[index]):
|
|
self.on[index] = None
|
|
self.on[index] = bool(self.actions[index].perform())
|
|
|
|
def index(self, action: type[Action]) -> int:
|
|
name = action.__name__
|
|
index = self.indices.get(name)
|
|
assert index is not None, "Action '%s' is not registered" % name
|
|
return index
|
|
|
|
def allow(self, action: type[Action], value=True):
|
|
index = self.index(action)
|
|
self.on[index] = False if value else None
|
|
|
|
def append(self, action: type[Action]):
|
|
name = action.__name__
|
|
index = len(self.indices)
|
|
|
|
self.indices[name] = index
|
|
self.actions.insert(index, action)
|
|
self.on.insert(index, None)
|
|
|
|
def remove(self, action: type[Action]):
|
|
name = action.__name__
|
|
index = self.index(action)
|
|
|
|
self.indices.pop(name)
|
|
self.actions.pop(index)
|
|
self.on.pop(index)
|
|
|
|
def capabilities(self) -> dict[str, bool]:
|
|
items = self.indices.items()
|
|
return { k: not self.on[v] for k, v in items if self.on[v] is not None }
|