core

drunc.fsm.core

Classes

FSM(conf)

Source code in drunc/fsm/core.py

def __init__(self, conf):
    self.log = get_logger("controller.FSM")
    self.configuration = conf

    self.initial_state = self.configuration.get_initial_state()
    self.states = self.configuration.get_states()

    self.transitions = self.configuration.get_transitions()
    self.sequences = self.configuration.get_sequences()
    self._enusure_unique_transition(self.transitions)
    self.pre_transition_sequences = (
        self.configuration.get_pre_transitions_sequences()
    )
    self.post_transition_sequences = (
        self.configuration.get_post_transitions_sequences()
    )

    self.log.debug(f'Initial state is "{self.initial_state}"')
    self.log.debug("Allowed transitions are:")
    for t in self.transitions:
        self.log.debug(str(t))
        self.log.debug(f"Pre transition: {self.pre_transition_sequences[t]}")
        self.log.debug(f"Post transition: {self.post_transition_sequences[t]}")

Functions

can_execute_transition(source_state, transition)

Check that this transition is allowed given the source_state

Source code in drunc/fsm/core.py

def can_execute_transition(self, source_state, transition) -> bool:
    """Check that this transition is allowed given the source_state"""
    self.log.debug(f"can_execute_transition {transition.source!s} {source_state}")
    return regex_match(transition.source, source_state)

get_all_sequences()

Grab all the transitions

Source code in drunc/fsm/core.py

def get_all_sequences(self) -> list[FSMSequence]:
    """Grab all the transitions"""
    return self.sequences

get_all_states()

Grabs all the states

Source code in drunc/fsm/core.py

def get_all_states(self) -> list[str]:
    """Grabs all the states"""
    return self.states

get_all_transitions()

Grab all the transitions

Source code in drunc/fsm/core.py

def get_all_transitions(self) -> list[Transition]:
    """Grab all the transitions"""
    return self.transitions

get_destination_state(source_state, transition)

Tells us where a particular transition will take us, given the source_state

Source code in drunc/fsm/core.py

def get_destination_state(self, source_state, transition) -> str:
    """Tells us where a particular transition will take us, given the source_state"""
    right_name = [t for t in self.transitions if t == transition]
    for tr in right_name:
        if self.can_execute_transition(source_state, transition):
            if tr.destination == "":
                return source_state
            else:
                return tr.destination

FSMAction(name)

Abstract class defining a generic action

Source code in drunc/fsm/core.py

def __init__(self, name):
    self.name = name

PreOrPostTransitionSequence(transition, pre_or_post='pre')

Source code in drunc/fsm/core.py

def __init__(self, transition: Transition, pre_or_post="pre"):
    self.transition = transition
    if pre_or_post not in ["pre", "post"]:
        raise DruncSetupException(
            f"pre_or_post should be either 'pre' of 'post', you provided '{pre_or_post}'"
        )

    self.prefix = pre_or_post

    self.sequence = []
    self.log = get_logger("controller.PreOrPostTransitionSequence")

Functions

get_arguments()

Creates a list of arguments This is a bit sloppy, as really, I shouldn't be using protobuf here, and convert them later, but...

Source code in drunc/fsm/core.py

def get_arguments(self):
    """Creates a list of arguments
    This is a bit sloppy, as really, I shouldn't be using protobuf here, and convert them later, but...
    """
    retr = []
    all_the_parameter_names = []

    for callback in self.sequence:
        method = callback.method
        s = signature(method)

        for pname, p in s.parameters.items():
            if pname in ["_input_data", "_context", "args", "kwargs"]:
                continue

            if pname in all_the_parameter_names:
                raise fsme.DoubleArgument(
                    f"Parameter {pname} is already in the list of parameters"
                )
            all_the_parameter_names.append(p)

            default_value = ""

            t = Argument.Type.INT

            if p.annotation in (str, Optional[str], Union[str, None]):
                t = Argument.Type.STRING

                if p.default != Parameter.empty:
                    default_value = pack_to_any(string_msg(value=p.default))

            elif p.annotation in (float, Optional[float], Union[float, None]):
                t = Argument.Type.FLOAT

                if p.default != Parameter.empty:
                    default_value = pack_to_any(float_msg(value=p.default))

            elif p.annotation in (int, Optional[int], Union[int, None]):
                t = Argument.Type.INT

                if p.default != Parameter.empty:
                    default_value = pack_to_any(int_msg(value=p.default))

            elif p.annotation in (bool, Optional[bool], Union[bool, None]):
                t = Argument.Type.BOOL

                if p.default != Parameter.empty:
                    default_value = pack_to_any(bool_msg(value=p.default))

            else:
                raise fsme.UnhandledArgumentType(p.annotation)

            presence = Argument.Presence.MANDATORY
            if p.annotation in (
                Optional[str],
                Optional[float],
                Optional[int],
                Optional[bool],
                Union[str, None],
                Union[float, None],
                Union[int, None],
                Union[bool, None],
            ):
                presence = Argument.Presence.OPTIONAL
            if default_value:
                presence = Argument.Presence.OPTIONAL

            a = Argument(
                name=p.name,
                presence=presence,
                type=t,
                help="",
            )

            if default_value:
                a.default_value.CopyFrom(default_value)
            retr += [a]

    return retr

Functions