Add dynamic propery assignment

[tall-josh]

1. swag can now be used entirely in the backend. ie: not seen by the user when implementing stream_*_handlers
2. member variables for parameters are created at run time
3. Compatible with cli
4. addition of expected_parameters, expected_inputs and expected_outputs tuple for help with readability
5. Default vaues can be added to expected_parameters (see example) below:

ELEMENTS_SIMPLE = "aiko_services.elements.simple"

pipeline_definition = [
    {
        "name": "RandInt",
        "module": ELEMENTS_SIMPLE,
        "successors": ["MathList"],
        "parameters": {
            # "list_len": 5,   # Defaults are defined in definition
            # "iterations": 2, # un comment to override
            # "min": 30,
            # "max": 100,
        },
    },
    {
        "name": "MathList",
        "module": ELEMENTS_SIMPLE,
        "parameters": {
            "operation": "add",
        },
        "inputs": {"numbers": ["RandInt", "list"]},
        "successors": ["Print"],
    },
    {
        "name": "Print",
        "module": ELEMENTS_SIMPLE,
        "parameters": {
            "message_1": "The random ints are: ",
            "message_2": "The result is: ",
        },
        "inputs": {
            "to_print_1": ["RandInt", "list"],
            "to_print_2": ["MathList", "result"],
        },
    },
]

class RandInt(StreamElement):
    """Generates list of ints in 'range'. Result of length 'list_len' for
    'iterations' loops
    """

    expected_parameters = (
        ("list_len", 10),  #  if tuples the 1st value is the default parameter
        ("iterations", 10),
        ("min", 0),
        ("max", 10),
    )
    expected_inputs = ()
    expected_outputs = ("list",)

    def stream_frame_handler(self, stream_id, frame_id, inputs):
        self.logger.debug(
            f"stream_frame_handler(): stream_id: {stream_id}, frame_id: {frame_id}"
        )
        if frame_id == self.iterations:
            self.logger.info("Number of iterations completed. Exiting")
            return False, None

        result = {
            "list": list(np.random.randint(self.min, self.max, size=(self.list_len)))
        }
        return True, result

[tall-josh]

Another syntax could be:

class MyElement(StreamElement):
    """Using constructor, '*' tells python;  'everything after this MUST be a keyword arg'
    """
    def __init__(self, parameters, predecessors, pipeline_state_machine, *, foo: int, bar: str="bailey"):
        # foo is a required parameter
        self.foo: int = foo

        # bar is a default parameter
        self.bar: int = bar

        # Type information for inputs
        self.expected_inputs = Stream.Inputs(x: str, y: np.ndarray)

        # Type information for outputs
        self.expected_outputs = Stream.Outputs(z: float)

        super().__init__(MyElement, parameters, predecessors, pipeline_state_machine)

[tall-josh]

Thinking more I do prefer the override __init__ syntax:

1. More pythonic I think
2. Allows language servers to recognise member veriables for auto completion
3. Linters/fixers tend to complain when you define new member variables outside the constrctor

Cons:

1. A bit more text to type
2. parameters, predecessors, pipeline_state_machine are not really used, but this may not always be the case

---

• Also, I think this syntax would affort us to drop the parameters argument as all parameters are passed in already via keyword args
• Maybe we could move the predecessors and pipeline_state_machine agrs to another functions that is hidden from the interface.

[tall-josh]

See elements/simple.py for the most up-to-date thinking

Notes:

1. I could not get the constructor syntax to behave as I would have liked so I fell back to the original syntax but with Pydantic. Because Pydantic only defines type information I don't think the 2 of the same element in a pipeline problem will be an issue.

2. Ran suggested maybe we could use python inspect to interrogate the constructor at run time and build the Pydantic.BaseModels that way. We could possibly use the original constructor method then, but the length of the constructor will become a bit of a mess.

3. I have tried to make things backwards compatible, but with a log waning

4. I have added a **kwargs to each of the handlers to you can still access the swag if needed for your state machine work

5. We may need to add some handling around what None means in the context of a return value from the handlers. If None is returned due to a mistake Pydantice may not pick it up.

[tall-josh]

We could revise the syntax and use python inspect to infer the type information and build the pydantic BaseModels at run time