Creating an Adapter

This tutorial shows how to create a simple Adapter for use in the ticket framework. This adapter will act as a simple TCP interface to the Shutter device which can vary the transmission of flux by changing position. The Adapter we create will be composable, meaning it will comprise of a pre-made Server and Interpreter.

Adapter Module File

We shall begin by creating a new python module named my_shutter_adapter.py, and open it with our preferred editor. This file will be used to store the our ShutterAdapter class which will determine the operation of our adapter.

Adapter Class

We shall begin by defining the ShutterAdapter class which inherits Adapter - by doing so a configuration dataclass will automatically be created for the adapter, allowing easy YAML configuration.

from tickit.core.adapter import Adapter

class ShutterAdapter(Adapter):

Adapter Constructor and Configuration

Next, we shall create the __init__ method, allowing for the adapter to be instantiated. The first two arguments of an adapter __init__ must be device of type Device and raise_interrupt of type Callable[[], Awaitable[None]], to which the device the adapter acts upon and a method used to request an immediate update of the device are passed respectively. Additionally, we will include the arguments host and port which will be used to configure our TcpServer. With these we will instantiate an adapter with a TcpServer which formats sent messages by appending a line break and a CommandInterpreter.

from typing import Awaitable, Callable

from tickit.adapters.interpreters.command.command_interpreter import CommandInterpreter
from tickit.adapters.servers.tcp import TcpServer
from tickit.core.adapter import Adapter

class ShutterAdapter(Adapter):
    def __init__(
        self,
        device: Shutter,
        raise_interrupt: Callable[[], Awaitable[None]],
        host: str = "localhost",
        port: int = 25565,
    ) -> None:
        super().__init__(
            device,
            raise_interrupt,
            TcpServer(host, port, ByteFormat(b"%b\r\n")),
            CommandInterpreter(),
        )

Note

Arguments to the __init__ method may be specified in the simulation config file if the device inherits Adapter (excluding device and raise_interrupt which are injected at run-time).

Adapter Commands

When using the CommandInterpreter, commands may be registered by decorating an adapter method with a command register. We shall begin by creating a method which returns the current position of the Shutter and register it as a RegexCommand which is called by sending P?. Since reading will not alter the device state we shall set interrupt to False. We shall also specify that the byte encoded message should be decoded to a string prior to matching using the utf-8 standard.

from tickit.adapters.interpreters.command.regex_command import RegexCommand
from tickit.core.adapter import Adapter

class ShutterAdapter(Adapter):

    ...

    @RegexCommand(r"P\?", False, "utf-8")
    async def get_position(self) -> bytes:
        return str(self._device.position).encode("utf-8")

We shall add a similar command to read back the current target position of the Shutter when T? is recieved.

from tickit.adapters.interpreters.command.regex_command import RegexCommand
from tickit.core.adapter import Adapter

class ShutterAdapter(Adapter):

    ...

    @RegexCommand(r"T\?", False, "utf-8")
    async def get_target(self) -> bytes:
        return str(self._device.target_position).encode("utf-8")

Next, we shall add a method which sets a new target position for the Shutter when T=(\d+\.?\d*) is recieved, where \d+\.?\d* denotes a decimal number and the parentheses form the capture group from which the argument is extracted.

from tickit.adapters.interpreters.command.regex_command import RegexCommand
from tickit.core.adapter import Adapter

class ShutterAdapter(Adapter):

    ...

    @RegexCommand(r"T=(\d+\.?\d*)", True, "utf-8")
    async def set_target(self, target: str) -> None:
        self._device.target_position = float(target)
        self._device.last_time = None

Using the Adapter

As explained in Creating a Device, an adapter allows us to control a device externally. And in order to use the device we created a ComponentConfig for said device. This is defined in the same file as the device and adapter, and defines any default initial configuration values. As well as this, we override the magic method __call__(), which return a DeviceSimulation object. In the DeviceSimulation object, we also define the device’s adapter(s):

@dataclass
class Shutter(ComponentConfig):

    default_position: float
    initial_position: Optional[float] = None
    host: str = "localhost"
    port: int = 25565

    def __call__(self) -> Component:
        return DeviceSimulation(
            name=self.name,
            device=ShutterDevice(
                default_position=self.default_position,
                initial_position=self.initial_position,
            ),
            adapters=[ShutterAdapter(host=self.host, port=self.port)],
        )

Using ComponentConfig the adapters are instantiated in Python so they are not referenced in the YAML config.

See also

See the Creating a Simulation tutorial for a walk-through of creating simulation configurations.

We may then run the simulation, this may be performed by running the following command:

python -m tickit all my_shutter_simulation.yaml

Additionally, we will start a telnet client which communicates with the TcpServer of the adapter, this may be performed by running the following command:

telnet localhost 25565

When run we expect a response akin to:

Trying ::1...
Connected to localhost.
Escape character is \'^]\'.

From this telnet client we can send various messages and recieve responses from our adapter. Examples of getting the position and target with P? and T? are shown below:

P?
0.2

T?
0.2

Finally, we may wish to set a new target with T=, an example of this with the value 0.16 is shown below, with accompanying tickit debug output:

T=0.16

DEBUG:tickit.adapters.servers.tcp:Recieved b'T=0.16\r\n' from ('::1', 33096, 0, 0)
DEBUG:tickit.core.management.schedulers.base:Scheduler got Interrupt(source='shutter')
DEBUG:tickit.core.management.schedulers.base:Scheduling Wakeup(component='shutter', when=209786950024)
DEBUG:tickit.core.management.ticker:Doing tick @ 209786950024
DEBUG:tickit.core.components.component:shutter got Input(target='shutter', time=209786950024, changes=immutables.Map({}))
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='shutter', time=209786950024, changes=immutables.Map({}), call_in=100000000)
DEBUG:tickit.core.management.schedulers.base:Scheduling Wakeup(component='shutter', when=209886950024)
DEBUG:tickit.core.components.component:sink got Input(target='sink', time=209786950024, changes=immutables.Map({}))
DEBUG:tickit.devices.sink:Sunk {'flux': 8.4}
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='sink', time=209786950024, changes=immutables.Map({}), call_in=None)
DEBUG:tickit.core.management.ticker:Doing tick @ 209886950024
DEBUG:tickit.core.components.component:shutter got Input(target='shutter', time=209886950024, changes=immutables.Map({}))
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='shutter', time=209886950024, changes=immutables.Map({'flux': 7.5600000000000005}), call_in=100000000)
DEBUG:tickit.core.management.schedulers.base:Scheduling Wakeup(component='shutter', when=209986950024)
DEBUG:tickit.core.components.component:sink got Input(target='sink', time=209886950024, changes=immutables.Map({'flux': 7.5600000000000005}))
DEBUG:tickit.devices.sink:Sunk {'flux': 7.5600000000000005}
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='sink', time=209886950024, changes=immutables.Map({}), call_in=None)
DEBUG:tickit.core.management.ticker:Doing tick @ 209986950024
DEBUG:tickit.core.components.component:shutter got Input(target='shutter', time=209986950024, changes=immutables.Map({}))
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='shutter', time=209986950024, changes=immutables.Map({'flux': 6.7200000000000015}), call_in=100000000)
DEBUG:tickit.core.management.schedulers.base:Scheduling Wakeup(component='shutter', when=210086950024)
DEBUG:tickit.core.components.component:sink got Input(target='sink', time=209986950024, changes=immutables.Map({'flux': 6.7200000000000015}))
DEBUG:tickit.devices.sink:Sunk {'flux': 6.7200000000000015}
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='sink', time=209986950024, changes=immutables.Map({}), call_in=None)
DEBUG:tickit.core.management.ticker:Doing tick @ 210086950024
DEBUG:tickit.core.components.component:shutter got Input(target='shutter', time=210086950024, changes=immutables.Map({}))
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='shutter', time=210086950024, changes=immutables.Map({'flux': 6.72}), call_in=None)
DEBUG:tickit.core.components.component:sink got Input(target='sink', time=210086950024, changes=immutables.Map({'flux': 6.72}))
DEBUG:tickit.devices.sink:Sunk {'flux': 6.72}
DEBUG:tickit.core.management.schedulers.base:Scheduler got Output(source='sink', time=210086950024, changes=immutables.Map({}), call_in=None)