core

drunc.fsm.core

Classes

FSM(conf)

Source code in drunc/fsm/core.py

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

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

    self.transitions: list[Transition] = 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) -> FSMDestinationResult:
    """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 == "":
                # if no destination is provided by transition, assume it's source -> source
                return FSMDestinationResult(
                    destination_state=source_state,
                    destination_type=FSMDestinationType.DESTINATION_IS_SOURCE,
                )
            else:
                # found a transition that matches the source state provided, return the new destination
                return FSMDestinationResult(
                    destination_state=tr.destination,
                    destination_type=FSMDestinationType.VALID,
                )
        else:
            if tr.destination == source_state:
                # if the transition is not valid from the source state provided,
                # but its destination is the same as the source state, return that information
                return FSMDestinationResult(
                    destination_state=source_state,
                    destination_type=FSMDestinationType.DESTINATION_IS_SOURCE,
                )

    # no transitions match the source state provided
    # or the transition doesn't exist at all
    return FSMDestinationResult(
        destination_state=None,
        destination_type=FSMDestinationType.TRANSITION_NOT_VALID,
    )

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: str = "pre"):
    self.transition: 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: str = pre_or_post

    self.sequence: list(Callback) = []
    self.log = get_logger("controller.core.PreOrPostTransitionSequence")

Functions

add_callback(action, mandatory=True)

Add a callback to the sequence. The method to be called will be determined by the name of the transition and the prefix (pre or post).

For example, if the transition is "start" and the prefix is "pre", the method to be called will be "pre_start".

Parameters:

Name	Type	Description	Default
action	FSMAction	The action to be added to the sequence.	required
mandatory	bool	Whether the callback is mandatory or not.	True

Returns:

Type	Description
None	None

Raises:

Type	Description
DruncSetupException	If the method to be called is not found in the action.

Source code in drunc/fsm/core.py

def add_callback(
    self, action: "conffwk.dal.FSMaction", mandatory: bool = True
) -> None:
    """
    Add a callback to the sequence. The method to be called will be determined by
    the name of the transition and the prefix (pre or post).

    For example, if the transition is "start" and the prefix is "pre", the method to
    be called will be "pre_start".

    Args:
        action (conffwk.dal.FSMAction): The action to be added to the sequence.
        mandatory (bool): Whether the callback is mandatory or not.

    Returns:
        None

    Raises:
        DruncSetupException: If the method to be called is not found in the action.
    """

    # Get the method to be called from the action, based on the name of the
    # transition and the prefix (pre or post)
    method = getattr(action, f"{self.prefix}_{self.transition.name}")

    # Sanity check
    if not method:
        raise DruncSetupException(
            f"{self.prefix}_{self.transition.name} method not found in {action.name}"
        )

    # Add the callback to the sequence
    self.sequence += [
        Callback(
            method=method,
            mandatory=mandatory,
        )
    ]

get_arguments()

Create a list of arguments.

This is a bit sloppy, as really, I shouldn't be using protobuf here, and convert them later, but... Thanks Pierre :/

Returns:

Name	Type	Description
list	Argument	A list of arguments that the sequence requires.

Raises:

Type	Description
UnhandledArgumentType	If the type of an argument is not one of the following: str, float, int, bool, Optional[str], Optional[float], Optional[int], Optional[bool], Union[str, None], Union[float, None], Union[int, None], Union[bool, None]

Source code in drunc/fsm/core.py

def get_arguments(self) -> list[Argument]:
    """
    Create a list of arguments.

    This is a bit sloppy, as really, I shouldn't be using protobuf here, and convert them later, but...
    Thanks Pierre :/

    Args:
        None

    Returns:
        list(Argument): A list of arguments that the sequence requires.

    Raises:
        fsme.UnhandledArgumentType: If the type of an argument is not one of the
            following: str, float, int, bool, Optional[str], Optional[float],
            Optional[int], Optional[bool], Union[str, None], Union[float, None],
            Union[int, None], Union[bool, None]
    """

    # Construct the list of arguments by looking at the signature of the methods
    arguments: list(Argument) = []

    # Check that there are no duplicate parameter names across the callbacks
    # otherwise, we won't know which one to use when executing the sequence
    all_sequence_arguments: set(str) = set()  # set(Argument names)

    # Iterate over the callbacks, construct the list of arguments
    for callback in self.sequence:
        # Get the signature of the method to determine the arguments
        method = callback.method
        s = signature(method)

        # Iterate over the parameters of the method
        for pname, p in s.parameters.items():
            # Skip the special parameters that are used to pass the input data and
            # context to the callbacks
            if pname in ["_input_data", "_context", "args", "kwargs"]:
                continue

            # Check that the parameter name is not already in the list of arguments
            if pname in all_sequence_arguments:
                raise fsme.DoubleArgument(
                    f"Parameter {pname} is already in the list of parameters"
                )

            # Keep track of the parameter names to check for duplicates
            all_sequence_arguments.add(pname)

            # Set the default value to an empty string, as protobuf doesn't allow
            # using default None
            default_value: any_pb2.Any | None = None

            # Determine the type of the argument, and set the default value if it is
            # optional. If the type is not one of the supported types, raise an
            # error
            t: Argument.Type = 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 default_value or 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

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

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

    return arguments

Functions