3. API Reference

Typing

myosin.typing.AsyncCallback

Asynchronous method type

alias of Coroutine[Any, Any, None]

myosin.typing.PrimaryKey

myosin.models.state.StateModel id type alias

alias of Union[int, str]

State Context Manager

class myosin.state.state.GenericModel

generic myosin.models.state.StateModel type

alias of TypeVar(‘GenericModel’, bound=StateModel)

class myosin.state.state.State(*args: Type[StateModel])

Bases: object

State access context manager. Read, write or subscribe to registered myosin.models.state.StateModel objects. Request mutex locks on one or multiple state models by passing the model class in the initializer. If possible avoid nested context manager entry.

from myosin import State

with State(Model) as state:
    model = state.checkout(Model)
    ...

checkout(state_type: Type[GenericModel]) → GenericModel

Return a deepcopy of a registered user-defined state model.

Parameters:: state_type (Type[GenericModel]) – user-defined registered state model type
Raises:: ModelNotFound – if the requested state type does not exist
Returns:: deep copy of requested state model
Return type:: GenericModel

commit(state: StateModel, cache: bool = False) → None

Commit new state to system state and update state subscriber callbacks

Parameters:

state (StateModel) – modified copy of state
cache (bool, optional) – cache the state to disk once updated, defaults to False

Raises:

ModelNotFound – if system state has no state registered of the requested type

load(model: GenericModel) → GenericModel

Register myosin.models.state.StateModel into global system state registry. If a model of the same type is found in the system cache, overwrite default properties with that of the cached state.

Parameters:: model (GenericModel) – user-defined state model. Must implement myosin.models.state.StateModel.
Raises:: UninitializedStateError – if user-defined state model cannot be serialized
Returns:: model loaded into state registry
Return type:: GenericModel

reset() → None: Reset all loaded state models and clear cached documents

subscribe(state_type: Type[GenericModel], callback: Callable[[GenericModel], Coroutine[Any, Any, None]]) → None

Subscribe an asynchronous state change listener to a designated runtime model

Parameters:

state_type (Type[GenericModel]) – model type to subscribe to
callback (Callable[[GenericModel], AsyncCallback]) – state change listener callback

State Model

System state model base class.

class myosin.models.state.StateModel(_id: str | int | None = None)

Bases: ABC

System state model base class. Provides methods and properties required for use with the global state context manager. Define a custom model by implementing this class and providing overrides for the serialize() and deserialize() methods.

class Telemetry(StateModel):

    def __init__(self) -> None:
        super().__init__()
        self.temperature = 25.5
        self.timestamp = datetime.now().timestamp()

    @property
    def temperature(self) -> float:
        return self.__temperature

    @temperature.setter
    def temperature(self, value: float) -> None:
        self.__temperature = value

    @property
    def timestamp(self) -> float:
        return  self.__timestamp

    @timestamp.setter
    def timestamp(self, value: float) -> None:
        self.__timestamp = value

    def serialize(self) -> Dict[str, Any]:
        return {
            'id': self.id,
            'temperature': self.temperature,
            'timestamp': self.timestamp
        }

    def deserialize(self, **kwargs) -> None:
        for k, v in kwargs.items():
            setattr(self, k, v)

cache() → None

Serialize contents and save to cache as a json file

Raises:

NullCachePathError – if cache base path is not set
CachePathError – if cache base path is not valid

clear() → None: Remove the cached json file associated with this model

abstract deserialize(**kwargs) → None

Deserialize StateModel properties keys and values from kwargs generated by StateModel.serialize() output. This function must be overridden.

def deserialize(self, **kwargs) -> None:
    for k, v in kwargs.items():
        setattr(self, k, v)

Raises:: NotImplementedError – if method is not overriden

property id: int | str

Get state model id

Returns:: state model id
Return type:: PrimaryKey

load() → None: Load contents from json into StateModel using deserialize(). If document is not found, log a warning and continue. If the cached model fails to be read into the runtime context remove the document.

abstract serialize() → Dict[str, Any]

Serialize StateModel properties keys and values into a python dictionary. Key names should match StateModel property names. This function must be overridden.

def serialize(self) -> Dict[str, Any]:
    return {
        'id': self.id,
        'temperature': self.temperature,
        'timestamp': self.timestamp
    }

Raises:: NotImplementedError – if method is not overriden
Returns:: serialized model as a python dictionary
Return type:: Dict[str, Any]

Model Caching Exceptions

exception myosin.exceptions.cache.CacheException(msg: str = 'Cache encountered a general exception')

Bases: Exception

General model cache exception.

exception myosin.exceptions.cache.CachePathError(msg: str = 'Cache encountered a general exception')

Bases: CacheException

Raised on failure resolving caching path.

exception myosin.exceptions.cache.NullCachePathError(msg: str = 'Cache encountered a general exception')

Bases: CacheException

Raised on unset caching basepath. Basepath must be specified with the MYOSIN_CACHE_BASE_PATH environment variable.

State Exceptions

exception myosin.exceptions.state.ModelNotFound(msg: str = 'The requested model is not found')

Bases: StateException

Raised if the requested state model is not found in the internal state registry. Usually this means state actions are being requested on a model before it has been registered.

exception myosin.exceptions.state.StateException(msg: str = 'State encountered a general exception')

Bases: Exception

General state context exception.

exception myosin.exceptions.state.UninitializedStateError(msg: str = '')

Bases: StateException

Raised if a registration request is made on a model with one or more uninitialized properties. Verify the model can be serialized before it is registered with myosin.models.state.StateModel.serialize()