Require a context for emulator constructors

src/comet/emulator/__init__.py

@@ -1,4 +1,4 @@
-from .emulator import Emulator, message
+from .emulator import Context, Emulator, message
 from .iec60488 import IEC60488Emulator
 from .resource import open_emulator
 from .response import BinaryResponse, RawResponse, TextResponse
@@ -7,6 +7,7 @@
 __all__ = [
     "BinaryResponse",
     "Emulator",
+    "Context",
     "IEC60488Emulator",
     "RawResponse",
     "TextResponse",

src/comet/emulator/__main__.py

@@ -35,7 +35,7 @@
 import yaml
 
 from .. import __version__
-from .emulator import emulator_factory
+from .emulator import Context, emulator_cls_factory
 from .tcpserver import TCPServer, TCPServerContext
 
 logger = logging.getLogger(__name__)
@@ -154,17 +154,18 @@ async def async_main() -> None:
         request_delay = params.get("request_delay")
         options = params.get("options", {})
 
-        emulator = emulator_factory(model)()
-        emulator.load_options(options)
+        context = Context(options=options)
+        cls = emulator_cls_factory(model)
+        emulator = cls(context)
 
-        context = TCPServerContext(
+        server_context = TCPServerContext(
             name=name,
             emulator=emulator,
             termination=termination_bytes,
             request_delay=request_delay,
             logger=logging.getLogger(name),
         )
-        server = TCPServer((host, port), context)
+        server = TCPServer((host, port), server_context)
         await server.start()
         servers.append(server)
 

src/comet/emulator/cts/itc.py

@@ -3,7 +3,7 @@
 import random
 from datetime import UTC, datetime
 
-from comet.emulator import Emulator, message, run
+from comet.emulator import Context, Emulator, message, run
 
 __all__ = ["ITCEmulator"]
 
@@ -18,14 +18,16 @@ def fake_analog_channel(channel, minimum, maximum):
 class ITCEmulator(Emulator):
     IDENTITY: str = "ITS Climate Chamber, v1.0 (Emulator)"
 
-    def __init__(self) -> None:
-        super().__init__()
+    def __init__(self, context: Context) -> None:
+        super().__init__(context)
 
-        self.current_temp: float = 24.0
-        self.target_temp: float = 24.0
+        options = context.options
 
-        self.current_humid: float = 55.0
-        self.target_humid: float = 55.0
+        self.current_temp: float = float(options.get("current_temp", 24.0))
+        self.target_temp: float = float(options.get("target_temp", 24.0))
+
+        self.current_humid: float = float(options.get("current_humid", 55.0))
+        self.target_humid: float = float(options.get("target_humid", 55.0))
 
         self.program: int = 0
 
@@ -121,4 +123,4 @@ def set_p(self, program) -> str:
 
 
 if __name__ == "__main__":
-    run(ITCEmulator())
+    run(ITCEmulator)

src/comet/emulator/emulator.py

@@ -4,20 +4,21 @@
 import inspect
 import logging
 import re
-from collections.abc import Callable, Mapping
+from collections.abc import Callable
+from dataclasses import dataclass, field
 from typing import Any
 
 from ..utils import parse_model_urn
 from .response import Response, make_response
 
-__all__ = ["Emulator", "emulator_factory", "message"]
+__all__ = ["Emulator", "Context", "emulator_cls_factory", "message"]
 
 logger = logging.getLogger(__name__)
 
 emulator_registry: dict[str, type[Emulator]] = {}
 
 
-def emulator_factory(model_urn: str) -> type[Emulator]:
+def emulator_cls_factory(model_urn: str) -> type[Emulator]:
     """Returns emulator class from model specified by URN."""
     module_name: str = parse_model_urn(model_urn)
     key: str = module_name
@@ -99,12 +100,14 @@ def decorator(method: Callable[..., Any]) -> Route:
     return decorator
 
 
-class Emulator:
-    def __init__(self) -> None:
-        self.options: dict[str, Any] = {}
+@dataclass
+class Context:
+    options: dict[str, Any] = field(default_factory=dict)
+
 
-    def load_options(self, options: Mapping[str, Any]) -> None:
-        self.options.update(options)
+class Emulator:
+    def __init__(self, context: Context) -> None:
+        self.context = context
 
     def __call__(self, message: str) -> Response | list[Response] | None:
         logger.debug("handle message: %s", message)

src/comet/emulator/ers/ac3.py

@@ -1,10 +1,12 @@
 """Driver for ECR AC3 thermal chuck"""
 
 import time
+from collections.abc import Mapping
 from dataclasses import dataclass
 from enum import IntEnum
+from typing import Any, Self
 
-from comet.emulator import Emulator, message, run
+from comet.emulator import Context, Emulator, message, run
 
 __all__ = ["AC3Emulator"]
 
@@ -23,7 +25,7 @@ class Status(IntEnum):
     ERROR = 8
 
 
-@dataclass
+@dataclass(slots=True)
 class State:
     temperature: float = 25.0
     target_temperature: float = 25.0
@@ -33,6 +35,17 @@ class State:
     dewpoint_control_status: bool = True
     dewpoint: float = -20.0
 
+    @classmethod
+    def from_mapping(cls, options: Mapping[str, Any]) -> Self:
+        state = cls()
+        if "temperature" in options:
+            state.temperature = float(options["temperature"])
+        if "target_temperature" in options:
+            state.target_temperature = float(options["target_temperature"])
+        if "dewpoint" in options:
+            state.dewpoint = float(options["dewpoint"])
+        return state
+
 
 class Logic:
     def __init__(self, state: State) -> None:
@@ -78,9 +91,10 @@ def update_state(self) -> None:
 
 
 class AC3Emulator(Emulator):
-    def __init__(self) -> None:
-        super().__init__()
-        self.state = State()
+    def __init__(self, context: Context) -> None:
+        super().__init__(context)
+
+        self.state = State.from_mapping(context.options)
         self.logic = Logic(self.state)
 
     @message(r"RC$")
@@ -143,4 +157,4 @@ def get_error(self) -> str:
 
 
 if __name__ == "__main__":
-    run(AC3Emulator())
+    run(AC3Emulator)

src/comet/emulator/hephy/brandbox.py

@@ -1,4 +1,4 @@
-from comet.emulator import Emulator, message, run
+from comet.emulator import Context, Emulator, message, run
 
 __all__ = ["BrandBoxEmulator"]
 
@@ -23,12 +23,15 @@ class BrandBoxEmulator(Emulator):
     CHANNELS: tuple[str, ...] = ("A1", "A2", "B1", "B2", "C1", "C2")
     MODS: tuple[str, ...] = ("IV", "CV")
 
-    IDENTITY: str = "BrandBox, v2.0 (Emulator)"
     SUCCESS: str = "OK"
     COMMAND_ERROR: str = format_error(99)
 
-    def __init__(self) -> None:
-        super().__init__()
+    def __init__(self, context: Context) -> None:
+        super().__init__(context)
+
+        options = context.options
+
+        self._identity = options.get("identity", "BrandBox, v2.0 (Emulator)")
         self.closed_channels: set[str] = set()
         self.test_state: bool = False
         self.mod: str = "N/A"
@@ -39,7 +42,7 @@ def opened_channels(self) -> set[str]:
 
     @message(r"\*IDN\?$")
     def get_idn(self) -> str:
-        return self.options.get("identity", self.IDENTITY)
+        return self._identity
 
     @message(r"\*RST$")
     def set_rst(self) -> str:
@@ -188,4 +191,4 @@ def has_channel(self, channel) -> bool:
 
 
 if __name__ == "__main__":
-    run(BrandBoxEmulator())
+    run(BrandBoxEmulator)

src/comet/emulator/hephy/environbox.py

@@ -1,7 +1,7 @@
 import random
 import time
 
-from comet.emulator import Emulator, message, run
+from comet.emulator import Context, Emulator, message, run
 from comet.utils import t_dew
 
 __all__ = ["EnvironBoxEmulator"]
@@ -19,14 +19,18 @@ def split_seconds(delta_seconds: float) -> tuple[int, int, int, int]:
 
 
 class EnvironBoxEmulator(Emulator):
-    IDENTITY: str = "EnvironBox, v2.0 (Emulator)"
-    VERSION: str = "V2.0"
     SUCCESS: str = "OK"
     PC_DATA_SIZE: int = 39
     SENSOR_ADRESSES: tuple[int, ...] = (40, 41, 42, 43, 44, 45)
 
-    def __init__(self) -> None:
-        super().__init__()
+    def __init__(self, context: Context) -> None:
+        super().__init__(context)
+
+        options = context.options
+
+        self._identity = options.get("identity", "EnvironBox, v2.0 (Emulator)")
+        self._version = options.get("version", "V2.0")
+
         self.boot_timestamp: float = time.time()
 
         self.sensor_address: list[int] = [40, 41, 42]
@@ -81,6 +85,15 @@ def __init__(self) -> None:
         self.pid_door_stop: bool = False
         self.door_auto_light: bool = False
 
+        self._temp_min = float(options.get("box_temperature.min", 23.5))
+        self._temp_max = float(options.get("box_temperature.max", 24.5))
+        self._humid_min = float(options.get("box_humidity.min", 39.5))
+        self._humid_max = float(options.get("box_humidity.max", 40.5))
+        self._pt100_1_min = float(options.get("pt100_1.min", 21.0))
+        self._pt100_1_max = float(options.get("pt100_1.max", 21.5))
+        self._pt100_2_min = float(options.get("pt100_2.min", 22.0))
+        self._pt100_2_max = float(options.get("pt100_2.max", 22.5))
+
     @property
     def pid_control_mode_index(self) -> int:
         return {"HUM": 1, "DEW": 2}[self.pid_control_mode]
@@ -91,15 +104,11 @@ def pid_prop_mode_index(self) -> int:
 
     @property
     def box_temperature(self) -> float:
-        minimum = float(self.options.get("box_temperature.min", 24.0))
-        maximum = float(self.options.get("box_temperature.max", 24.5))
-        return round(random.uniform(minimum, maximum), 1)
+        return round(random.uniform(self._temp_min, self._temp_max), 1)
 
     @property
     def box_humidity(self) -> float:
-        minimum = float(self.options.get("box_humidity.min", 40.0))
-        maximum = float(self.options.get("box_humidity.max", 40.5))
-        return round(random.uniform(minimum, maximum), 1)
+        return round(random.uniform(self._humid_min, self._humid_max), 1)
 
     @property
     def box_dewpoint(self) -> float:
@@ -116,17 +125,13 @@ def pid_setpoint(self, value: float) -> None:
     @property
     def pt100_1(self) -> float:
         if self.pt100_1_enabled:
-            minimum = float(self.options.get("pt100_1.min", 21.0))
-            maximum = float(self.options.get("pt100_1.max", 21.5))
-            return round(random.uniform(minimum, maximum), 1)
+            return round(random.uniform(self._pt100_1_min, self._pt100_1_max), 1)
         return float("nan")
 
     @property
     def pt100_2(self) -> float:
         if self.pt100_2_enabled:
-            minimum = float(self.options.get("pt100_2.min", 22.0))
-            maximum = float(self.options.get("pt100_2.max", 22.5))
-            return round(random.uniform(minimum, maximum), 1)
+            return round(random.uniform(self._pt100_2_min, self._pt100_2_max), 1)
         return float("nan")
 
     @property
@@ -203,7 +208,7 @@ def create_pc_data(self) -> list[str]:
 
     @message(r"\*IDN\?$")
     def get_idn(self) -> str:
-        return self.options.get("identity", self.IDENTITY)
+        return self._identity
 
     @message(r"SET:NEW_ADDR (\d+)$")
     def set_new_addr(self, address) -> str:
@@ -586,12 +591,12 @@ def get_env(self) -> str:
 
     @message(r"GET:VERSION \?$")
     def get_version(self) -> str:
-        return self.options.get("version", self.VERSION)
+        return self._version
 
     @message(r".*")
     def unknown_message(self) -> str:
         return format_error(999)
 
 
 if __name__ == "__main__":
-    run(EnvironBoxEmulator())
+    run(EnvironBoxEmulator)