diff --git a/src/comet/emulator/__init__.py b/src/comet/emulator/__init__.py index 0cc2568..7935d54 100644 --- a/src/comet/emulator/__init__.py +++ b/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", diff --git a/src/comet/emulator/__main__.py b/src/comet/emulator/__main__.py index ef897a6..b89a3e4 100644 --- a/src/comet/emulator/__main__.py +++ b/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) diff --git a/src/comet/emulator/cts/itc.py b/src/comet/emulator/cts/itc.py index 4711b97..628e4ee 100644 --- a/src/comet/emulator/cts/itc.py +++ b/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) diff --git a/src/comet/emulator/emulator.py b/src/comet/emulator/emulator.py index 66a2876..36395e7 100644 --- a/src/comet/emulator/emulator.py +++ b/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) diff --git a/src/comet/emulator/ers/ac3.py b/src/comet/emulator/ers/ac3.py index 79f1803..60bce15 100644 --- a/src/comet/emulator/ers/ac3.py +++ b/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) diff --git a/src/comet/emulator/hephy/brandbox.py b/src/comet/emulator/hephy/brandbox.py index b7730fd..4f7b029 100644 --- a/src/comet/emulator/hephy/brandbox.py +++ b/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) diff --git a/src/comet/emulator/hephy/environbox.py b/src/comet/emulator/hephy/environbox.py index f95b2ff..3ad70df 100644 --- a/src/comet/emulator/hephy/environbox.py +++ b/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,7 +591,7 @@ 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: @@ -594,4 +599,4 @@ def unknown_message(self) -> str: if __name__ == "__main__": - run(EnvironBoxEmulator()) + run(EnvironBoxEmulator) diff --git a/src/comet/emulator/hephy/shuntbox.py b/src/comet/emulator/hephy/shuntbox.py index 2ea75e3..eb07771 100644 --- a/src/comet/emulator/hephy/shuntbox.py +++ b/src/comet/emulator/hephy/shuntbox.py @@ -3,7 +3,7 @@ import random import time -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run __all__ = ["ShuntBoxEmulator"] @@ -13,26 +13,31 @@ def format_error(code: int) -> str: class ShuntBoxEmulator(Emulator): - IDENTITY: str = "ShuntBox, v1.0 (Emulator)" MEMORY_BYTES: int = 4200 CHANNELS: int = 10 SUCCESS: str = "OK" - def __init__(self) -> None: - super().__init__() - self.start_time: float = time.time() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + + self._identity = options.get("identity", "ShuntBox, v1.0 (Emulator)") + self._start_time: float = float(options.get("start_time", time.monotonic())) + self._temp_min: float = float(options.get("temp.min", 22.0)) + self._temp_max: float = float(options.get("temp.max", 26.0)) @property - def uptime(self) -> int: - return round(time.time() - self.start_time) + def _uptime(self) -> int: + return round(time.monotonic() - self._start_time) @message(r"\*IDN\?$") def get_idn(self) -> str: - return self.options.get("identity", self.IDENTITY) + return self._identity @message(r"GET:UP \?$") def get_up(self) -> str: - return format(self.uptime) + return format(self._uptime) @message(r"GET:RAM \?$") def get_ram(self) -> str: @@ -42,12 +47,12 @@ def get_ram(self) -> str: def get_temp_all(self) -> str: values = [] for i in range(self.CHANNELS): - values.append(format(random.uniform(22.0, 26.0), ".1f")) + values.append(format(random.uniform(self._temp_min, self._temp_max), ".1f")) return ",".join(values) @message(r"GET:TEMP (\d+)$") def get_temp(self, value) -> str: - return format(random.uniform(22.0, 26.0), ".1f") + return format(random.uniform(self._temp_min, self._temp_max), ".1f") @message(r"SET:REL_(ON|OFF) (\d+|ALL)$") def set_rel(self, state, value) -> str: @@ -67,4 +72,4 @@ def unknown_message(self) -> str: if __name__ == "__main__": - run(ShuntBoxEmulator()) + run(ShuntBoxEmulator) diff --git a/src/comet/emulator/iec60488.py b/src/comet/emulator/iec60488.py index 05935d2..021fd22 100644 --- a/src/comet/emulator/iec60488.py +++ b/src/comet/emulator/iec60488.py @@ -10,7 +10,7 @@ class IEC60488Emulator(Emulator): @message(r"\*IDN\?$") def get_idn(self): - return self.options.get("identity", self.IDENTITY) + return self.context.options.get("identity", self.IDENTITY) @message(r"\*ESR\?$") def get_esr(self): diff --git a/src/comet/emulator/itk/corvustt.py b/src/comet/emulator/itk/corvustt.py index 27b0bf8..41cdeae 100644 --- a/src/comet/emulator/itk/corvustt.py +++ b/src/comet/emulator/itk/corvustt.py @@ -2,16 +2,14 @@ import random import time -from collections.abc import Mapping from dataclasses import astuple, dataclass -from typing import Any -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run __all__ = ["CorvusTTEmulator"] -@dataclass +@dataclass(frozen=True, slots=True) class TableLimits: a1: float b1: float @@ -24,21 +22,27 @@ class TableLimits: class CorvusTTEmulator(Emulator): """Corvus TT (Venus-1) emulator.""" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) - self.identity: str = "Corvus 0 0 0 0" - self.version: str = "1.0" - self.mac_address: str = "00:00:00:00:00:00" - self.serial_no: str = "01011234" + options = context.options - self.x_pos: float = 0.0 - self.y_pos: float = 0.0 - self.z_pos: float = 0.0 + self.identity: str = options.get("identity", "Corvus 0 0 0 0") + self.version: str = options.get("version", "1.0") + self.mac_address: str = options.get("mac_address", "00:00:00:00:00:00") + self.serial_no: str = options.get("serial_no", "01011234") - self.x_unit: int = 1 - self.y_unit: int = 1 - self.z_unit: int = 1 + position = options.get("position", {}) + + self.x_pos: float = position.get("x", 0.0) + self.y_pos: float = position.get("y", 0.0) + self.z_pos: float = position.get("z", 0.0) + + unit = options.get("unit", {}) + + self.x_unit: int = max(1, min(3, unit.get("x", 1))) + self.y_unit: int = max(1, min(3, unit.get("y", 1))) + self.z_unit: int = max(1, min(3, unit.get("z", 1))) self.table_limits = TableLimits(0.0, 0.0, 0.0, 1000000.0, 100000.0, 25000.0) @@ -47,38 +51,12 @@ def __init__(self) -> None: self.geterror: int = 0 self.getmerror: int = 0 - self.joystick: bool = False + self.joystick: bool = options.get("joystick", False) self.status: int = 0 self.ticks_t0: float = time.monotonic() - def load_options(self, options: Mapping[str, Any]) -> None: - if isinstance(identity := options.get("identity"), str): - self.identity = identity - if isinstance(version := options.get("version"), str): - self.version = version - if isinstance(mac_address := options.get("mac_address"), str): - self.mac_address = mac_address - if isinstance(serial_no := options.get("serial_no"), str): - self.serial_no = serial_no - - if isinstance(position := options.get("position"), dict): - if isinstance(x := position.get("x"), (int, float)): - self.x_pos = float(x) - if isinstance(y := position.get("y"), (int, float)): - self.y_pos = float(y) - if isinstance(z := position.get("z"), (int, float)): - self.z_pos = float(z) - - if isinstance(unit := options.get("unit"), dict): - if isinstance(x := unit.get("x"), int): - self.x_unit = int(max(1, min(3, x))) - if isinstance(y := unit.get("y"), int): - self.y_unit = int(max(1, min(3, y))) - if isinstance(z := unit.get("z"), int): - self.z_unit = int(max(1, min(3, z))) - @message(r"identify$") def get_identify(self) -> str: return self.identity @@ -234,4 +212,4 @@ def set_nrm(self, axis) -> None: if __name__ == "__main__": - run(CorvusTTEmulator()) + run(CorvusTTEmulator) diff --git a/src/comet/emulator/itk/hydra.py b/src/comet/emulator/itk/hydra.py index 62069a1..028bb95 100644 --- a/src/comet/emulator/itk/hydra.py +++ b/src/comet/emulator/itk/hydra.py @@ -1,10 +1,8 @@ """Hydra (Venus-3) emulator.""" import random -from collections.abc import Mapping -from typing import Any -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run __all__ = ["HydraEmulator"] @@ -12,42 +10,27 @@ class HydraEmulator(Emulator): """Hydra (Venus-3) emulator.""" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) - self.identity: str = "Hydra 0 0 0 0" - self.version: float = 1.0 - self.mac_address: str = "00:00:00:00:00:00" - self.serial_no: str = "01010042" + options = context.options - self.x_pos: float = 0.0 - self.y_pos: float = 0.0 + self.identity: str = options.get("identity", "Hydra 0 0 0 0") + self.version: float = options.get("version", 1.0) + self.mac_address: str = options.get("mac_address", "00:00:00:00:00:00") + self.serial_no: str = options.get("serial_no", "01010042") + + position = options.get("position", {}) + + self.x_pos: float = position.get("x", 0.0) + self.y_pos: float = position.get("y", 0.0) self.calibrate: dict[str, int] = {"1": 3, "2": 3} self.axes_moving: int = 0 self.manual_move: int = 0 - self.cpu_temperature: float = 40.0 - - def load_options(self, options: Mapping[str, Any]) -> None: - if isinstance(identity := options.get("identity"), str): - self.identity = identity - if isinstance(version := options.get("version"), float): - self.version = version - if isinstance(mac_address := options.get("mac_address"), str): - self.mac_address = mac_address - if isinstance(serial_no := options.get("serial_no"), str): - self.serial_no = serial_no - - if isinstance(position := options.get("position"), dict): - if isinstance(x := position.get("x"), (int, float)): - self.x_pos = float(x) - if isinstance(y := position.get("y"), (int, float)): - self.y_pos = float(y) - - if isinstance(cpu_temperature := options.get("cpu_temperature"), float): - self.cpu_temperature = cpu_temperature + self.cpu_temperature: float = float(options.get("cpu_temperature", 40.0)) @message(r"identify$") def get_identify(self) -> str: @@ -133,4 +116,4 @@ def set_nrangemeasure(self, axis) -> None: if __name__ == "__main__": - run(HydraEmulator()) + run(HydraEmulator) diff --git a/src/comet/emulator/keithley/k2400.py b/src/comet/emulator/keithley/k2400.py index 92e6f0c..96eab06 100644 --- a/src/comet/emulator/keithley/k2400.py +++ b/src/comet/emulator/keithley/k2400.py @@ -1,7 +1,7 @@ import random from typing import ClassVar -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error @@ -10,8 +10,11 @@ class K2400Emulator(IEC60488Emulator): DEFAULT_VOLTAGE_PROTECTION_LEVEL: float = 210.0 - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.error_queue: list[Error] = [] self.system_beeper_state: bool = True self.system_rsense: bool = False @@ -36,6 +39,11 @@ def __init__(self) -> None: self.format_elements = FormatElements() self.format_elements.update(["VOLT", "CURR", "RES", "TIME", "STAT"]) + self.volt_min = float(options.get("volt.min", 0)) + self.volt_max = float(options.get("volt.max", 10)) + self.curr_min = float(options.get("curr.min", 1e-6)) + self.curr_max = float(options.get("curr.max", 1e-7)) + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() @@ -314,17 +322,9 @@ def set_initiate(self) -> None: ... def get_read(self) -> str: result = [] if "VOLT" in self.format_elements._values: - curr_min = float( - self.options.get("volt.min", self.source_level.get("VOLT", 0)) - ) - curr_max = float( - self.options.get("volt.max", self.source_level.get("VOLT", 0)) - ) - result.append(format(random.uniform(curr_min, curr_max), "E")) + result.append(format(random.uniform(self.volt_min, self.volt_max), "E")) if "CURR" in self.format_elements._values: - curr_min = float(self.options.get("curr.min", 1e-6)) - curr_max = float(self.options.get("curr.max", 1e-7)) - result.append(format(random.uniform(curr_min, curr_max), "E")) + result.append(format(random.uniform(self.curr_min, self.curr_max), "E")) if "RES" in self.format_elements._values: result.append(format(float("nan"))) if "TIME" in self.format_elements._values: @@ -335,9 +335,7 @@ def get_read(self) -> str: @message(r":?FETC[H]?\?$") def get_fetch(self) -> str: - curr_min = float(self.options.get("curr.min", 1e-6)) - curr_max = float(self.options.get("curr.max", 1e-7)) - return format(random.uniform(curr_min, curr_max), "E") + return format(random.uniform(self.curr_min, self.curr_max), "E") @message(r".*") def unknown_message(self) -> None: @@ -406,4 +404,4 @@ def __str__(self) -> str: if __name__ == "__main__": - run(K2400Emulator()) + run(K2400Emulator) diff --git a/src/comet/emulator/keithley/k2410.py b/src/comet/emulator/keithley/k2410.py index e56c214..20d7d9a 100644 --- a/src/comet/emulator/keithley/k2410.py +++ b/src/comet/emulator/keithley/k2410.py @@ -1,4 +1,5 @@ -from comet.emulator.keithley.k2400 import K2400Emulator, run +from comet.emulator import run +from comet.emulator.keithley.k2400 import K2400Emulator class K2410Emulator(K2400Emulator): @@ -8,4 +9,4 @@ class K2410Emulator(K2400Emulator): if __name__ == "__main__": - run(K2410Emulator()) + run(K2410Emulator) diff --git a/src/comet/emulator/keithley/k2470.py b/src/comet/emulator/keithley/k2470.py index 150a505..a4fc1f8 100644 --- a/src/comet/emulator/keithley/k2470.py +++ b/src/comet/emulator/keithley/k2470.py @@ -1,6 +1,6 @@ import random -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error, tsp_assign, tsp_print @@ -10,9 +10,12 @@ class K2470Emulator(IEC60488Emulator): DEFAULT_VOLTAGE_PROTECTION_LEVEL: float = 1050.0 - def __init__(self) -> None: - super().__init__() - self.language: str = str(self.options.get("language", self.LANGUAGE)) + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + + self.language: str = str(options.get("language", self.LANGUAGE)) self.error_queue: list[Error] = [] self.route_terminals: str = "FRON" self.output_state: bool = False @@ -31,10 +34,12 @@ def __init__(self) -> None: self.sense_average_state: dict[str, bool] = {"VOLT": False, "CURR": False} self.sense_nplc: float = 1.0 self.system_breakdown_protection: str = "OFF" + self.output_interlock_tripped = bool(options.get("interlock.tripped", True)) - @property - def output_interlock_tripped(self) -> bool: - return bool(self.options.get("interlock.tripped", True)) + self.volt_min = float(options.get("volt.min", 0)) + self.volt_max = float(options.get("volt.max", 10)) + self.curr_min = float(options.get("curr.min", 1e-6)) + self.curr_max = float(options.get("curr.max", 1e-7)) @message(r"\*LANG\?$") def get_lang(self) -> str: @@ -338,15 +343,11 @@ def unknown_message(self) -> None: self.error_queue.append(Error(101, "malformed command")) def _read_voltage(self) -> float: - volt_min = float(self.options.get("volt.min", 0)) - volt_max = float(self.options.get("volt.max", 10)) - return random.uniform(volt_min, volt_max) + return random.uniform(self.volt_min, self.volt_max) def _read_current(self) -> float: - curr_min = float(self.options.get("curr.min", 1e-6)) - curr_max = float(self.options.get("curr.max", 1e-7)) - return random.uniform(curr_min, curr_max) + return random.uniform(self.curr_min, self.curr_max) if __name__ == "__main__": - run(K2470Emulator()) + run(K2470Emulator) diff --git a/src/comet/emulator/keithley/k2657a.py b/src/comet/emulator/keithley/k2657a.py index 948494a..24fa88a 100644 --- a/src/comet/emulator/keithley/k2657a.py +++ b/src/comet/emulator/keithley/k2657a.py @@ -1,14 +1,17 @@ import random -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error, tsp_assign, tsp_print class K2657AEmulator(IEC60488Emulator): IDENTITY: str = "Keithley Inc., Model 2657A, 43768438, v1.0 (Emulator)" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.error_queue: list[Error] = [] self.beeper_enable: bool = True self.display_measure_function: int = 1 @@ -24,6 +27,9 @@ def __init__(self) -> None: self.smua_measure_nplc: float = 1.0 self.source_protectv: float = 0.0 + self.curr_min = float(options.get("curr.min", 1e-6)) + self.curr_max = float(options.get("curr.max", 1e-7)) + @message(r"reset\(\)$") def set_reset(self): self.error_queue.clear() @@ -217,9 +223,7 @@ def set_source_limit(self, function, level) -> None: @message(tsp_print(r"smua\.measure\.i\(\)")) def get_measure_i(self) -> str: - curr_min = float(self.options.get("curr.min", 1e-6)) - curr_max = float(self.options.get("curr.max", 1e-7)) - return format(random.uniform(curr_min, curr_max), "E") + return format(random.uniform(self.curr_min, self.curr_max), "E") @message(tsp_print(r"smua\.measure\.v\(\)")) def get_measure_v(self) -> str: @@ -293,4 +297,4 @@ def unknown_message(self) -> None: if __name__ == "__main__": - run(K2657AEmulator()) + run(K2657AEmulator) diff --git a/src/comet/emulator/keithley/k2700.py b/src/comet/emulator/keithley/k2700.py index c0748d8..47b8f93 100644 --- a/src/comet/emulator/keithley/k2700.py +++ b/src/comet/emulator/keithley/k2700.py @@ -1,7 +1,7 @@ import random import time -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error @@ -69,8 +69,11 @@ def format_reading(self, reading: Reading) -> str: class K2700Emulator(IEC60488Emulator): IDENTITY: str = "Keithley Inc., Model 2700, 43768438, v1.0 (Emulator)" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.reading_number: int = 0 self.format_elements: FormatElements = FormatElements() self.sense_function: str = "VOLT:DC" @@ -82,6 +85,13 @@ def __init__(self) -> None: self.trigger_delay_auto: bool = True self.trigger_delay: float = 0.001 + self.volt_min = float(options.get("volt.min", 0)) + self.volt_max = float(options.get("volt.max", 10)) + self.curr_min = float(options.get("curr.min", 1e-6)) + self.curr_max = float(options.get("curr.max", 1e-7)) + self.temp_min = float(options.get("temp.min", 24)) + self.temp_max = float(options.get("temp.max", 25)) + @message(r"\*RST$") def set_rst(self) -> None: self.reading_number = 0 @@ -228,17 +238,11 @@ def unknown_message(self) -> None: def _read(self) -> str: """Returns formatted reading.""" if self.sense_function == "VOLT:DC": - volt_min = float(self.options.get("volt.min", 0)) - volt_max = float(self.options.get("volt.max", 10)) - reading = Reading(random.uniform(volt_min, volt_max), "VDC") + reading = Reading(random.uniform(self.volt_min, self.volt_max), "VDC") elif self.sense_function == "CURR:DC": - curr_min = float(self.options.get("curr.min", 1e-6)) - curr_max = float(self.options.get("curr.max", 1e-7)) - reading = Reading(random.uniform(curr_min, curr_max), "ADC") + reading = Reading(random.uniform(self.curr_min, self.curr_max), "ADC") else: - temp_min = float(self.options.get("temp.min", 24)) - temp_max = float(self.options.get("temp.max", 25)) - reading = Reading(random.uniform(temp_min, temp_max), "") # TEMP + reading = Reading(random.uniform(self.temp_min, self.temp_max), "") # TEMP time.sleep(random.uniform(0.5, 1.0)) # rev B10 ;) reading.reading_number = self.reading_number self.reading_number += 1 @@ -246,4 +250,4 @@ def _read(self) -> str: if __name__ == "__main__": - run(K2700Emulator()) + run(K2700Emulator) diff --git a/src/comet/emulator/keithley/k4215cvu.py b/src/comet/emulator/keithley/k4215cvu.py index a2f91cc..89c002c 100644 --- a/src/comet/emulator/keithley/k4215cvu.py +++ b/src/comet/emulator/keithley/k4215cvu.py @@ -4,7 +4,7 @@ from dataclasses import astuple, dataclass from typing import ClassVar -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error @@ -30,12 +30,15 @@ class K4215CVUEmulator(IEC60488Emulator): } MODEL_MAP_INV: ClassVar[dict[int, str]] = {v: k for k, v in MODEL_MAP.items()} - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.error_queue: list[Error] = [] # Core state - self.cvu_mode: int = 0 # 0=user mode + self.cvu_mode: int = int(options.get("cvu_mode", 0)) # 0=user mode self.cvu_output: bool = False self.config_acvhi: int = 1 @@ -477,4 +480,4 @@ def unknown_message(self, request: str) -> None: if __name__ == "__main__": - run(K4215CVUEmulator()) + run(K4215CVUEmulator) diff --git a/src/comet/emulator/keithley/k6510.py b/src/comet/emulator/keithley/k6510.py index b7c0ebb..5786bcc 100644 --- a/src/comet/emulator/keithley/k6510.py +++ b/src/comet/emulator/keithley/k6510.py @@ -1,6 +1,6 @@ import random -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error __all__ = ["K6510Emulator"] @@ -9,10 +9,19 @@ class K6510Emulator(IEC60488Emulator): IDENTITY: str = "Keithley Inc., Model DAQ6510, 54313645, v1.0 (Emulator)" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.error_queue: list[Error] = [] + self.route_terminals = options.get("route.terminals", "front") + self.curr_min = float(options.get("curr.min", 1e-6)) + self.curr_max = float(options.get("curr.max", 1e-7)) + self.volt_min = float(options.get("volt.min", 1e3)) + self.volt_max = float(options.get("volt.max", 1e2)) + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() @@ -37,8 +46,7 @@ def get_system_error_next(self) -> str: @message(r":?ROUT(?:e)?:TERM(?:inal(?:s)?)?\?$") def get_route_terminals(self) -> str: - value = self.options.get("route.terminals", "front") - if value.lower().startswith("rear"): + if self.route_terminals.lower().startswith("rear"): return "REAR" return "FRON" @@ -46,15 +54,11 @@ def get_route_terminals(self) -> str: @message(r":?MEAS(?:ure)?:VOLT(?:age)?\?$") def get_measure_voltage(self) -> str: - volt_min = float(self.options.get("volt.min", 1e3)) - volt_max = float(self.options.get("volt.max", 1e2)) - return format(random.uniform(volt_min, volt_max), "E") + return format(random.uniform(self.volt_min, self.volt_max), "E") @message(r":?MEAS(?:ure)?:CURR(?:ent)?\?$") def get_measure_current(self) -> str: - curr_min = float(self.options.get("curr.min", 1e-6)) - curr_max = float(self.options.get("curr.max", 1e-7)) - return format(random.uniform(curr_min, curr_max), "E") + return format(random.uniform(self.curr_min, self.curr_max), "E") @message(r".*") def unknown_message(self) -> None: @@ -62,4 +66,4 @@ def unknown_message(self) -> None: if __name__ == "__main__": - run(K6510Emulator()) + run(K6510Emulator) diff --git a/src/comet/emulator/keithley/k6514.py b/src/comet/emulator/keithley/k6514.py index 98c5d17..aed5732 100644 --- a/src/comet/emulator/keithley/k6514.py +++ b/src/comet/emulator/keithley/k6514.py @@ -1,15 +1,18 @@ import random import time -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error class K6514Emulator(IEC60488Emulator): IDENTITY: str = "Keithley Inc., Model 5614, 43768438, v1.0 (Emulator)" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.error_queue: list[Error] = [] self.zero_check: bool = False self.zero_correction: bool = False @@ -21,6 +24,11 @@ def __init__(self) -> None: self.sense_current_range_auto: int = 1 self.sense_nplc: float = 5.0 + self.curr_min = float(options.get("curr.min", 2.5e-10)) + self.curr_max = float(options.get("curr.max", 2.5e-9)) + self.volt_min = float(options.get("volt.min", -5)) + self.volt_max = float(options.get("volt.max", +5)) + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() @@ -63,13 +71,9 @@ def set_init(self) -> None: def _reading(self) -> float: if self.sense_function == "CURR": - curr_min = float(self.options.get("curr.min", 2.5e-10)) - curr_max = float(self.options.get("curr.max", 2.5e-9)) - return random.uniform(curr_min, curr_max) + return random.uniform(self.curr_min, self.curr_max) elif self.sense_function == "VOLT": - volt_min = float(self.options.get("volt.min", -5)) - volt_max = float(self.options.get("volt.max", +5)) - return random.uniform(volt_min, volt_max) + return random.uniform(self.volt_min, self.volt_max) return 0 @message(r":?FETC[H]?\?$") @@ -177,4 +181,4 @@ def unknown_message(self) -> None: if __name__ == "__main__": - run(K6514Emulator()) + run(K6514Emulator) diff --git a/src/comet/emulator/keithley/k6517b.py b/src/comet/emulator/keithley/k6517b.py index 4b849b2..e20ca1a 100644 --- a/src/comet/emulator/keithley/k6517b.py +++ b/src/comet/emulator/keithley/k6517b.py @@ -1,15 +1,18 @@ import random import time -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error class K6517BEmulator(IEC60488Emulator): IDENTITY: str = "Keithley Inc., Model 6517B, 43768438, v1.0 (Emulator)" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + self.error_queue: list[Error] = [] self.zero_check: bool = False self.zero_correction: bool = False @@ -24,6 +27,11 @@ def __init__(self) -> None: self.source_current_limit_state: bool = False self.sense_nplc: float = 5.0 + self.curr_min = float(options.get("curr.min", 2.5e-10)) + self.curr_max = float(options.get("curr.max", 2.5e-9)) + self.volt_min = float(options.get("volt.min", -5)) + self.volt_max = float(options.get("volt.max", +5)) + @message(r"\*RST$") def set_reset(self) -> None: self.error_queue.clear() @@ -153,13 +161,9 @@ def set_init(self) -> None: ... def _reading(self) -> float: if self.sense_function == "CURR": - curr_min = float(self.options.get("curr.min", 2.5e-10)) - curr_max = float(self.options.get("curr.max", 2.5e-9)) - return random.uniform(curr_min, curr_max) + return random.uniform(self.curr_min, self.curr_max) elif self.sense_function == "VOLT": - volt_min = float(self.options.get("volt.min", -5)) - volt_max = float(self.options.get("volt.max", +5)) - return random.uniform(volt_min, volt_max) + return random.uniform(self.volt_min, self.volt_max) return 0 @message(r":?READ\?$") @@ -255,4 +259,4 @@ def unknown_message(self) -> None: if __name__ == "__main__": - run(K6517BEmulator()) + run(K6517BEmulator) diff --git a/src/comet/emulator/keithley/k707b.py b/src/comet/emulator/keithley/k707b.py index 6bfe8f8..32ea899 100644 --- a/src/comet/emulator/keithley/k707b.py +++ b/src/comet/emulator/keithley/k707b.py @@ -1,6 +1,6 @@ from typing import ClassVar -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error, tsp_print from comet.utils import combine_matrix @@ -11,8 +11,9 @@ class K707BEmulator(IEC60488Emulator): "1234", "ABCDEFGH", (format(i, "02d") for i in range(1, 13)) ) - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + self.error_queue: list[Error] = [] self.closed_channels: set[str] = set() @@ -67,4 +68,4 @@ def unknown_message(self) -> None: if __name__ == "__main__": - run(K707BEmulator()) + run(K707BEmulator) diff --git a/src/comet/emulator/keithley/k708b.py b/src/comet/emulator/keithley/k708b.py index 2b7b754..5119222 100644 --- a/src/comet/emulator/keithley/k708b.py +++ b/src/comet/emulator/keithley/k708b.py @@ -1,8 +1,7 @@ from comet.emulator import run +from comet.emulator.keithley.k707b import K707BEmulator from comet.utils import combine_matrix -from .k707b import K707BEmulator - class K708BEmulator(K707BEmulator): IDENTITY: str = "Keithley Inc., Model 708B, 43768438, v1.0 (Emulator)" @@ -10,4 +9,4 @@ class K708BEmulator(K707BEmulator): if __name__ == "__main__": - run(K708BEmulator()) + run(K708BEmulator) diff --git a/src/comet/emulator/keysight/e4980a.py b/src/comet/emulator/keysight/e4980a.py index 02ab1ee..45219de 100644 --- a/src/comet/emulator/keysight/e4980a.py +++ b/src/comet/emulator/keysight/e4980a.py @@ -1,17 +1,18 @@ import random import time -from comet.emulator import IEC60488Emulator, message, run +from comet.emulator import Context, IEC60488Emulator, message, run from comet.emulator.utils import Error class E4980AEmulator(IEC60488Emulator): IDENTITY: str = "Keysight Inc., Model E4980A, v1.0 (Emulator)" - CORRECTION_OPEN_DELAY: float = 4.0 + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options - def __init__(self) -> None: - super().__init__() self.error_queue: list[Error] = [] self.function_impedance_type: str = "CPD" self.correction_open_state: int = 0 @@ -22,6 +23,12 @@ def __init__(self) -> None: self.bias_voltage_level: float = 0.0 self.bias_state: bool = False + self.cp_min = float(options.get("cp.min", 2.5e-10)) + self.cp_max = float(options.get("cp.max", 2.5e-9)) + self.rp_min = float(options.get("rp.min", 100)) + self.rp_max = float(options.get("rp.max", 120)) + self.correction_open_delay = float(options.get("correction_open_delay", 4.0)) + @message(r"\*RST$") def set_rst(self) -> None: self.error_queue.clear() @@ -66,8 +73,7 @@ def set_correction_open_state(self, state: str) -> None: @message(r":?CORR:OPEN$") def get_correction_open(self) -> None: - delay = self.options.get("correction_open_delay", self.CORRECTION_OPEN_DELAY) - time.sleep(delay) + time.sleep(self.correction_open_delay) @message(r":?CORR:USE\?$") def get_correction_use(self) -> str: @@ -99,13 +105,8 @@ def set_correction_length(self, length: str) -> None: @message(r":?FETC[H]?(?:(?::IMP)?:FORM)?\?$") def get_fetch(self) -> str: - # TODO - cp_min = float(self.options.get("cp.min", 2.5e-10)) - cp_max = float(self.options.get("cp.max", 2.5e-9)) - rp_min = float(self.options.get("rp.min", 100)) - rp_max = float(self.options.get("rp.max", 120)) - prim = random.uniform(cp_min, cp_max) - sec = random.uniform(rp_min, rp_max) + prim = random.uniform(self.cp_min, self.cp_max) + sec = random.uniform(self.rp_min, self.rp_max) return f"{prim:E},{sec:E},{0:+d}" @message(r":?BIAS:POL:CURR(\::LEV)?\?$") @@ -138,4 +139,4 @@ def undefined_header(self): if __name__ == "__main__": - run(E4980AEmulator()) + run(E4980AEmulator) diff --git a/src/comet/emulator/marzhauser/tango.py b/src/comet/emulator/marzhauser/tango.py index 2d41da4..7beb75d 100644 --- a/src/comet/emulator/marzhauser/tango.py +++ b/src/comet/emulator/marzhauser/tango.py @@ -1,6 +1,6 @@ """TANGO emulator.""" -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run __all__ = ["TangoEmulator"] @@ -10,19 +10,37 @@ class TangoEmulator(Emulator): VERSION: str = "TANGO-MINI3-EMULATOR, Version 1.00, Mar 11 2022, 13:51:01" - def __init__(self) -> None: - super().__init__() - self.position: dict[str, float] = {"x": 0.0, "y": 0.0, "z": 0.0} + def __init__(self, context: Context) -> None: + super().__init__(context) + + options = context.options + + self.version = options.get("version", self.VERSION) + + position = options.get("position", {}) + + self.position: dict[str, float] = { + "x": position.get("x", 0.0), + "y": position.get("y", 0.0), + "z": position.get("z", 0.0), + } self.calst: dict[str, int] = {"x": 3, "y": 3, "z": 3} self.statusaxis: dict[str, str] = {"x": "@", "y": "@", "z": "@"} - self.velocity: dict[str, float] = {"x": 10.0, "y": 10.0, "z": 10.0} + + velocity = options.get("velocity", {}) + + self.velocity: dict[str, float] = { + "x": velocity.get("x", 10.0), + "y": velocity.get("y", 10.0), + "z": velocity.get("z", 10.0), + } self.autostatus: bool = True # Controller informations @message(r"\??version$") def get_version(self) -> str: - return format(self.options.get("version", self.VERSION)) + return format(self.version) @message(r"\?autostatus$") def get_autostatus(self) -> str: @@ -170,4 +188,4 @@ def action_reset(self) -> None: ... if __name__ == "__main__": - run(TangoEmulator()) + run(TangoEmulator) diff --git a/src/comet/emulator/nkt_photonics/pilas.py b/src/comet/emulator/nkt_photonics/pilas.py index aee1280..6370a45 100644 --- a/src/comet/emulator/nkt_photonics/pilas.py +++ b/src/comet/emulator/nkt_photonics/pilas.py @@ -1,6 +1,6 @@ """NKT Photonics PILAS picosecond pulsed diode laser emulator""" -from comet.emulator import Emulator, TextResponse, message, run +from comet.emulator import Context, Emulator, TextResponse, message, run __all__ = ["PILASEmulator"] @@ -17,8 +17,8 @@ class PILASEmulator(Emulator): + "laser head hardware version: PiLas_Laser_Head_PCB_Rev.2.0" ) - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) self.tune: float = 0 self.tune_mode: bool = False @@ -81,4 +81,4 @@ def get_laser_diode_temperature(self) -> str: if __name__ == "__main__": - run(PILASEmulator()) + run(PILASEmulator) diff --git a/src/comet/emulator/photonic/f3000.py b/src/comet/emulator/photonic/f3000.py index 871a0a3..310d8cb 100644 --- a/src/comet/emulator/photonic/f3000.py +++ b/src/comet/emulator/photonic/f3000.py @@ -1,6 +1,6 @@ """Photonic F3000 LED light source emulator""" -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run __all__ = ["F3000Emulator"] @@ -8,8 +8,8 @@ class F3000Emulator(Emulator): IDENTITY: str = "F3000 v2.09, Emulator" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) self.current_brightness: int = 50 self.light_enabled: bool = False @@ -37,4 +37,4 @@ def set_light_enabled(self, light_enabled: str) -> None: if __name__ == "__main__": - run(F3000Emulator()) + run(F3000Emulator) diff --git a/src/comet/emulator/resource.py b/src/comet/emulator/resource.py index bebac81..0024397 100644 --- a/src/comet/emulator/resource.py +++ b/src/comet/emulator/resource.py @@ -11,13 +11,14 @@ import time from typing import Self -from .emulator import Emulator, emulator_factory +from .emulator import Context, Emulator, emulator_cls_factory -def open_emulator(module_name: str, options: dict | None = None) -> EmulatorResource: - emulator = emulator_factory(module_name)() - if options: - emulator.options.update(options) +def open_emulator(model_urn: str, options: dict | None = None) -> EmulatorResource: + if options is None: + options = {} + context = Context(options=options) + emulator = emulator_cls_factory(model_urn)(context) return EmulatorResource(emulator) diff --git a/src/comet/emulator/rohde_schwarz/nge100.py b/src/comet/emulator/rohde_schwarz/nge100.py index 240f2d9..6ed24fd 100644 --- a/src/comet/emulator/rohde_schwarz/nge100.py +++ b/src/comet/emulator/rohde_schwarz/nge100.py @@ -2,7 +2,7 @@ import math -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run __all__ = ["NGE100Emulator"] @@ -10,8 +10,8 @@ class NGE100Emulator(Emulator): IDENTITY: str = "Rohde&Schwarz,NGE103B,5601.3800k03/101863,1.54" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) self.voltage_levels: list[float] = [0.0, 0.0, 0.0] self.current_limits: list[float] = [0.0, 00.0, 0.0] @@ -97,4 +97,4 @@ def measure_power(self) -> str: if __name__ == "__main__": - run(NGE100Emulator()) + run(NGE100Emulator) diff --git a/src/comet/emulator/rohde_schwarz/rto6.py b/src/comet/emulator/rohde_schwarz/rto6.py index 5bc8a6d..7f41e41 100644 --- a/src/comet/emulator/rohde_schwarz/rto6.py +++ b/src/comet/emulator/rohde_schwarz/rto6.py @@ -1,8 +1,7 @@ """Rohde Schwarz RTO6 oscilloscope emulator""" from comet.emulator import run - -from .rtp164 import RTP164Emulator +from comet.emulator.rohde_schwarz.rtp164 import RTP164Emulator __all__ = ["RTO6Emulator"] @@ -12,4 +11,4 @@ class RTO6Emulator(RTP164Emulator): if __name__ == "__main__": - run(RTO6Emulator()) + run(RTO6Emulator) diff --git a/src/comet/emulator/rohde_schwarz/rtp164.py b/src/comet/emulator/rohde_schwarz/rtp164.py index 9f6083b..13cfefe 100644 --- a/src/comet/emulator/rohde_schwarz/rtp164.py +++ b/src/comet/emulator/rohde_schwarz/rtp164.py @@ -1,6 +1,6 @@ """Rohde Schwarz RTP164 oscilloscope emulator""" -from comet.emulator import BinaryResponse, IEC60488Emulator, message, run +from comet.emulator import BinaryResponse, Context, IEC60488Emulator, message, run from comet.emulator.utils import SCPIError, generate_waveform, scpi_parse_bool __all__ = ["RTP164Emulator"] @@ -9,8 +9,9 @@ class RTP164Emulator(IEC60488Emulator): IDENTITY: str = "Rohde&Schwarz,RTP,1320.5007k16/123456,5.50.2.0" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) + self.error_queue: list[SCPIError] = [] self.format_border: str = "LSBF" self.format_data: str = "ASC,0" @@ -86,4 +87,4 @@ def undefined_header(self) -> None: if __name__ == "__main__": - run(RTP164Emulator()) + run(RTP164Emulator) diff --git a/src/comet/emulator/rohde_schwarz/sma100b.py b/src/comet/emulator/rohde_schwarz/sma100b.py index 135ed11..53a03d6 100644 --- a/src/comet/emulator/rohde_schwarz/sma100b.py +++ b/src/comet/emulator/rohde_schwarz/sma100b.py @@ -1,6 +1,6 @@ """Rohde Schwarz SMA100B signal generator emulator""" -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run from comet.emulator.utils import Error __all__ = ["SMA100BEmulator"] @@ -9,8 +9,8 @@ class SMA100BEmulator(Emulator): IDENTITY: str = "Rohde&Schwarz,SMA100B,1419.8888K02/120399,5.00.122.24 SP1" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) self.error_queue: list[Error] = [] @@ -84,4 +84,4 @@ def set_output(self, state) -> None: if __name__ == "__main__": - run(SMA100BEmulator()) + run(SMA100BEmulator) diff --git a/src/comet/emulator/tcpserver.py b/src/comet/emulator/tcpserver.py index dbbecc1..37c4647 100644 --- a/src/comet/emulator/tcpserver.py +++ b/src/comet/emulator/tcpserver.py @@ -10,7 +10,7 @@ from collections.abc import Iterable from dataclasses import dataclass -from .emulator import Emulator +from .emulator import Context, Emulator from .response import Response __all__ = ["TCPRequestHandler", "TCPServer", "TCPServerContext"] @@ -194,23 +194,20 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def run(emulator: Emulator) -> int: +def run(cls: type[Emulator]) -> int: """Convenience emulator runner using asyncio TCP server.""" - if not isinstance(emulator, Emulator): - raise TypeError(f"Emulator must inherit from {Emulator}") args = parse_args() - emulator.options.update({key: value for key, value in args.option}) + options = {key: value for key, value in args.option} + context = Context(options=options) + emulator = cls(context) logging.basicConfig(level=logging.INFO) - mod = inspect.getmodule(emulator.__class__) - name = ( - getattr(getattr(mod, "__spec__", None), "name", None) - or emulator.__class__.__module__ - ) + mod = inspect.getmodule(cls) + name = getattr(getattr(mod, "__spec__", None), "name", None) or cls.__module__ - context = TCPServerContext( + server_context = TCPServerContext( name=name, emulator=emulator, termination=args.termination.encode(), @@ -220,11 +217,11 @@ def run(emulator: Emulator) -> int: address = (args.host, args.port) async def main() -> None: - server = TCPServer(address, context) + server = TCPServer(address, server_context) await server.start() host, port = server.server_address - context.logger.info("starting... %s:%s", host, port) + server_context.logger.info("starting... %s:%s", host, port) stop_event = asyncio.Event() @@ -232,7 +229,7 @@ def request_shutdown() -> None: if stop_event.is_set(): return - context.logger.info("stopping... %s:%s", host, port) + server_context.logger.info("stopping... %s:%s", host, port) stop_event.set() loop = asyncio.get_running_loop() diff --git a/src/comet/emulator/thorlabs/pm100.py b/src/comet/emulator/thorlabs/pm100.py index a2e0eab..fcc1bae 100644 --- a/src/comet/emulator/thorlabs/pm100.py +++ b/src/comet/emulator/thorlabs/pm100.py @@ -2,7 +2,7 @@ import random -from comet.emulator import Emulator, message, run +from comet.emulator import Context, Emulator, message, run from comet.emulator.utils import Error __all__ = ["PM100Emulator"] @@ -11,8 +11,8 @@ class PM100Emulator(Emulator): IDENTITY: str = "Thorlabs,PM100USB,P2004525,1.4.0" - def __init__(self) -> None: - super().__init__() + def __init__(self, context: Context) -> None: + super().__init__(context) self.error_queue: list[Error] = [] @@ -60,11 +60,11 @@ def get_wavelength(self) -> int: def set_wavelength(self, wavelength) -> None: self.wavelength = wavelength - @message(r"MEAS(?:ure)?(?::SCAL(?:ar)?)?(?::POW(?:er)?)?$") + @message(r"MEAS(?:ure)?(?::SCAL(?:ar)?)?(?::POW(?:er)?)?\?$") def measure_power(self) -> str: power = random.uniform(1e-9, 2e-9) return format(power, "E") if __name__ == "__main__": - run(PM100Emulator()) + run(PM100Emulator) diff --git a/tests/test_emulator_cts_itc.py b/tests/test_emulator_cts_itc.py index 75f5a5e..9f0abae 100644 --- a/tests/test_emulator_cts_itc.py +++ b/tests/test_emulator_cts_itc.py @@ -2,12 +2,13 @@ import pytest +from comet.emulator import Context from comet.emulator.cts.itc import ITCEmulator @pytest.fixture def emulator(): - return ITCEmulator() + return ITCEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_emulator.py b/tests/test_emulator_emulator.py index 9d4449c..d70e954 100644 --- a/tests/test_emulator_emulator.py +++ b/tests/test_emulator_emulator.py @@ -2,21 +2,21 @@ import pytest -from comet.emulator.emulator import emulator_factory, get_routes +from comet.emulator.emulator import emulator_cls_factory, get_routes from comet.emulator.keithley.k2410 import K2410Emulator -def test_emulator_factory(): - cls = emulator_factory("urn:comet:model:keithley:2410") +def test_emulator_cls_factory(): + cls = emulator_cls_factory("urn:comet:model:keithley:2410") assert cls is K2410Emulator -def test_emulator_factory_not_found(): +def test_emulator_cls_factory_not_found(): with warnings.catch_warnings(): warnings.simplefilter("ignore") with pytest.raises(ModuleNotFoundError): - emulator_factory("shrubbery.ni") + emulator_cls_factory("shrubbery.ni") def test_get_routes(): diff --git a/tests/test_emulator_ers_ac3.py b/tests/test_emulator_ers_ac3.py index 48c7458..0c3d302 100644 --- a/tests/test_emulator_ers_ac3.py +++ b/tests/test_emulator_ers_ac3.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.ers.ac3 import AC3Emulator, Mode @pytest.fixture def emulator(): - return AC3Emulator() + return AC3Emulator(Context()) def test_identify(emulator): diff --git a/tests/test_emulator_hephy_brandbox.py b/tests/test_emulator_hephy_brandbox.py index bb5973e..1c0d420 100644 --- a/tests/test_emulator_hephy_brandbox.py +++ b/tests/test_emulator_hephy_brandbox.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.hephy.brandbox import BrandBoxEmulator @pytest.fixture def emulator(): - return BrandBoxEmulator() + return BrandBoxEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_hephy_environbox.py b/tests/test_emulator_hephy_environbox.py index 05cad14..101c619 100644 --- a/tests/test_emulator_hephy_environbox.py +++ b/tests/test_emulator_hephy_environbox.py @@ -2,25 +2,23 @@ import pytest +from comet.emulator import Context from comet.emulator.hephy.environbox import EnvironBoxEmulator @pytest.fixture def emulator(): - emulator = EnvironBoxEmulator() - emulator.options.update( - { - "box_temperature.min": 24.0, - "box_temperature.max": 24.0, - "box_humidity.min": 40.0, - "box_humidity.max": 40.0, - "pt100_1.min": 21.5, - "pt100_1.max": 21.5, - "pt100_2.min": 22.5, - "pt100_2.max": 22.5, - } - ) - return emulator + options = { + "box_temperature.min": 24.0, + "box_temperature.max": 24.0, + "box_humidity.min": 40.0, + "box_humidity.max": 40.0, + "pt100_1.min": 21.5, + "pt100_1.max": 21.5, + "pt100_2.min": 22.5, + "pt100_2.max": 22.5, + } + return EnvironBoxEmulator(Context(options=options)) def test_basic(emulator): diff --git a/tests/test_emulator_hephy_shuntbox.py b/tests/test_emulator_hephy_shuntbox.py index 0a024a0..e8badab 100644 --- a/tests/test_emulator_hephy_shuntbox.py +++ b/tests/test_emulator_hephy_shuntbox.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.hephy.shuntbox import ShuntBoxEmulator @pytest.fixture def emulator(): - return ShuntBoxEmulator() + return ShuntBoxEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_iec60488.py b/tests/test_emulator_iec60488.py index 898377e..376825c 100644 --- a/tests/test_emulator_iec60488.py +++ b/tests/test_emulator_iec60488.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.iec60488 import IEC60488Emulator @pytest.fixture def emulator(): - return IEC60488Emulator() + return IEC60488Emulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_itk_corvustt.py b/tests/test_emulator_itk_corvustt.py index 74a15e9..abb8277 100644 --- a/tests/test_emulator_itk_corvustt.py +++ b/tests/test_emulator_itk_corvustt.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.itk.corvustt import CorvusTTEmulator @pytest.fixture def emulator(): - return CorvusTTEmulator() + return CorvusTTEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_itk_hydra.py b/tests/test_emulator_itk_hydra.py index 7c8af13..97f2323 100644 --- a/tests/test_emulator_itk_hydra.py +++ b/tests/test_emulator_itk_hydra.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.itk.hydra import HydraEmulator @pytest.fixture def emulator(): - return HydraEmulator() + return HydraEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k2400.py b/tests/test_emulator_keithley_k2400.py index 5a0c81a..50fb10e 100644 --- a/tests/test_emulator_keithley_k2400.py +++ b/tests/test_emulator_keithley_k2400.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k2400 import K2400Emulator @pytest.fixture def emulator(): - return K2400Emulator() + return K2400Emulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k2470.py b/tests/test_emulator_keithley_k2470.py index b510e37..d03f31d 100644 --- a/tests/test_emulator_keithley_k2470.py +++ b/tests/test_emulator_keithley_k2470.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k2470 import K2470Emulator @pytest.fixture def emulator(): - return K2470Emulator() + return K2470Emulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k2657a.py b/tests/test_emulator_keithley_k2657a.py index d862839..45779f0 100644 --- a/tests/test_emulator_keithley_k2657a.py +++ b/tests/test_emulator_keithley_k2657a.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k2657a import K2657AEmulator @pytest.fixture def emulator(): - return K2657AEmulator() + return K2657AEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k2700.py b/tests/test_emulator_keithley_k2700.py index 70f7ed1..31d0fd5 100644 --- a/tests/test_emulator_keithley_k2700.py +++ b/tests/test_emulator_keithley_k2700.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k2700 import K2700Emulator @pytest.fixture def emulator(): - return K2700Emulator() + return K2700Emulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k4215cvu.py b/tests/test_emulator_keithley_k4215cvu.py index 039d2cc..4cb02e7 100644 --- a/tests/test_emulator_keithley_k4215cvu.py +++ b/tests/test_emulator_keithley_k4215cvu.py @@ -2,12 +2,13 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k4215cvu import K4215CVUEmulator @pytest.fixture def emulator(): - return K4215CVUEmulator() + return K4215CVUEmulator(Context()) def assert_two_floats_csv(s: str): diff --git a/tests/test_emulator_keithley_k6510.py b/tests/test_emulator_keithley_k6510.py index 2d7ab5b..319e2e7 100644 --- a/tests/test_emulator_keithley_k6510.py +++ b/tests/test_emulator_keithley_k6510.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k6510 import K6510Emulator @pytest.fixture def emulator(): - return K6510Emulator() + return K6510Emulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k6517b.py b/tests/test_emulator_keithley_k6517b.py index b0bc422..b6b2657 100644 --- a/tests/test_emulator_keithley_k6517b.py +++ b/tests/test_emulator_keithley_k6517b.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k6517b import K6517BEmulator @pytest.fixture def emulator(): - return K6517BEmulator() + return K6517BEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_keithley_k707b.py b/tests/test_emulator_keithley_k707b.py index 2ede7da..5d9a076 100644 --- a/tests/test_emulator_keithley_k707b.py +++ b/tests/test_emulator_keithley_k707b.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k707b import K707BEmulator @pytest.fixture def emulator(): - return K707BEmulator() + return K707BEmulator(Context()) def test_constants(emulator): diff --git a/tests/test_emulator_keithley_k708b.py b/tests/test_emulator_keithley_k708b.py index 91f3051..2be21ff 100644 --- a/tests/test_emulator_keithley_k708b.py +++ b/tests/test_emulator_keithley_k708b.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keithley.k708b import K708BEmulator @pytest.fixture def emulator(): - return K708BEmulator() + return K708BEmulator(Context()) def test_constants(emulator): diff --git a/tests/test_emulator_keysight_e4980a.py b/tests/test_emulator_keysight_e4980a.py index a5cd4e0..93bc0af 100644 --- a/tests/test_emulator_keysight_e4980a.py +++ b/tests/test_emulator_keysight_e4980a.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.keysight.e4980a import E4980AEmulator @pytest.fixture def emulator(): - return E4980AEmulator() + return E4980AEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_marzhauser_tango.py b/tests/test_emulator_marzhauser_tango.py index d35cfd8..5cd6a4b 100644 --- a/tests/test_emulator_marzhauser_tango.py +++ b/tests/test_emulator_marzhauser_tango.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.marzhauser.tango import TangoEmulator @pytest.fixture def emulator(): - emulator = TangoEmulator() + emulator = TangoEmulator(Context()) emulator("!autostatus 0") return emulator diff --git a/tests/test_emulator_nkt_photonics_pilas.py b/tests/test_emulator_nkt_photonics_pilas.py index 6a19465..559e55c 100644 --- a/tests/test_emulator_nkt_photonics_pilas.py +++ b/tests/test_emulator_nkt_photonics_pilas.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.nkt_photonics.pilas import PILASEmulator @pytest.fixture def emulator(): - return PILASEmulator() + return PILASEmulator(Context()) def test_identify(emulator): diff --git a/tests/test_emulator_photonic_f3000.py b/tests/test_emulator_photonic_f3000.py index bc9b4e2..31ddcb0 100644 --- a/tests/test_emulator_photonic_f3000.py +++ b/tests/test_emulator_photonic_f3000.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.photonic.f3000 import F3000Emulator @pytest.fixture def emulator(): - return F3000Emulator() + return F3000Emulator(Context()) def test_identify(emulator): diff --git a/tests/test_emulator_rohde_schwarz_nge100.py b/tests/test_emulator_rohde_schwarz_nge100.py index 6502571..0e8b7fa 100644 --- a/tests/test_emulator_rohde_schwarz_nge100.py +++ b/tests/test_emulator_rohde_schwarz_nge100.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.rohde_schwarz.nge100 import NGE100Emulator @pytest.fixture def emulator(): - return NGE100Emulator() + return NGE100Emulator(Context()) def test_identify(emulator): diff --git a/tests/test_emulator_rohde_schwarz_rto6.py b/tests/test_emulator_rohde_schwarz_rto6.py index 1479ee0..368a2af 100644 --- a/tests/test_emulator_rohde_schwarz_rto6.py +++ b/tests/test_emulator_rohde_schwarz_rto6.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.rohde_schwarz.rto6 import RTO6Emulator @pytest.fixture def emulator(): - return RTO6Emulator() + return RTO6Emulator(Context()) def test_identify(emulator): diff --git a/tests/test_emulator_rohde_schwarz_rtp164.py b/tests/test_emulator_rohde_schwarz_rtp164.py index eb03dfd..a920ccf 100644 --- a/tests/test_emulator_rohde_schwarz_rtp164.py +++ b/tests/test_emulator_rohde_schwarz_rtp164.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.rohde_schwarz.rtp164 import RTP164Emulator @pytest.fixture def emulator(): - return RTP164Emulator() + return RTP164Emulator(Context()) def test_identify(emulator): diff --git a/tests/test_emulator_rohde_schwarz_sma100b.py b/tests/test_emulator_rohde_schwarz_sma100b.py index 5edc50a..9d28e9d 100644 --- a/tests/test_emulator_rohde_schwarz_sma100b.py +++ b/tests/test_emulator_rohde_schwarz_sma100b.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.rohde_schwarz.sma100b import SMA100BEmulator @pytest.fixture def emulator(): - return SMA100BEmulator() + return SMA100BEmulator(Context()) def test_basic(emulator): diff --git a/tests/test_emulator_tcpserver.py b/tests/test_emulator_tcpserver.py index b2e9024..00935cb 100644 --- a/tests/test_emulator_tcpserver.py +++ b/tests/test_emulator_tcpserver.py @@ -1,6 +1,5 @@ import argparse import asyncio -import inspect import logging import types from dataclasses import dataclass @@ -8,7 +7,7 @@ import pytest -from comet.emulator import Emulator, tcpserver +from comet.emulator import Context, Emulator, tcpserver from comet.emulator.response import Response, TextResponse from comet.emulator.tcpserver import TCPRequestHandler, TCPServer, TCPServerContext @@ -225,7 +224,7 @@ async def fake_sleep(delay): class TestEmulator(Emulator): def __init__(self, response: Response | list[Response] | None): - super().__init__() + super().__init__(Context()) self.response = response def __call__(self, message: str) -> Response | list[Response] | None: @@ -357,52 +356,3 @@ def test_parse_args_defaults(monkeypatch): assert args.termination == "\n" assert args.request_delay == 0.1 assert args.option == [] - - -def test_run_rejects_non_emulator(): - with pytest.raises(TypeError, match="Emulator must inherit from"): - tcpserver.run(object()) # type: ignore - - -def test_run_configures_options_and_returns_zero(monkeypatch): - class BaseEmulator: - def __init__(self): - self.options = {} - - monkeypatch.setattr(tcpserver, "Emulator", BaseEmulator) - - emulator = BaseEmulator() - - monkeypatch.setattr( - tcpserver, - "parse_args", - lambda: argparse.Namespace( - host="127.0.0.1", - port=9999, - termination="\n", - request_delay=0.01, - option=[("mode", "test"), ("version", "1")], - ), - ) - - monkeypatch.setattr( - inspect, - "getmodule", - lambda cls: types.SimpleNamespace( - __spec__=types.SimpleNamespace(name="pkg.tcpserver") - ), - ) - - ran = {} - - def fake_asyncio_run(coro): - ran["called"] = True - coro.close() - - monkeypatch.setattr(asyncio, "run", fake_asyncio_run) - - rc = tcpserver.run(cast(Emulator, emulator)) - - assert rc == 0 - assert ran["called"] is True - assert emulator.options == {"mode": "test", "version": "1"} diff --git a/tests/test_emulator_thorlabs_pm100.py b/tests/test_emulator_thorlabs_pm100.py index d94f8f1..5a8aee2 100644 --- a/tests/test_emulator_thorlabs_pm100.py +++ b/tests/test_emulator_thorlabs_pm100.py @@ -1,11 +1,12 @@ import pytest +from comet.emulator import Context from comet.emulator.thorlabs.pm100 import PM100Emulator @pytest.fixture def emulator(): - return PM100Emulator() + return PM100Emulator(Context()) def test_basic(emulator): @@ -27,7 +28,6 @@ def test_wavelength(emulator): def test_measure_power(emulator): - - power = float(emulator("MEASure:SCALar:POWer")) + power = float(emulator("MEASure:SCALar:POWer?")) assert power >= 1e-9 assert power <= 2e-9