Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "superstate"
version = "1.6.2a0"
version = "1.6.2a1"
description = "Robust statechart for configurable automation rules."
readme = "README.md"
license = {file = "LICENSE"}
Expand Down
2 changes: 1 addition & 1 deletion src/superstate/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@
__author_email__ = 'jpj6652@gmail.com'
__title__ = 'superstate'
__description__ = 'Compact statechart that can be vendored.'
__version__ = '1.6.2a0'
__version__ = '1.6.2a1'
__license__ = 'MIT'
__copyright__ = 'Copyright 2022 Jesse Johnson.'
__all__ = (
Expand Down
10 changes: 6 additions & 4 deletions src/superstate/config/system.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
"""Provide system info for statechart components."""

from __future__ import annotations

from typing_extensions import NotRequired, TypedDict


Expand Down Expand Up @@ -37,7 +39,7 @@ class TimeInfo(TypedDict):
class SystemInfo(TypedDict):
"""Provide system info."""

host: NotRequired['HostInfo']
time: NotRequired['TimeInfo']
runtime: NotRequired['RuntimeInfo']
platform: NotRequired['PlatformInfo']
host: NotRequired[HostInfo]
time: NotRequired[TimeInfo]
runtime: NotRequired[RuntimeInfo]
platform: NotRequired[PlatformInfo]
22 changes: 12 additions & 10 deletions src/superstate/context.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
"""Provide context for distributing settings within statechart."""

from __future__ import annotations

import os
from typing import TYPE_CHECKING, List, Tuple
from uuid import UUID
Expand All @@ -18,7 +20,7 @@
class Context:
"""Provide context for statechart."""

def __init__(self, root: 'State') -> None:
def __init__(self, root: State) -> None:
"""Initialize a statechart from root state."""
self.__sessionid = UUID(
bytes=os.urandom(16), version=4 # pylint: disable=no-member
Expand All @@ -32,29 +34,29 @@ def _sessionid(self) -> str:
return str(self.__sessionid)

@property
def current_state(self) -> 'State':
def current_state(self) -> State:
"""Return the current state."""
# TODO: rename to "head" or "position"
return self.__current_state

@current_state.setter
def current_state(self, state: 'State') -> None:
def current_state(self, state: State) -> None:
"""Return the current state."""
# TODO: rename to "head" or "position"
self.__current_state = state

@property
def root(self) -> 'State':
def root(self) -> State:
"""Return root state of statechart."""
return self.__root

@property
def parent(self) -> 'State':
def parent(self) -> State:
"""Return parent."""
return self.current_state.parent or self.root

@property
def children(self) -> Tuple['State', ...]:
def children(self) -> Tuple[State, ...]:
"""Return list of states."""
return (
tuple(self.__current_state.states.values())
Expand All @@ -63,19 +65,19 @@ def children(self) -> Tuple['State', ...]:
)

@property
def states(self) -> Tuple['State', ...]:
def states(self) -> Tuple[State, ...]:
"""Return list of states."""
return tuple(self.parent.states.values())

@property
def siblings(self) -> Tuple['State', ...]:
def siblings(self) -> Tuple[State, ...]:
"""Return list of states."""
return tuple(self.parent.states.values())

@property
def active(self) -> Tuple['State', ...]:
def active(self) -> Tuple[State, ...]:
"""Return active states."""
states: List['State'] = []
states: List[State] = []
parents = list(reversed(self.current_state))
for i, x in enumerate(parents):
n = i + 1
Expand Down
48 changes: 25 additions & 23 deletions src/superstate/machine.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
"""Provide parent core statechart capability."""

from __future__ import annotations

import logging
import logging.config
import os
Expand Down Expand Up @@ -48,11 +50,11 @@ class MetaStateChart(type):
"""Instantiate statecharts from class metadata."""

__name__: str
__initial__: 'Initial'
__initial__: Initial
__binding__: str = cast(str, Selection('early', 'late'))
__datamodel__: str
_root: 'SubstateMixin'
datamodel: 'DataModel'
_root: SubstateMixin
datamodel: DataModel

def __new__(
mcs,
Expand Down Expand Up @@ -106,9 +108,9 @@ class StateChart(metaclass=MetaStateChart):
# __slots__ = [
# '__dict__', '__current_state', '__parent', '__root', 'initial'
# ]
__root: 'SubstateMixin'
__parent: 'SubstateMixin'
__current_state: 'State'
__root: SubstateMixin
__parent: SubstateMixin
__current_state: State

# # System Variables
# _name: str
Expand Down Expand Up @@ -195,23 +197,23 @@ def initial(self) -> Optional[str]:
return self.__initial__

@property
def current_state(self) -> 'State':
def current_state(self) -> State:
"""Return the current state."""
# TODO: rename to head or position potentially
return self.__current_state

@property
def root(self) -> 'SubstateMixin':
def root(self) -> SubstateMixin:
"""Return root state of statechart."""
return self.__root

@property
def parent(self) -> 'SubstateMixin':
def parent(self) -> SubstateMixin:
"""Return parent."""
return self.current_state.parent or self.root

@property
def children(self) -> Tuple['State', ...]:
def children(self) -> Tuple[State, ...]:
"""Return list of states."""
return (
tuple(self.__current_state.states.values())
Expand All @@ -220,19 +222,19 @@ def children(self) -> Tuple['State', ...]:
)

@property
def states(self) -> Tuple['State', ...]:
def states(self) -> Tuple[State, ...]:
"""Return list of states."""
return tuple(self.parent.states.values())

@property
def siblings(self) -> Tuple['State', ...]:
def siblings(self) -> Tuple[State, ...]:
"""Return list of states."""
return tuple(self.parent.states.values())

@property
def active(self) -> Tuple['State', ...]:
def active(self) -> Tuple[State, ...]:
"""Return active states."""
states: List['State'] = []
states: List[State] = []
parents = list(reversed(self.current_state))
for i, x in enumerate(parents):
n = i + 1
Expand Down Expand Up @@ -299,9 +301,9 @@ def change_state(self, statepath: str) -> None:
# print('state transition not complete')
log.info('changed state to %s', statepath)

def get_state(self, statepath: str) -> 'State':
def get_state(self, statepath: str) -> State:
"""Get state."""
state: 'State' = self.root
state: State = self.root
macrostep = statepath.split('.')

# general recursive search for single query
Expand Down Expand Up @@ -341,7 +343,7 @@ def get_state(self, statepath: str) -> 'State':
raise InvalidState(f"state could not be found: {statepath}")

def add_state(
self, state: 'State', statepath: Optional[str] = None
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
Expand All @@ -354,7 +356,7 @@ def add_state(
)

@property
def transitions(self) -> Tuple['Transition', ...]:
def transitions(self) -> Tuple[Transition, ...]:
"""Return list of current transitions."""
return (
tuple(self.current_state.transitions)
Expand All @@ -363,7 +365,7 @@ def transitions(self) -> Tuple['Transition', ...]:
)

def add_transition(
self, transition: 'Transition', statepath: Optional[str] = None
self, transition: Transition, statepath: Optional[str] = None
) -> None:
"""Add transition to either parent or target state."""
target = self.get_state(statepath) if statepath else self.parent
Expand All @@ -374,7 +376,7 @@ def add_transition(
raise InvalidState('cannot add transition to %s', target)

@staticmethod
def _lookup_transitions(event: str, state: 'State') -> List["Transition"]:
def _lookup_transitions(event: str, state: State) -> List[Transition]:
return (
state.get_transition(event)
if hasattr(state, 'get_transition')
Expand All @@ -383,13 +385,13 @@ def _lookup_transitions(event: str, state: 'State') -> List["Transition"]:

def process_transitions(
self, event: str, /, *args: Any, **kwargs: Any
) -> 'Transition':
) -> Transition:
"""Get transition event from active states."""
# TODO: must use datamodel to process transitions
# child => parent => grandparent
guarded: List['Transition'] = []
guarded: List[Transition] = []
for current in self.active:
transitions: List['Transition'] = []
transitions: List[Transition] = []

# search parallel states for transitions
if isinstance(current, ParallelState):
Expand Down
30 changes: 11 additions & 19 deletions src/superstate/model/action.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
"""Provide common types for statechart components."""

from __future__ import annotations

import logging
import logging.config
from dataclasses import InitVar, asdict, dataclass
Expand Down Expand Up @@ -33,9 +35,7 @@ class Assign(Action):
location: str
expr: Optional[Expression] = None # expression

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
) -> None:
def callback(self, provider: Provider, *args: Any, **kwargs: Any) -> None:
"""Provide callback from datamodel provider."""
kwargs['__mode__'] = 'single'
result = provider.exec(self.expr, *args, **kwargs)
Expand Down Expand Up @@ -65,9 +65,7 @@ class ForEach(Action):
def __post_init__(self, content: List[str]) -> None:
self.__content = [Action.create(x) for x in content] # type: ignore

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
) -> None:
def callback(self, provider: Provider, *args: Any, **kwargs: Any) -> None:
"""Provide callback from datamodel provider."""
array = provider.ctx.current_state.datamodel[self.array]
if array:
Expand Down Expand Up @@ -95,9 +93,7 @@ class Log(Action):
# def __post_init__(self) -> None:
# self.__log = logging.getLogger(self.label or provider.ctx.__name__)

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
) -> None:
def callback(self, provider: Provider, *args: Any, **kwargs: Any) -> None:
"""Provide callback from datamodel provider."""
kwargs['__mode__'] = 'single'
logger = logging.getLogger(self.label)
Expand All @@ -110,11 +106,9 @@ def callback(
class Raise(Action):
"""Data item providing state data."""

event: 'Event'
event: Event

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
) -> None:
def callback(self, provider: Provider, *args: Any, **kwargs: Any) -> None:
"""Provide callback from datamodel provider."""
kwargs['__mode__'] = 'single'

Expand All @@ -128,7 +122,7 @@ class Script(Action):
src: Union[Callable, str]

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
self, provider: Provider, *args: Any, **kwargs: Any
) -> Optional[Any]:
"""Provide callback from datamodel provider."""
# need ability to download src URI
Expand All @@ -140,13 +134,13 @@ def callback(
class If(Conditional):
"""Data item providing state data."""

content: Sequence['ExecutableContent']
content: Sequence[ExecutableContent]

def __post_init__(self) -> None:
self.content = [Action.create(x) for x in self.content]

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
self, provider: Provider, *args: Any, **kwargs: Any
) -> Optional[Any]:
"""Provide callback from datamodel provider."""
if provider.eval(self.cond, *args, **kwargs):
Expand All @@ -170,9 +164,7 @@ def __post_init__(self) -> None:
self.cond = True
self.content = [Action.create(x) for x in self.content] # type: ignore

def callback(
self, provider: 'Provider', *args: Any, **kwargs: Any
) -> None:
def callback(self, provider: Provider, *args: Any, **kwargs: Any) -> None:
"""Provide callback from datamodel provider."""
for action in self.content:
provider.handle(action, *args, **kwargs)
Loading