From 712ba17de058096186f6d5206e228c6d7d8080f3 Mon Sep 17 00:00:00 2001 From: appel_c Date: Thu, 20 Mar 2025 20:25:55 +0100 Subject: [PATCH] refactor: improve typehints, remove redundant signal --- ophyd_devices/sim/sim_monitor.py | 1 + ophyd_devices/sim/sim_signals.py | 139 +++---------------------------- 2 files changed, 11 insertions(+), 129 deletions(-) diff --git a/ophyd_devices/sim/sim_monitor.py b/ophyd_devices/sim/sim_monitor.py index 1d7c7e9..24be75c 100644 --- a/ophyd_devices/sim/sim_monitor.py +++ b/ophyd_devices/sim/sim_monitor.py @@ -180,6 +180,7 @@ class SimMonitorAsync(PSIDeviceBase, SimMonitorAsyncControl): if async_update not in ["extend", "append"]: raise ValueError(f"Invalid async_update value for device {self.name}: {async_update}") + metadata = None if async_update == "extend": metadata = {"async_update": {"type": "add", "max_shape": [None]}} elif async_update == "append": diff --git a/ophyd_devices/sim/sim_signals.py b/ophyd_devices/sim/sim_signals.py index 9cc443d..850d117 100644 --- a/ophyd_devices/sim/sim_signals.py +++ b/ophyd_devices/sim/sim_signals.py @@ -1,6 +1,7 @@ """Module for signals of the ophyd_devices simulation.""" import time +from typing import Any import numpy as np from bec_lib import bec_logger @@ -55,25 +56,25 @@ class SetableSignal(Signal): self._update_sim_state(value) self._metadata.update(write_access=True) - def _update_sim_state(self, value: any) -> None: + def _update_sim_state(self, value: Any) -> None: """Update the readback value.""" if self.sim: self.sim.update_sim_state(self.name, value) - def _get_value(self) -> any: + def _get_value(self) -> Any: """Update the timestamp of the readback value.""" if self.sim: return self.sim.sim_state[self.name]["value"] return self._value - def _get_timestamp(self) -> any: + def _get_timestamp(self) -> float: """Update the timestamp of the readback value.""" if self.sim: return self.sim.sim_state[self.name]["timestamp"] return time.time() # pylint: disable=arguments-differ - def get(self, **kwargs): + def get(self, **kwargs) -> Any: """Get the current position of the simulated device. Core function for signal. @@ -82,7 +83,7 @@ class SetableSignal(Signal): return self._value # pylint: disable=arguments-differ - def put(self, value): + def put(self, value) -> None: """Put the value to the simulated device. Core function for signal. @@ -90,7 +91,8 @@ class SetableSignal(Signal): self.check_value(value) self._update_sim_state(value) self._value = value - self._run_subs(sub_type=self.SUB_VALUE, value=value) + # self._run_subs(sub_type=self.SUB_VALUE, old_value=old_value, value=value, **md_for_callback) + super().put(value) def set(self, value): """Set method""" @@ -110,7 +112,7 @@ class SetableSignal(Signal): return res @property - def timestamp(self): + def timestamp(self) -> float: """Timestamp of the readback value""" return self._get_timestamp() @@ -151,7 +153,7 @@ class ReadOnlySignal(Signal): ): super().__init__(*args, name=name, parent=parent, value=value, kind=kind, **kwargs) self._metadata.update(connected=True, write_access=False) - self._value = value + self._value = value # In a signal, this is self._readback self.precision = precision self.compute_readback = compute_readback self.sim = sim if sim is not None else getattr(self.parent, "sim", None) @@ -205,124 +207,3 @@ class ReadOnlySignal(Signal): if self.sim: return self._get_timestamp() return time.time() - - -class CustomSetableSignal(BECDeviceBase): - """Custom signal for simulated devices. The custom signal can be read-only, setable or computed. - In comparison to above, this signal is not a class from ophyd, but an own implementation of a signal. - - It works in the same fashion as the SetableSignal and ReadOnlySignal, however, it is - not needed to initiate it as a Component (ophyd) within the parent device class. - - >>> signal = SetableSignal(name="signal", parent=parent, value=0) - - Parameters - ---------- - - name (string) : Name of the signal - parent (object) : Parent object of the signal, default none. - value (any) : Initial value of the signal, default 0. - kind (int) : Kind of the signal, default Kind.normal. - precision (float) : Precision of the signal, default PRECISION. - """ - - USER_ACCESS = ["put", "get", "set"] - - def __init__( - self, - name: str, - *args, - parent=None, - value: any = 0, - kind: int = Kind.normal, - precision: float = PRECISION, - sim=None, - **kwargs, - ): - if parent: - name = f"{parent.name}_{name}" - super().__init__(*args, name=name, parent=parent, kind=kind, **kwargs) - self._metadata = {"connected": self.connected, "write_access": True} - self._value = value - self._timestamp = time.time() - self._dtype = type(value) - self._shape = self._get_shape(value) - self.precision = precision - self.sim = sim if sim is not None else getattr(self.parent, "sim", None) - self._update_sim_state(value) - - def _get_shape(self, value: any) -> list: - """Get the shape of the value. - **Note: This logic is from ophyd, and replicated here. - There would be more sophisticated ways, but to keep it consistent, it is replicated here.** - """ - if isinstance(value, np.ndarray): - return list(value.shape) - if isinstance(value, list): - return len(value) - return [] - - def _update_sim_state(self, value: any) -> None: - """Update the readback value.""" - if self.sim: - self.sim.update_sim_state(self.name, value) - - def _get_value(self) -> any: - """Update the timestamp of the readback value.""" - if self.sim: - return self.sim.sim_state[self.name]["value"] - return self._value - - def _get_timestamp(self) -> any: - """Update the timestamp of the readback value.""" - if self.sim: - return self.sim.sim_state[self.name]["timestamp"] - return self._timestamp - - # pylint: disable=arguments-differ - def get(self): - """Get the current position of the simulated device. - - Core function for signal. - """ - self._value = self._get_value() - return self._value - - # pylint: disable=arguments-differ - def put(self, value): - """Put the value to the simulated device. - - Core function for signal. - """ - self._update_sim_state(value) - self._value = value - self._timestamp = time.time() - - def describe(self): - """Describe the readback signal. - - Core function for signal. - """ - res = { - self.name: {"source": str(self.__class__), "dtype": self._dtype, "shape": self._shape} - } - if self.precision is not None: - res[self.name]["precision"] = self.precision - return res - - def set(self, value): - """Set method""" - self.put(value) - - @property - def timestamp(self): - """Timestamp of the readback value""" - return self._get_timestamp() - - def read(self): - """Read method""" - return {self.name: {"value": self.get(), "timestamp": self.timestamp}} - - def read_configuration(self): - """Read method""" - return self.read()