Define Model Interfaces #22

Closed podhrmic closed 3 years ago

In GitLab by @EthanJamesLewon Aug 5, 2020, 14:03

Summary

We've defined configuration and interfaces of a CSAF component, but not the interfaces to the CSAF model. Currently, a simple interface exists for proof of concept, but more work is needed to refine it.

configure time invariant parameters
get representation specific info (fuzzy, neural network, linear, etc)
state update
output

In GitLab by @bauer-matthews on Aug 12, 2020, 11:03

Have the model API be cleaner to implement for a user:

Initialize with parameters
receive input
compute
send output

In GitLab by @bauer-matthews on Aug 12, 2020, 11:50

@EthanJamesLewLet's bump this up on the priority stack.

In GitLab by @EthanJamesLewon Aug 18, 2020, 14:58

Added model section in !14 CC @bauer-matthews

In GitLab by @bauer-matthews on Aug 19, 2020, 08:31

I'm having trouble reviewing the docs, can't generate the PDF. One other thing I'm feeling here is that the way we interact with models files (and define them), needs to be cleaned up a bit. Here are a few things:

I understand why ModelExecutable is the way it is. It allows for maximum compatibility with anything. But I would rather not have other model instantiations emulate its style. They should instead have a more rigid interface for interaction. And we should prefer instantiating these other model forms. I'll explain more in point 2 below.
I think the user (i.e. the person the defines a model) should be implementing a subclass of ModelNative directly. Right now we create one of these based on requirements on the interface of the main method in an arbitrary python program. I don't think its natural for a programmer to think this way. Furthermore, we aren't leveraging the compiler to enforce compliance with our interface.

Building on point 2, this would allow us to avoid some duplicate code. For example, I find the following in every model implementation. This should be defined once and for all in ModelNative.

if len(parameters.keys()) == 0:
    this_path = os.path.dirname(os.path.realpath(__file__))
    info_file = os.path.join(this_path, "ipcontroller.toml")
    with open(info_file, 'r') as ifp:
        info = toml.load(ifp)
    parameters = info["parameters"]

I'd separately also like to dive into the model interface design and see what the most natural way to interact with a scheduler is.

In GitLab by @EthanJamesLewon Aug 21, 2020, 15:22

@bauer-matthews For an abstract interface, here is a python class that I've been considering

""" Proposed Model Interface
"""

import collections.abc as cabc
import abc
import functools
import typing as typ
import keyword

def dynamical_input(func: typ.Callable):
    """asserts that the input passed to a callable matches the signature of a dynamical system, being a 4-tuple
    (m, t: float, x: typ.Sized, u: typ.Sized), being
    m: (Model) model with parameter attributes
    t: (float) current time
    x: (vector like) state vector
    u: (vector like) input vector
    """
    @functools.wraps(func)
    def check_input(*args, **kwargs):
        assert len[args] == 4
        assert isinstance(args[0], Model), f"argument 0 must be a model in dynamical function {func.__name__}"
        assert isinstance(args[1], float), f"argument 1 must be a time value in dynamical function {func.__name__}"
        assert isinstance(args[2], cabc.Sized), f"argument 2 must be sized in dynamical function {func.__name__}"
        assert isinstance(args[3], cabc.Sized), f"argument 3 must be sized in dynamical function {func.__name__}"
        assert kwargs == {}, f"no keyword arguments are permitted in dynamical function {func.__name__}"
        return func(*args, **kwargs)
    return check_input

class Model(abc.ABC):
    """CSAF Model Interface

    A CSAF Model encapsulates models seen in dynamical system theory. It is initialized with parameters, a
    representation, and whether the system time is continuous or discrete. Users implemented methods starting with an
    underscore. Their corresponding interface callables have additional input/output assertions for safety.
        1. If the state vector dimension is greater than zero, the method _get_state_update must be implemented,
        2. If the output message is non-empty, the method _get_output must be implemented
        3. _get_info is available for representation specific calls
        4. _update_model is available for updating entities that are NOT relevant to the dynamic model
        5. parameters are initialized by default but also settable
    """
    dynamic_callables: typ.Sequence[str] = ["get_output", "get_state_update", "get_info", "update_model"]

    def __init__(self, parameters, representation, is_discrete):
        self._parameters: typ.Mapping =  parameters
        self._representation: str = representation
        self._is_discrete: bool = is_discrete

    @dynamical_input
    def get_output(self, t: float, x: typ.Sized, u: typ.Sized) -> typ.Sized:
        """returns system output"""
        return self._get_output(t, x, u)

    @dynamical_input
    def get_state_update(self, t: float, x: typ.Sized, u: typ.Sized) -> typ.Sized:
        """returns system state update"""
        xp = self._get_state_update(t, x, u)
        assert xp == x, f"state update dimension must equal the dimension of the state (update is {len(xp)}, " \
                        f"but state is {len(x)})"
        return xp

    @dynamical_input
    def get_info(self, t: float, x: typ.Sized, u: typ.Sized) -> typ.Any:
        """returns representation specific information"""
        return self._get_info(t, x, u)

    @dynamical_input
    def update_model(self, t: float, x: typ.Sized, u: typ.Sized) -> None:
        """update attributes in the Model, but not the state x"""
        return self._update_model(t, x, u)

    @abc.abstractmethod
    def _get_output(self, t: float, x: typ.Sized, u: typ.Sized) -> typ.Sized:
        """user implemented"""
        raise NotImplementedError

    @abc.abstractmethod
    def _get_state_update(self, t: float, x: typ.Sized, u: typ.Sized) -> typ.Sized:
        """user implemented"""
        raise NotImplementedError

    @abc.abstractmethod
    def _get_info(self, t: float, x: typ.Sized, u: typ.Sized) -> typ.Any:
        """user implemented"""
        raise NotImplementedError

    @abc.abstractmethod
    def _update_info(self, t: float, x: typ.Sized, u: typ.Sized) -> None:
        """user implemented"""
        raise NotImplementedError

    @property
    def representation(self) -> str:
        return self._representation

    @property
    def is_discrete(self) -> bool:
        return self._is_discrete

    @property
    def is_continuous(self) -> bool:
        return not self._is_discrete

    @property
    def parameters(self) -> typ.Mapping:
        return self._parameters

    @parameters.setter
    def parameters(self, parameters):
        assert set(parameters.keys()) == set(self._parameters.keys())
        self._parameters = parameters

    def __getattr__(self, item) -> typ.Any:
        """allow parameters to be accessed at object level"""
        # error case - item is a keyword
        item = item[:-1] if (item[-1] == '_' and keyword.iskeyword(item[:-1])) else item
        if item in self.parameters:
            return self.parameters[item]
        else:
            raise AttributeError(f"{item} not found as a parameter")

In GitLab by @bauer-matthews on Aug 25, 2020, 09:55

@elew, per our discussion, this looks great. My only suggestion would be to go through the exercise of writing an example model from the user perspective.

In GitLab by @EthanJamesLewon Aug 25, 2020, 14:09

mentioned in merge request !23

In GitLab by @podhrmic on Aug 27, 2020, 16:50

closed

GaloisInc / csaf

Define Model Interfaces #22