If the 'interrupt' parameter is True, when it ought to be False, the tickit device has a chance of triggering an exception that is not properly handled. This error will cause all future scheduled updates to fail to run.
I have tried to produce a minimum working example that raises the exception in question.
I have found the input / output field necessary for any tickit device; I am not sure why. It's called 'flux' as it was copied from an example.
Run the simulation:
python -m tickit all ./temp_controller.yaml
On a separate terminal, send the multiple requests:
./test_comms.py
Problematic code
```python
@RegexCommand(r"T\?", True, "utf-8") # Causes issues
@RegexCommand(r"T\?", False, "utf-8") # Works fine
```
Error messages
Benign messages:
```
DEBUG:tickit.core.management.schedulers.base:Scheduler (MasterScheduler) got Interrupt(source='tempcont')
DEBUG:tickit.core.management.schedulers.base:Scheduling tempcont for wakeup at 14296887047
```
Main error:
```
Task exception:
Traceback (most recent call last):
File "/venv/lib/python3.10/site-packages/tickit/core/runner.py", line 21, in run_with_error_handling
await awaitable
File "/venv/lib/python3.10/site-packages/tickit/core/management/schedulers/master.py", line 72, in run_forever
await self._do_tick()
File "/venv/lib/python3.10/site-packages/tickit/core/management/schedulers/master.py", line 88, in _do_tick
assert when is not None
AssertionError
```
device.py
```python
from dataclasses import dataclass
from tickit.adapters.composed import ComposedAdapter
from tickit.adapters.interpreters.command import CommandInterpreter
from tickit.adapters.interpreters.command.regex_command import RegexCommand
from tickit.adapters.servers.tcp import ByteFormat, TcpServer
from tickit.core.components.component import Component, ComponentConfig
from tickit.core.components.device_simulation import DeviceSimulation
from tickit.core.device import Device, DeviceUpdate
from tickit.core.typedefs import SimTime
from typing_extensions import TypedDict
class TempControllerDevice(Device):
Inputs: type = TypedDict("Inputs", {"flux": float})
Outputs: type = TypedDict("Outputs", {"flux": float})
def __init__(
self,
) -> None:
self._temp = 10.0
def get_temp(self):
return self._temp
def update(self, time: SimTime, inputs: Inputs) -> DeviceUpdate[Outputs]:
return DeviceUpdate(TempControllerDevice.Outputs(flux=inputs["flux"]), None)
class TempControllerAdapter(ComposedAdapter):
device: TempControllerDevice
def __init__(
self,
host: str = "localhost",
port: int = 25565,
) -> None:
super().__init__(
TcpServer(host, port, ByteFormat(b"%b\r\n")),
CommandInterpreter(),
)
@RegexCommand(r"T\?", True, "utf-8")
async def get_temperature(self) -> bytes:
return str(self.device.get_temp()).encode("utf-8")
@dataclass
class TempController(ComponentConfig):
host: str = "localhost"
port: int = 25565
def __call__(self) -> Component:
return DeviceSimulation(
name=self.name,
device=TempControllerDevice(),
adapters=[TempControllerAdapter(host=self.host, port=self.port)],
)
```
If the 'interrupt' parameter is True, when it ought to be False, the tickit device has a chance of triggering an exception that is not properly handled. This error will cause all future scheduled updates to fail to run.
I have tried to produce a minimum working example that raises the exception in question.
I have found the input / output field necessary for any tickit device; I am not sure why. It's called 'flux' as it was copied from an example.
Run the simulation:
python -m tickit all ./temp_controller.yamlOn a separate terminal, send the multiple requests:
./test_comms.pyProblematic code
```python @RegexCommand(r"T\?", True, "utf-8") # Causes issues @RegexCommand(r"T\?", False, "utf-8") # Works fine ```
Error messages
Benign messages: ``` DEBUG:tickit.core.management.schedulers.base:Scheduler (MasterScheduler) got Interrupt(source='tempcont') DEBUG:tickit.core.management.schedulers.base:Scheduling tempcont for wakeup at 14296887047 ``` Main error: ``` Task exception: Traceback (most recent call last): File "/venv/lib/python3.10/site-packages/tickit/core/runner.py", line 21, in run_with_error_handling await awaitable File "/venv/lib/python3.10/site-packages/tickit/core/management/schedulers/master.py", line 72, in run_forever await self._do_tick() File "/venv/lib/python3.10/site-packages/tickit/core/management/schedulers/master.py", line 88, in _do_tick assert when is not None AssertionError ```
device.py
```python from dataclasses import dataclass from tickit.adapters.composed import ComposedAdapter from tickit.adapters.interpreters.command import CommandInterpreter from tickit.adapters.interpreters.command.regex_command import RegexCommand from tickit.adapters.servers.tcp import ByteFormat, TcpServer from tickit.core.components.component import Component, ComponentConfig from tickit.core.components.device_simulation import DeviceSimulation from tickit.core.device import Device, DeviceUpdate from tickit.core.typedefs import SimTime from typing_extensions import TypedDict class TempControllerDevice(Device): Inputs: type = TypedDict("Inputs", {"flux": float}) Outputs: type = TypedDict("Outputs", {"flux": float}) def __init__( self, ) -> None: self._temp = 10.0 def get_temp(self): return self._temp def update(self, time: SimTime, inputs: Inputs) -> DeviceUpdate[Outputs]: return DeviceUpdate(TempControllerDevice.Outputs(flux=inputs["flux"]), None) class TempControllerAdapter(ComposedAdapter): device: TempControllerDevice def __init__( self, host: str = "localhost", port: int = 25565, ) -> None: super().__init__( TcpServer(host, port, ByteFormat(b"%b\r\n")), CommandInterpreter(), ) @RegexCommand(r"T\?", True, "utf-8") async def get_temperature(self) -> bytes: return str(self.device.get_temp()).encode("utf-8") @dataclass class TempController(ComponentConfig): host: str = "localhost" port: int = 25565 def __call__(self) -> Component: return DeviceSimulation( name=self.name, device=TempControllerDevice(), adapters=[TempControllerAdapter(host=self.host, port=self.port)], ) ```
temp_controller.yaml
``` - tickit.devices.source.Source: name: source inputs: {} value: 66.0 - device.TempController: name: tempcont inputs: flux: source:value - tickit.devices.sink.Sink: name: sink inputs: flux: tempcont:flux ```
test_comms.py
```python #! /bin/env python import socket HOSTNAME = "localhost" PORT = 25565 COUNT = 10 def send_message(s): message = b"T?\r\n" s.sendall(message) print(f"Sent: {message}") response = s.recv(1024) print(f"Received: {response}") def send_commands(): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((HOSTNAME, PORT)) for _ in range(COUNT): send_message(s) s.close() if __name__ == "__main__": send_commands() ```