From f6c063e263c60c94dde2f30969fa8613459c2b93 Mon Sep 17 00:00:00 2001 From: "Jesse P. Johnson" Date: Mon, 21 Apr 2025 21:14:35 +0200 Subject: [PATCH 1/3] refactor: remove callback cruft from transitions --- src/superstate/machine.py | 222 +++++++++++------------- src/superstate/state.py | 30 +--- src/superstate/transition.py | 79 +++++---- tests/state/test_change_notification.py | 13 +- tests/test_individuation.py | 6 +- tests/transition/test_transition.py | 11 -- 6 files changed, 158 insertions(+), 203 deletions(-) diff --git a/src/superstate/machine.py b/src/superstate/machine.py index 1d92b67..bd1ad89 100644 --- a/src/superstate/machine.py +++ b/src/superstate/machine.py @@ -7,13 +7,13 @@ import os from copy import deepcopy from itertools import zip_longest -from typing import ( +from typing import ( # Sequence, TYPE_CHECKING, Any, Dict, + Iterator, List, Optional, - # Sequence, Tuple, cast, ) @@ -29,12 +29,11 @@ ) from superstate.model.data import DataModel from superstate.provider import PROVIDERS -from superstate.state import ( +from superstate.state import ( # CompoundState, AtomicState, - SubstateMixin, - # CompoundState, ParallelState, State, + SubstateMixin, ) from superstate.types import Selection @@ -51,7 +50,7 @@ class MetaStateChart(type): __name__: str __initial__: Initial - __binding__: str = cast(str, Selection('early', 'late')) + __binding__: str = cast(str, Selection("early", "late")) __datamodel__: str _root: SubstateMixin datamodel: DataModel @@ -61,24 +60,24 @@ def __new__( name: str, bases: Tuple[type, ...], attrs: Dict[str, Any], - ) -> 'MetaStateChart': - if '__name__' not in attrs: + ) -> "MetaStateChart": + if "__name__" not in attrs: name = name.lower() - attrs['__name__'] = name + attrs["__name__"] = name else: - name = attrs.get('__name__', name.lower()) + name = attrs.get("__name__", name.lower()) - initial = attrs.get('__initial__', None) - root = State.create(attrs.pop('state')) if 'state' in attrs else None + initial = attrs.get("__initial__", None) + root = State.create(attrs.pop("state")) if "state" in attrs else None # setup datamodel - binding = attrs.get('__binding__', DEFAULT_BINDING) + binding = attrs.get("__binding__", DEFAULT_BINDING) if binding: DataModel.binding = binding - provider = attrs.get('__datamodel__', DEFAULT_PROVIDER) + provider = attrs.get("__datamodel__", DEFAULT_PROVIDER) if provider != DEFAULT_PROVIDER: DataModel.provider = PROVIDERS[provider] - datamodel = DataModel.create(attrs.pop('datamodel', {'data': []})) + datamodel = DataModel.create(attrs.pop("datamodel", {"data": []})) # XXX: chaining datamodels not working # datamodel['data'].append({'id': 'root', 'expr': root}) @@ -127,40 +126,41 @@ def __init__( # *args: Any, **kwargs: Any, ) -> None: - if 'logging_enabled' in kwargs and kwargs['logging_enabled']: - if 'logging_level' in kwargs: - log.setLevel(kwargs.pop('logging_level').upper()) - log.info('initializing statechart') + if "logging_enabled" in kwargs and kwargs["logging_enabled"]: + if "logging_level" in kwargs: + log.setLevel(kwargs.pop("logging_level").upper()) + log.info("initializing statechart") self._sessionid = UUID( - bytes=os.urandom(16), version=4 # pylint: disable=no-member + bytes=os.urandom(16), + version=4, # pylint: disable=no-member ) self.datamodel.populate() - if hasattr(self.__class__, '_root'): + if hasattr(self.__class__, "_root"): self.__root = deepcopy(self.__class__._root) self._root = None - elif 'superstate' in kwargs: - self.__root = kwargs.pop('superstate') + elif "superstate" in kwargs: + self.__root = kwargs.pop("superstate") else: - raise InvalidConfig('attempted initialization with empty parent') + raise InvalidConfig("attempted initialization with empty parent") self.__current_state = self.__root if not isinstance(self.__root, ParallelState): self.__initial__: Optional[str] = kwargs.get( - 'initial', self.__initial__ + "initial", self.__initial__ ) if self.initial: self.__current_state = self.get_state( self.initial # self.initial.transition.target ) - log.info('loaded states and transitions') + log.info("loaded states and transitions") # XXX: require composite state self.current_state.run_on_entry(self) - log.info('statechart initialization complete') + log.info("statechart initialization complete") # XXX: need to process after initial state # if isinstance(self.current_state, AtomicState): @@ -170,25 +170,16 @@ def __getattr__(self, name: str) -> Any: # do not attempt to resolve missing dunders if name.startswith('__'): raise AttributeError - # handle state check for active states if name.startswith('is_'): return name[3:] in self.active - - # handle automatic transitions - if name == '_auto_': - - def wrapper(*args: Any, **kwargs: Any) -> Optional[Any]: - return self.trigger('', *args, **kwargs) - - return wrapper raise AttributeError(f"cannot find attribute: {name}") @property def initial(self) -> Optional[str]: """Return initial state of current parent.""" if self.__initial__ is None: - if hasattr(self.root, 'initial'): + if hasattr(self.root, "initial"): self.__initial__ = self.root.initial elif self.root.states: self.__initial__ = list(self.root.states.values())[0].name @@ -202,6 +193,18 @@ def current_state(self) -> State: # TODO: rename to head or position potentially return self.__current_state + @current_state.setter + def current_state(self, state: State) -> None: + """Set the current state.""" + if hasattr(self.current_state, 'states'): + if state in list(self.current_state.states.values()): + self.__current_state = state + return + if len(self.active) > 1 and self.active[1] == state: + self.__current_state = state + else: + raise InvalidTransition('cannot transition from final state') + @property def root(self) -> SubstateMixin: """Return root state of statechart.""" @@ -217,7 +220,7 @@ def children(self) -> Tuple[State, ...]: """Return list of states.""" return ( tuple(self.__current_state.states.values()) - if hasattr(self.__current_state, 'states') + if hasattr(self.__current_state, "states") else () ) @@ -246,42 +249,42 @@ def active(self) -> Tuple[State, ...]: def get_relpath(self, target: str) -> str: """Get relative statepath of target state to current state.""" - if target in ('', self.current_state): # self reference - relpath = '.' + if target in ("", self.current_state): # self reference + relpath = "." else: - path = [''] - source_path = self.current_state.path.split('.') - target_path = self.get_state(target).path.split('.') + path = [""] + source_path = self.current_state.path.split(".") + target_path = self.get_state(target).path.split(".") for i, x in enumerate( - zip_longest(source_path, target_path, fillvalue='') + zip_longest(source_path, target_path, fillvalue="") ): if x[0] != x[1]: - if x[0] != '': # target is a descendent - path.extend(['' for x in source_path[i:]]) - if x[1] == '': # target is a ascendent - path.extend(['']) - if x[1] != '': # target is child of a ascendent + if x[0] != "": # target is a descendent + path.extend(["" for x in source_path[i:]]) + if x[1] == "": # target is a ascendent + path.extend([""]) + if x[1] != "": # target is child of a ascendent path.extend(target_path[i:]) if i == 0: raise InvalidPath( f"no relative path exists for: {target!s}" ) break - relpath = '.'.join(path) + relpath = ".".join(path) return relpath def change_state(self, statepath: str) -> None: """Traverse statepath.""" relpath = self.get_relpath(statepath) - if relpath == '.': # handle self transition + if relpath == ".": # handle self transition self.current_state.run_on_exit(self) self.current_state.run_on_entry(self) else: - s = 2 if relpath.endswith('.') else 1 # stupid black - macrostep = relpath.split('.')[s:] + s = 2 if relpath.endswith(".") else 1 # stupid black + macrostep = relpath.split(".")[s:] for microstep in macrostep: try: - if microstep == '': # reverse + if microstep == "": # reverse self.current_state.run_on_exit(self) self.__current_state = self.active[1] elif ( @@ -295,16 +298,16 @@ def change_state(self, statepath: str) -> None: raise InvalidPath(f"statepath not found: {statepath}") except Exception as err: log.error(err) - raise KeyError('parent is undefined') from err + raise KeyError("parent is undefined") from err # if type(self.current_state) not in [AtomicState, ParallelState]: # # TODO: need to transition from CompoundState to AtomicState # print('state transition not complete') - log.info('changed state to %s', statepath) + log.info("changed state to %s", statepath) def get_state(self, statepath: str) -> State: """Get state.""" state: State = self.root - macrostep = statepath.split('.') + macrostep = statepath.split(".") # general recursive search for single query if len(macrostep) == 1 and isinstance(state, SubstateMixin): @@ -312,15 +315,15 @@ def get_state(self, statepath: str) -> State: if x == macrostep[0]: return x # set start point for relative lookups - elif statepath.startswith('.'): - relative = len(statepath) - len(statepath.lstrip('.')) - 1 + elif statepath.startswith("."): + relative = len(statepath) - len(statepath.lstrip(".")) - 1 state = self.active[relative:][0] rel = relative + 1 macrostep = [state.name] + macrostep[rel:] # check relative lookup is done target = macrostep[-1] - if target in ('', state): + if target in ("", state): return state # path based search @@ -333,7 +336,7 @@ def get_state(self, statepath: str) -> State: if state == target: return state # walk path if exists - if hasattr(state, 'states') and microstep in state.states.keys(): + if hasattr(state, "states") and microstep in state.states.keys(): state = state.states[microstep] # check if target is found if not macrostep: @@ -342,27 +345,25 @@ def get_state(self, statepath: str) -> State: break raise InvalidState(f"state could not be found: {statepath}") - def add_state( - self, state: State, statepath: Optional[str] = None - ) -> None: + def add_state(self, state: State, statepath: Optional[str] = None) -> None: """Add state to either parent or target state.""" parent = self.get_state(statepath) if statepath else self.parent if isinstance(parent, SubstateMixin): parent.add_state(state) - log.info('added state %s', state.name) + log.info("added state %s", state.name) else: raise InvalidState( f"cannot add state to non-composite state {parent.name}" ) @property - def transitions(self) -> Tuple[Transition, ...]: + def transitions(self) -> Iterator[Transition]: """Return list of current transitions.""" - return ( - tuple(self.current_state.transitions) - if hasattr(self.current_state, 'transitions') - else () - ) + for state in self.active: + if hasattr(state, 'transitions'): + yield from state.transitions + else: + continue def add_transition( self, transition: Transition, statepath: Optional[str] = None @@ -371,61 +372,32 @@ def add_transition( target = self.get_state(statepath) if statepath else self.parent if isinstance(target, AtomicState): target.add_transition(transition) - log.info('added transition %s', transition.event) + log.info("added transition %s", transition.event) else: - raise InvalidState('cannot add transition to %s', target) + raise InvalidState("cannot add transition to %s", target) - @staticmethod - def _lookup_transitions(event: str, state: State) -> List[Transition]: - return ( - state.get_transition(event) - if hasattr(state, 'get_transition') - else [] - ) + def get_transitions(self, event: str) -> tuple[Transition, ...]: + """Get each transition maching event.""" + return tuple(filter(lambda t: t.event == event, self.transitions)) - def process_transitions( - self, event: str, /, *args: Any, **kwargs: Any - ) -> Transition: - """Get transition event from active states.""" - # TODO: must use datamodel to process transitions - # child => parent => grandparent - guarded: List[Transition] = [] - for current in self.active: - transitions: List[Transition] = [] - - # search parallel states for transitions - if isinstance(current, ParallelState): - for state in current.states.values(): - transitions += self._lookup_transitions(event, state) - else: - transitions = self._lookup_transitions(event, current) - - # evaluate conditions - allowed = [ - t for t in transitions if t.evaluate(self, *args, **kwargs) - ] - if allowed: - # if len(allowed) > 1: - # raise InvalidConfig( - # 'Conflicting transitions were allowed for event', - # event - # ) - return allowed[0] - guarded += transitions - if len(guarded) != 0: - raise ConditionNotSatisfied('no transition possible from state') - raise InvalidTransition(f"transition could not be found: {event}") - - def trigger( - self, event: str, /, *args: Any, **kwargs: Any - ) -> Optional[Any]: + def trigger(self, event: str, /, *args: Any, **kwargs: Any) -> None: """Transition from event to target state.""" - # NOTE: 'on' should register an event with an event loop for callback - # trigger - # XXX: currently does not allow contional transient states - transition = self.process_transitions(event, *args, **kwargs) - if transition: - log.info('transitioning to %r', event) - result = transition(self, *args, **kwargs) - return result - raise InvalidTransition('transition %r not found', event) + # TODO: need to consider superstate transitions. + if self.current_state.type == 'final': + raise InvalidTransition('cannot transition from final state') + + transitions = self.get_transitions(event) + if not transitions: + raise InvalidTransition('no transitions match event') + allowed = [t for t in transitions if t.evaluate(self, *args, **kwargs)] + if not allowed: + raise ConditionNotSatisfied( + 'Condition is not satisfied for this transition' + ) + if len(allowed) > 1: + raise InvalidTransition( + 'More than one transition was allowed for this event' + ) + log.info("processed guard for %s", allowed[0].event) + allowed[0].execute(self, *args, **kwargs) + log.info("processed transition event %s", allowed[0].event) diff --git a/src/superstate/state.py b/src/superstate/state.py index 1cf40b4..8417259 100644 --- a/src/superstate/state.py +++ b/src/superstate/state.py @@ -72,21 +72,6 @@ class TransitionMixin: __transitions: List[Transition] - def __register_transition_callback(self, transition: Transition) -> None: - # XXX: currently mapping to class instead of instance - setattr( - self, - transition.event if transition.event != '' else '_auto_', - # pylint: disable-next=unnecessary-dunder-call - transition.callback().__get__(self, self.__class__), - ) - - def _process_transient_state(self, ctx: StateChart) -> None: - for transition in self.transitions: - if transition.event == '': - ctx._auto_() # pylint: disable=protected-access - break - @property def transitions(self) -> Tuple[Transition, ...]: """Return transitions of this state.""" @@ -96,13 +81,10 @@ def transitions(self) -> Tuple[Transition, ...]: def transitions(self, transitions: List[Transition]) -> None: """Initialize atomic state.""" self.__transitions = transitions - for transition in self.transitions: - self.__register_transition_callback(transition) def add_transition(self, transition: Transition) -> None: """Add transition to this state.""" self.__transitions.append(transition) - self.__register_transition_callback(transition) def get_transition(self, event: str) -> Tuple[Transition, ...]: """Get each transition maching event.""" @@ -184,7 +166,7 @@ class State: name: str = cast(str, Identifier()) # history: Optional['HistoryState'] # final: Optional[FinalState] - states: Dict[str, State] + # states: Dict[str, State] # transitions: Tuple[Transition, ...] # onentry: Tuple[ActionTypes, ...] # onexit: Tuple[ActionTypes, ...] @@ -493,8 +475,14 @@ def run_on_entry(self, ctx: StateChart) -> Optional[Any]: self.datamodel, 'maps' ): self.datamodel.populate() - self._process_transient_state(ctx) - return super().run_on_entry(ctx) + log.info("executing 'on_entry' state change actions for %s", self.name) + results = super().run_on_entry(ctx) + # process transient states + for transition in self.transitions: + if transition.event == '': + ctx.trigger(transition.event) + break + return results class SubstateMixin(State): diff --git a/src/superstate/transition.py b/src/superstate/transition.py index 183fc18..d39f403 100644 --- a/src/superstate/transition.py +++ b/src/superstate/transition.py @@ -31,7 +31,7 @@ class Transition: name. In all cases, the token matching is case sensitive.] """ - # __slots__ = ['event', 'target', 'action', 'cond', 'type'] + # __slots__ = ['event', 'target', 'content', 'cond', 'type'] __source: Optional[State] = None event: str = cast(str, Identifier(TRANSITION_PATTERN)) @@ -57,32 +57,6 @@ def __init__( def __repr__(self) -> str: return repr(f"Transition(event={self.event}, target={self.target})") - def __call__( - self, ctx: StateChart, *args: Any, **kwargs: Any - ) -> Optional[Any]: - """Run transition process.""" - # TODO: move change_state to process_transitions - if 'statepath' in kwargs: - superstate_path = kwargs['statepath'].split('.')[:-1] - target = ( - '.'.join(superstate_path + [self.target]) - if superstate_path != [] - else self.target - ) - else: - target = self.target - - results = None - if self.content: - results = [] - provider = ctx.datamodel.provider(ctx) - for expression in tuplize(self.content): - results.append(provider.handle(expression, *args, **kwargs)) - log.info("executed action event for %r", self.event) - ctx.change_state(target) - log.info("no action event for %r", self.event) - return results - @classmethod def create(cls, settings: Union[Transition, dict]) -> Transition: """Create transition from configuration.""" @@ -119,16 +93,47 @@ def source(self, state: State) -> None: else: raise SuperstateException('cannot change source of transition') - def callback(self) -> Callable: - """Provide callback from source state when transition is called.""" - - def event(ctx: StateChart, *args: Any, **kwargs: Any) -> None: - """Provide callback event.""" - ctx.process_transitions(self.event, *args, **kwargs) - - event.__name__ = self.event - event.__doc__ = f"Transition event: '{self.event}'" - return event + def execute( + self, ctx: StateChart, *args: Any, **kwargs: Any + ) -> Optional[List[Any]]: + """Transition the state of the statechart.""" + log.info("executing transition contents for event %r", self.event) + results: Optional[List[Any]] = None + if self.content: + results = [] + provider = ctx.datamodel.provider(ctx) + for expression in tuplize(self.content): + results.append(provider.handle(expression, *args, **kwargs)) + log.info("completed transition contents for event %r", self.event) + relpath = ctx.get_relpath(self.target) + if relpath == '.': # handle self transition + ctx.current_state.run_on_exit(ctx) + ctx.current_state.run_on_entry(ctx) + else: + macrostep = relpath.split('.')[2 if relpath.endswith('.') else 1 :] + while macrostep[0] == '': # reverse + ctx.current_state.run_on_exit(ctx) + ctx.current_state = ctx.active[1] + macrostep.pop(0) + for microstep in macrostep: # forward + try: + if ( + # isinstance(ctx.current_state, State) + hasattr(ctx.current_state, 'states') + and microstep in ctx.current_state.states + ): + state = ctx.get_state(microstep) + ctx.current_state = state + state.run_on_entry(ctx) + else: + raise InvalidState( + f"statepath not found: {self.target}" + ) + except SuperstateException as err: + log.error(err) + raise KeyError('superstate is undefined') from err + log.info('changed state to %s', self.target) + return results def evaluate(self, ctx: StateChart, *args: Any, **kwargs: Any) -> bool: """Evaluate conditionss of transition.""" diff --git a/tests/state/test_change_notification.py b/tests/state/test_change_notification.py index 4731897..ee56759 100644 --- a/tests/state/test_change_notification.py +++ b/tests/state/test_change_notification.py @@ -8,11 +8,12 @@ class Door(StateChart): """Represent a door.""" - state = { - 'initial': 'closed', - 'states': [ + state = State( + name='door', + initial='closed', + states=[ State( - 'open', + name='open', transitions=[ Transition( event='close', @@ -39,7 +40,7 @@ class Door(StateChart): ], ), State( - 'closed', + name='closed', transitions=[ Transition( event='open', @@ -67,7 +68,7 @@ class Door(StateChart): ), State('broken'), ], - } + ) def __init__(self) -> None: super().__init__() diff --git a/tests/test_individuation.py b/tests/test_individuation.py index ee25cd1..42e1a65 100644 --- a/tests/test_individuation.py +++ b/tests/test_individuation.py @@ -25,9 +25,9 @@ class Door(StateChart): ) -def test_it_responds_to_an_event() -> None: - """Test door responds to an event.""" - assert hasattr(door.current_state, 'crack') +# def test_it_responds_to_an_event() -> None: +# """Test door responds to an event.""" +# assert hasattr(door.current_state, 'crack') def test_event_changes_state_when_called() -> None: diff --git a/tests/transition/test_transition.py b/tests/transition/test_transition.py index 3f114d2..629fbff 100644 --- a/tests/transition/test_transition.py +++ b/tests/transition/test_transition.py @@ -34,17 +34,6 @@ class MyMachine(StateChart): } -def test_its_declaration_creates_a_method_with_its_name(): - machine = MyMachine() - assert hasattr(machine.current_state, 'queue') and callable( - machine.current_state.queue - ) - assert hasattr(machine.current_state, 'cancel') and callable( - machine.current_state.cancel - ) - machine.trigger('queue') - - def test_it_changes_machine_state(): machine = MyMachine() assert machine.current_state == 'created' From 37a770dc822e51f8234bebe6a165b4511d530ddf Mon Sep 17 00:00:00 2001 From: "Jesse P. Johnson" Date: Mon, 21 Apr 2025 21:52:27 +0200 Subject: [PATCH 2/3] refactor: restore commits --- src/superstate/machine.py | 108 +++++++++++++++++++------------------- 1 file changed, 55 insertions(+), 53 deletions(-) diff --git a/src/superstate/machine.py b/src/superstate/machine.py index bd1ad89..18450db 100644 --- a/src/superstate/machine.py +++ b/src/superstate/machine.py @@ -7,13 +7,14 @@ import os from copy import deepcopy from itertools import zip_longest -from typing import ( # Sequence, +from typing import ( TYPE_CHECKING, Any, Dict, Iterator, List, Optional, + # Sequence, Tuple, cast, ) @@ -29,8 +30,9 @@ ) from superstate.model.data import DataModel from superstate.provider import PROVIDERS -from superstate.state import ( # CompoundState, +from superstate.state import ( AtomicState, + # CompoundState, ParallelState, State, SubstateMixin, @@ -50,7 +52,7 @@ class MetaStateChart(type): __name__: str __initial__: Initial - __binding__: str = cast(str, Selection("early", "late")) + __binding__: str = cast(str, Selection('early', 'late')) __datamodel__: str _root: SubstateMixin datamodel: DataModel @@ -60,24 +62,24 @@ def __new__( name: str, bases: Tuple[type, ...], attrs: Dict[str, Any], - ) -> "MetaStateChart": - if "__name__" not in attrs: + ) -> 'MetaStateChart': + if '__name__' not in attrs: name = name.lower() - attrs["__name__"] = name + attrs['__name__'] = name else: - name = attrs.get("__name__", name.lower()) + name = attrs.get('__name__', name.lower()) - initial = attrs.get("__initial__", None) - root = State.create(attrs.pop("state")) if "state" in attrs else None + initial = attrs.get('__initial__', None) + root = State.create(attrs.pop('state')) if 'state' in attrs else None # setup datamodel - binding = attrs.get("__binding__", DEFAULT_BINDING) + binding = attrs.get('__binding__', DEFAULT_BINDING) if binding: DataModel.binding = binding - provider = attrs.get("__datamodel__", DEFAULT_PROVIDER) + provider = attrs.get('__datamodel__', DEFAULT_PROVIDER) if provider != DEFAULT_PROVIDER: DataModel.provider = PROVIDERS[provider] - datamodel = DataModel.create(attrs.pop("datamodel", {"data": []})) + datamodel = DataModel.create(attrs.pop('datamodel', {'data': []})) # XXX: chaining datamodels not working # datamodel['data'].append({'id': 'root', 'expr': root}) @@ -126,10 +128,10 @@ def __init__( # *args: Any, **kwargs: Any, ) -> None: - if "logging_enabled" in kwargs and kwargs["logging_enabled"]: - if "logging_level" in kwargs: - log.setLevel(kwargs.pop("logging_level").upper()) - log.info("initializing statechart") + if 'logging_enabled' in kwargs and kwargs['logging_enabled']: + if 'logging_level' in kwargs: + log.setLevel(kwargs.pop('logging_level').upper()) + log.info('initializing statechart') self._sessionid = UUID( bytes=os.urandom(16), @@ -138,29 +140,29 @@ def __init__( self.datamodel.populate() - if hasattr(self.__class__, "_root"): + if hasattr(self.__class__, '_root'): self.__root = deepcopy(self.__class__._root) self._root = None - elif "superstate" in kwargs: - self.__root = kwargs.pop("superstate") + elif 'superstate' in kwargs: + self.__root = kwargs.pop('superstate') else: - raise InvalidConfig("attempted initialization with empty parent") + raise InvalidConfig('attempted initialization with empty parent') self.__current_state = self.__root if not isinstance(self.__root, ParallelState): self.__initial__: Optional[str] = kwargs.get( - "initial", self.__initial__ + 'initial', self.__initial__ ) if self.initial: self.__current_state = self.get_state( self.initial # self.initial.transition.target ) - log.info("loaded states and transitions") + log.info('loaded states and transitions') # XXX: require composite state self.current_state.run_on_entry(self) - log.info("statechart initialization complete") + log.info('statechart initialization complete') # XXX: need to process after initial state # if isinstance(self.current_state, AtomicState): @@ -179,7 +181,7 @@ def __getattr__(self, name: str) -> Any: def initial(self) -> Optional[str]: """Return initial state of current parent.""" if self.__initial__ is None: - if hasattr(self.root, "initial"): + if hasattr(self.root, 'initial'): self.__initial__ = self.root.initial elif self.root.states: self.__initial__ = list(self.root.states.values())[0].name @@ -220,7 +222,7 @@ def children(self) -> Tuple[State, ...]: """Return list of states.""" return ( tuple(self.__current_state.states.values()) - if hasattr(self.__current_state, "states") + if hasattr(self.__current_state, 'states') else () ) @@ -249,42 +251,42 @@ def active(self) -> Tuple[State, ...]: def get_relpath(self, target: str) -> str: """Get relative statepath of target state to current state.""" - if target in ("", self.current_state): # self reference - relpath = "." + if target in ('', self.current_state): # self reference + relpath = '.' else: - path = [""] - source_path = self.current_state.path.split(".") - target_path = self.get_state(target).path.split(".") + path = [''] + source_path = self.current_state.path.split('.') + target_path = self.get_state(target).path.split('.') for i, x in enumerate( - zip_longest(source_path, target_path, fillvalue="") + zip_longest(source_path, target_path, fillvalue='') ): if x[0] != x[1]: - if x[0] != "": # target is a descendent - path.extend(["" for x in source_path[i:]]) - if x[1] == "": # target is a ascendent - path.extend([""]) - if x[1] != "": # target is child of a ascendent + if x[0] != '': # target is a descendent + path.extend(['' for x in source_path[i:]]) + if x[1] == '': # target is a ascendent + path.extend(['']) + if x[1] != '': # target is child of a ascendent path.extend(target_path[i:]) if i == 0: raise InvalidPath( f"no relative path exists for: {target!s}" ) break - relpath = ".".join(path) + relpath = '.'.join(path) return relpath def change_state(self, statepath: str) -> None: """Traverse statepath.""" relpath = self.get_relpath(statepath) - if relpath == ".": # handle self transition + if relpath == '.': # handle self transition self.current_state.run_on_exit(self) self.current_state.run_on_entry(self) else: - s = 2 if relpath.endswith(".") else 1 # stupid black - macrostep = relpath.split(".")[s:] + s = 2 if relpath.endswith('.') else 1 # stupid black + macrostep = relpath.split('.')[s:] for microstep in macrostep: try: - if microstep == "": # reverse + if microstep == '': # reverse self.current_state.run_on_exit(self) self.__current_state = self.active[1] elif ( @@ -298,16 +300,16 @@ def change_state(self, statepath: str) -> None: raise InvalidPath(f"statepath not found: {statepath}") except Exception as err: log.error(err) - raise KeyError("parent is undefined") from err + raise KeyError('parent is undefined') from err # if type(self.current_state) not in [AtomicState, ParallelState]: # # TODO: need to transition from CompoundState to AtomicState # print('state transition not complete') - log.info("changed state to %s", statepath) + log.info('changed state to %s', statepath) def get_state(self, statepath: str) -> State: """Get state.""" state: State = self.root - macrostep = statepath.split(".") + macrostep = statepath.split('.') # general recursive search for single query if len(macrostep) == 1 and isinstance(state, SubstateMixin): @@ -315,15 +317,15 @@ def get_state(self, statepath: str) -> State: if x == macrostep[0]: return x # set start point for relative lookups - elif statepath.startswith("."): - relative = len(statepath) - len(statepath.lstrip(".")) - 1 + elif statepath.startswith('.'): + relative = len(statepath) - len(statepath.lstrip('.')) - 1 state = self.active[relative:][0] rel = relative + 1 macrostep = [state.name] + macrostep[rel:] # check relative lookup is done target = macrostep[-1] - if target in ("", state): + if target in ('', state): return state # path based search @@ -336,7 +338,7 @@ def get_state(self, statepath: str) -> State: if state == target: return state # walk path if exists - if hasattr(state, "states") and microstep in state.states.keys(): + if hasattr(state, 'states') and microstep in state.states.keys(): state = state.states[microstep] # check if target is found if not macrostep: @@ -350,7 +352,7 @@ def add_state(self, state: State, statepath: Optional[str] = None) -> None: parent = self.get_state(statepath) if statepath else self.parent if isinstance(parent, SubstateMixin): parent.add_state(state) - log.info("added state %s", state.name) + log.info('added state %s', state.name) else: raise InvalidState( f"cannot add state to non-composite state {parent.name}" @@ -372,9 +374,9 @@ def add_transition( target = self.get_state(statepath) if statepath else self.parent if isinstance(target, AtomicState): target.add_transition(transition) - log.info("added transition %s", transition.event) + log.info('added transition %s', transition.event) else: - raise InvalidState("cannot add transition to %s", target) + raise InvalidState('cannot add transition to %s', target) def get_transitions(self, event: str) -> tuple[Transition, ...]: """Get each transition maching event.""" @@ -398,6 +400,6 @@ def trigger(self, event: str, /, *args: Any, **kwargs: Any) -> None: raise InvalidTransition( 'More than one transition was allowed for this event' ) - log.info("processed guard for %s", allowed[0].event) + log.info('processed guard for %s', allowed[0].event) allowed[0].execute(self, *args, **kwargs) - log.info("processed transition event %s", allowed[0].event) + log.info('processed transition event %s', allowed[0].event) From b8774caf7b505ed2e3b21e62afeae9cd94f4b8fe Mon Sep 17 00:00:00 2001 From: "Jesse P. Johnson" Date: Mon, 21 Apr 2025 21:52:36 +0200 Subject: [PATCH 3/3] ci(version): apply 1.6.2a2 updates --- pyproject.toml | 2 +- src/superstate/__init__.py | 2 +- tests/test_version.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index fc64b82..3660460 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "superstate" -version = "1.6.2a1" +version = "1.6.2a2" description = "Robust statechart for configurable automation rules." readme = "README.md" license = {file = "LICENSE"} diff --git a/src/superstate/__init__.py b/src/superstate/__init__.py index 14d7b6e..a6b015c 100644 --- a/src/superstate/__init__.py +++ b/src/superstate/__init__.py @@ -64,7 +64,7 @@ __author_email__ = 'jpj6652@gmail.com' __title__ = 'superstate' __description__ = 'Compact statechart that can be vendored.' -__version__ = '1.6.2a1' +__version__ = '1.6.2a2' __license__ = 'MIT' __copyright__ = 'Copyright 2022 Jesse Johnson.' __all__ = ( diff --git a/tests/test_version.py b/tests/test_version.py index 76ec871..0ec3c0e 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -4,4 +4,4 @@ def test_version() -> None: """Test project metadata version.""" - assert __version__ == '1.6.2a1' + assert __version__ == '1.6.2a2'