From 1e8a00a0ce062fb94c49df7e4d34f0fd2f802927 Mon Sep 17 00:00:00 2001 From: Devin Burke Date: Thu, 19 Sep 2024 11:14:37 +0200 Subject: [PATCH] New formatting to comply with ruff --- .../tango/base_devices/_base_device.py | 30 +++-- .../tango/base_devices/_tango_readable.py | 9 +- src/ophyd_async/tango/demo/_counter.py | 4 +- src/ophyd_async/tango/demo/_mover.py | 9 +- src/ophyd_async/tango/signal/_signal.py | 41 ++++--- .../tango/signal/_tango_transport.py | 104 +++++++++--------- tests/tango/test_base_device.py | 9 +- tests/tango/test_tango_signals.py | 32 +++--- 8 files changed, 117 insertions(+), 121 deletions(-) diff --git a/src/ophyd_async/tango/base_devices/_base_device.py b/src/ophyd_async/tango/base_devices/_base_device.py index 5bc7ea4bc0..db009a8fb9 100644 --- a/src/ophyd_async/tango/base_devices/_base_device.py +++ b/src/ophyd_async/tango/base_devices/_base_device.py @@ -1,10 +1,6 @@ from __future__ import annotations from typing import ( - Dict, - Optional, - Tuple, - Type, TypeVar, Union, get_args, @@ -22,7 +18,7 @@ make_backend, tango_signal_auto, ) -from tango import DeviceProxy as SyncDeviceProxy +from tango import DeviceProxy as DeviceProxy from tango.asyncio import DeviceProxy as AsyncDeviceProxy T = TypeVar("T") @@ -44,15 +40,15 @@ class TangoDevice(Device): """ trl: str = "" - proxy: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]] = None - _polling: Tuple[bool, float, float, float] = (False, 0.1, None, 0.1) - _signal_polling: Dict[str, Tuple[bool, float, float, float]] = {} + proxy: DeviceProxy | None = None + _polling: tuple[bool, float, float, float] = (False, 0.1, None, 0.1) + _signal_polling: dict[str, tuple[bool, float, float, float]] = {} _poll_only_annotated_signals: bool = True def __init__( self, - trl: Optional[str] = None, - device_proxy: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]] = None, + trl: str | None = None, + device_proxy: DeviceProxy | None = None, name: str = "", ) -> None: self.trl = trl if trl else "" @@ -112,10 +108,10 @@ def register_signals(self): def tango_polling( - polling: Optional[ - Union[Tuple[float, float, float], Dict[str, Tuple[float, float, float]]] - ] = None, - signal_polling: Optional[Dict[str, Tuple[float, float, float]]] = None, + polling: tuple[float, float, float] + | dict[str, tuple[float, float, float]] + | None = None, + signal_polling: dict[str, tuple[float, float, float]] | None = None, ): """ Class decorator to configure polling for Tango devices. @@ -172,7 +168,7 @@ def decorator(cls): def tango_create_children_from_annotations( - device: TangoDevice, included_optional_fields: Tuple[str, ...] = () + device: TangoDevice, included_optional_fields: tuple[str, ...] = () ): """Initialize blocks at __init__ of `device`.""" for name, device_type in get_type_hints(type(device)).items(): @@ -227,7 +223,7 @@ async def fill_proxy_entries(device: TangoDevice): raise e -def _strip_union(field: Union[Union[T], T]) -> Tuple[T, bool]: +def _strip_union(field: T | T) -> tuple[T, bool]: if get_origin(field) is Union: args = get_args(field) is_optional = type(None) in args @@ -237,7 +233,7 @@ def _strip_union(field: Union[Union[T], T]) -> Tuple[T, bool]: return field, False -def _strip_device_vector(field: Union[Type[Device]]) -> Tuple[bool, Type[Device]]: +def _strip_device_vector(field: type[Device]) -> tuple[bool, type[Device]]: if get_origin(field) is DeviceVector: return True, get_args(field)[0] return False, field diff --git a/src/ophyd_async/tango/base_devices/_tango_readable.py b/src/ophyd_async/tango/base_devices/_tango_readable.py index 94e85d96a3..44bfc93888 100644 --- a/src/ophyd_async/tango/base_devices/_tango_readable.py +++ b/src/ophyd_async/tango/base_devices/_tango_readable.py @@ -1,13 +1,10 @@ from __future__ import annotations -from typing import Optional, Union - from ophyd_async.core import ( StandardReadable, ) from ophyd_async.tango.base_devices._base_device import TangoDevice -from tango import DeviceProxy as SyncDeviceProxy -from tango.asyncio import DeviceProxy as AsyncDeviceProxy +from tango import DeviceProxy class TangoReadable(TangoDevice, StandardReadable): @@ -26,8 +23,8 @@ class TangoReadable(TangoDevice, StandardReadable): def __init__( self, - trl: Optional[str] = None, - device_proxy: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]] = None, + trl: str | None = None, + device_proxy: DeviceProxy | None = None, name: str = "", ) -> None: TangoDevice.__init__(self, trl, device_proxy=device_proxy, name=name) diff --git a/src/ophyd_async/tango/demo/_counter.py b/src/ophyd_async/tango/demo/_counter.py index 3a6892b21f..20d6bdcb74 100644 --- a/src/ophyd_async/tango/demo/_counter.py +++ b/src/ophyd_async/tango/demo/_counter.py @@ -1,5 +1,3 @@ -from typing import Optional - from ophyd_async.core import ( DEFAULT_TIMEOUT, AsyncStatus, @@ -23,7 +21,7 @@ class TangoCounter(TangoReadable): start: SignalX reset: SignalX - def __init__(self, trl: Optional[str] = "", name=""): + def __init__(self, trl: str | None = "", name=""): super().__init__(trl, name=name) self.add_readables([self.counts], HintedSignal) self.add_readables([self.sample_time], ConfigSignal) diff --git a/src/ophyd_async/tango/demo/_mover.py b/src/ophyd_async/tango/demo/_mover.py index 582aa5e344..8705e0b57f 100644 --- a/src/ophyd_async/tango/demo/_mover.py +++ b/src/ophyd_async/tango/demo/_mover.py @@ -1,13 +1,12 @@ import asyncio -from typing import Optional from bluesky.protocols import Movable, Stoppable from ophyd_async.core import ( + CALCULATE_TIMEOUT, DEFAULT_TIMEOUT, AsyncStatus, CalculatableTimeout, - CalculateTimeout, ConfigSignal, HintedSignal, SignalRW, @@ -30,19 +29,19 @@ class TangoMover(TangoReadable, Movable, Stoppable): velocity: SignalRW[float] _stop: SignalX - def __init__(self, trl: Optional[str] = "", name=""): + def __init__(self, trl: str | None = "", name=""): super().__init__(trl, name=name) self.add_readables([self.position], HintedSignal) self.add_readables([self.velocity], ConfigSignal) self._set_success = True @WatchableAsyncStatus.wrap - async def set(self, value: float, timeout: CalculatableTimeout = CalculateTimeout): + async def set(self, value: float, timeout: CalculatableTimeout = CALCULATE_TIMEOUT): self._set_success = True (old_position, velocity) = await asyncio.gather( self.position.get_value(), self.velocity.get_value() ) - if timeout is CalculateTimeout: + if timeout is CALCULATE_TIMEOUT: assert velocity > 0, "Motor has zero velocity" timeout = abs(value - old_position) / velocity + DEFAULT_TIMEOUT diff --git a/src/ophyd_async/tango/signal/_signal.py b/src/ophyd_async/tango/signal/_signal.py index 0875b968ff..7698290efa 100644 --- a/src/ophyd_async/tango/signal/_signal.py +++ b/src/ophyd_async/tango/signal/_signal.py @@ -3,7 +3,6 @@ from __future__ import annotations from enum import Enum, IntEnum -from typing import Optional, Type, Union import numpy.typing as npt @@ -12,24 +11,24 @@ TangoSignalBackend, get_python_type, ) -from tango import AttrDataFormat, AttrWriteType, CmdArgType, DevState -from tango.asyncio import DeviceProxy +from tango import AttrDataFormat, AttrWriteType, CmdArgType, DeviceProxy, DevState +from tango.asyncio import DeviceProxy as AsyncDeviceProxy def make_backend( - datatype: Optional[Type[T]], - read_trl: Optional[str] = "", - write_trl: Optional[str] = "", - device_proxy: Optional[DeviceProxy] = None, + datatype: type[T] | None, + read_trl: str | None = "", + write_trl: str | None = "", + device_proxy: DeviceProxy | None = None, ) -> TangoSignalBackend: return TangoSignalBackend(datatype, read_trl, write_trl, device_proxy) def tango_signal_rw( - datatype: Type[T], + datatype: type[T], read_trl: str, - write_trl: Optional[str] = None, - device_proxy: Optional[DeviceProxy] = None, + write_trl: str | None = None, + device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = "", ) -> SignalRW[T]: @@ -55,9 +54,9 @@ def tango_signal_rw( def tango_signal_r( - datatype: Type[T], + datatype: type[T], read_trl: str, - device_proxy: Optional[DeviceProxy] = None, + device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = "", ) -> SignalR[T]: @@ -81,9 +80,9 @@ def tango_signal_r( def tango_signal_w( - datatype: Type[T], + datatype: type[T], write_trl: str, - device_proxy: Optional[DeviceProxy] = None, + device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = "", ) -> SignalW[T]: @@ -108,7 +107,7 @@ def tango_signal_w( def tango_signal_x( write_trl: str, - device_proxy: Optional[DeviceProxy] = None, + device_proxy: DeviceProxy | None = None, timeout: float = DEFAULT_TIMEOUT, name: str = "", ) -> SignalX: @@ -130,13 +129,13 @@ def tango_signal_x( async def tango_signal_auto( - datatype: Optional[Type[T]] = None, + datatype: type[T] | None = None, *, trl: str, - device_proxy: Optional[DeviceProxy], + device_proxy: DeviceProxy | None, timeout: float = DEFAULT_TIMEOUT, name: str = "", -) -> Union[SignalW, SignalX, SignalR, SignalRW, None]: +) -> SignalW | SignalX | SignalR | SignalRW | None: try: signal_character = await infer_signal_character(trl, device_proxy) except RuntimeError as e: @@ -159,10 +158,10 @@ async def tango_signal_auto( return SignalX(backend=backend, timeout=timeout, name=name) -async def infer_python_type(trl: str = "", proxy: DeviceProxy = None) -> Type[T]: +async def infer_python_type(trl: str = "", proxy: DeviceProxy = None) -> type[T]: device_trl, tr_name = trl.rsplit("/", 1) if proxy is None: - dev_proxy = await DeviceProxy(device_trl) + dev_proxy = await AsyncDeviceProxy(device_trl) else: dev_proxy = proxy @@ -189,7 +188,7 @@ async def infer_python_type(trl: str = "", proxy: DeviceProxy = None) -> Type[T] async def infer_signal_character(trl, proxy: DeviceProxy = None) -> str: device_trl, tr_name = trl.rsplit("/", 1) if proxy is None: - dev_proxy = await DeviceProxy(device_trl) + dev_proxy = await AsyncDeviceProxy(device_trl) else: dev_proxy = proxy diff --git a/src/ophyd_async/tango/signal/_tango_transport.py b/src/ophyd_async/tango/signal/_tango_transport.py index 89a0f0007a..e9ae18dc77 100644 --- a/src/ophyd_async/tango/signal/_tango_transport.py +++ b/src/ophyd_async/tango/signal/_tango_transport.py @@ -4,7 +4,6 @@ from abc import abstractmethod from asyncio import CancelledError from enum import Enum -from typing import Dict, Optional, Type, Union import numpy as np from bluesky.protocols import DataKey, Descriptor, Reading @@ -26,10 +25,11 @@ CmdArgType, CommandInfo, DevFailed, + DeviceProxy, DevState, EventType, ) -from tango.asyncio import DeviceProxy +from tango.asyncio import DeviceProxy as AsyncDeviceProxy from tango.asyncio_executor import ( AsyncioExecutor, get_global_executor, @@ -96,12 +96,12 @@ async def get_w_value(self) -> T: @abstractmethod async def put( - self, value: Optional[T], wait: bool = True, timeout: Optional[float] = None + self, value: T | None, wait: bool = True, timeout: float | None = None ) -> None: """Put value to TRL""" @abstractmethod - async def get_config(self) -> Union[AttributeInfoEx, CommandInfo]: + async def get_config(self) -> AttributeInfoEx | CommandInfo: """Get TRL config async""" @abstractmethod @@ -112,7 +112,7 @@ def has_subscription(self) -> bool: """indicates, that this trl already subscribed""" @abstractmethod - def subscribe_callback(self, callback: Optional[ReadingValueCallback]): + def subscribe_callback(self, callback: ReadingValueCallback | None): """subscribe tango CHANGE event to callback""" @abstractmethod @@ -164,7 +164,7 @@ async def get_w_value(self) -> T: @ensure_proper_executor async def put( - self, value: Optional[T], wait: bool = True, timeout: Optional[float] = None + self, value: T | None, wait: bool = True, timeout: float | None = None ) -> None or AsyncStatus: if wait: try: @@ -174,10 +174,12 @@ async def _write(): task = asyncio.create_task(_write()) await asyncio.wait_for(task, timeout) - except asyncio.TimeoutError: - raise TimeoutError(f"{self._name} attr put failed: Timeout") - except DevFailed as e: - raise RuntimeError(f"{self._name} device failure: {e.args[0].desc}") + except asyncio.TimeoutError as te: + raise TimeoutError(f"{self._name} attr put failed: Timeout") from te + except DevFailed as de: + raise RuntimeError( + f"{self._name} device" f" failure: {de.args[0].desc}" + ) from de else: rid = await self._proxy.write_attribute_asynch(self._name, value) @@ -194,11 +196,11 @@ async def wait_for_reply(rd, to): if to and time.time() - start_time > to: raise TimeoutError( f"{self._name} attr put failed:" f" Timeout" - ) + ) from exc else: raise RuntimeError( f"{self._name} device failure:" f" {exc.args[0].desc}" - ) + ) from exc return AsyncStatus(wait_for_reply(rid, timeout)) @@ -218,7 +220,7 @@ async def get_reading(self) -> Reading: def has_subscription(self) -> bool: return bool(self._callback) - def subscribe_callback(self, callback: Optional[ReadingValueCallback]): + def subscribe_callback(self, callback: ReadingValueCallback | None): # If the attribute supports events, then we can subscribe to them # If the callback is not a callable, then we raise an error if callback is not None and not callable(callback): @@ -293,11 +295,11 @@ async def poll(self): if self._callback is not None: self._callback(last_reading, last_reading["value"]) except Exception as e: - raise RuntimeError(f"Could not poll the attribute: {e}") + raise RuntimeError(f"Could not poll the attribute: {e}") from e try: # If the value is a number, we can check for changes - if isinstance(last_reading["value"], (int, float)): + if isinstance(last_reading["value"], int | float): while True: await asyncio.sleep(self._polling_period) reading = await self.get_reading() @@ -350,7 +352,7 @@ async def poll(self): break last_reading = reading.copy() except Exception as e: - raise RuntimeError(f"Could not poll the attribute: {e}") + raise RuntimeError(f"Could not poll the attribute: {e}") from e def set_polling( self, @@ -380,7 +382,7 @@ async def get_w_value(self) -> T: @ensure_proper_executor async def put( - self, value: Optional[T], wait: bool = True, timeout: Optional[float] = None + self, value: T | None, wait: bool = True, timeout: float | None = None ) -> None or AsyncStatus: if wait: try: @@ -395,10 +397,12 @@ async def _put(): "timestamp": time.time(), "alarm_severity": 0, } - except asyncio.TimeoutError: - raise TimeoutError(f"{self._name} command failed: Timeout") - except DevFailed as e: - raise RuntimeError(f"{self._name} device failure: {e.args[0].desc}") + except asyncio.TimeoutError as te: + raise TimeoutError(f"{self._name} command failed: Timeout") from te + except DevFailed as de: + raise RuntimeError( + f"{self._name} device" f" failure: {de.args[0].desc}" + ) from de else: rid = self._proxy.command_inout_asynch(self._name, value) @@ -415,17 +419,17 @@ async def wait_for_reply(rd, to): "alarm_severity": 0, } break - except DevFailed as e: - if e.args[0].reason == "API_AsynReplyNotArrived": + except DevFailed as de: + if de.args[0].reason == "API_AsynReplyNotArrived": await asyncio.sleep(A_BIT) if to and time.time() - start_time > to: raise TimeoutError( "Timeout while waiting for command reply" - ) + ) from de else: raise RuntimeError( - f"{self._name} device failure:" f" {e.args[0].desc}" - ) + f"{self._name} device failure:" f" {de.args[0].desc}" + ) from de return AsyncStatus(wait_for_reply(rid, timeout)) @@ -461,9 +465,9 @@ def get_dtype_extended(datatype): def get_trl_descriptor( - datatype: Optional[Type], + datatype: type | None, tango_resource: str, - tr_configs: Dict[str, Union[AttributeInfoEx, CommandInfo]], + tr_configs: dict[str, AttributeInfoEx | CommandInfo], ) -> dict: tr_dtype = {} for tr_name, config in tr_configs.items(): @@ -524,7 +528,7 @@ def get_trl_descriptor( trl_choices = list(DevState.names.keys()) if datatype: - if not issubclass(datatype, (Enum, DevState)): + if not issubclass(datatype, Enum | DevState): raise TypeError( f"{tango_resource} has type Enum not {datatype.__name__}" ) @@ -549,12 +553,10 @@ def get_trl_descriptor( return {"source": tango_resource, "dtype": tr_dtype_desc, "shape": []} -async def get_tango_trl( - full_trl: str, device_proxy: Optional[DeviceProxy] -) -> TangoProxy: +async def get_tango_trl(full_trl: str, device_proxy: DeviceProxy | None) -> TangoProxy: device_trl, trl_name = full_trl.rsplit("/", 1) trl_name = trl_name.lower() - device_proxy = device_proxy or await DeviceProxy(device_trl) + device_proxy = device_proxy or await AsyncDeviceProxy(device_trl) # all attributes can be always accessible with low register all_attrs = [attr_name.lower() for attr_name in device_proxy.get_attribute_list()] @@ -579,29 +581,29 @@ async def get_tango_trl( class TangoSignalBackend(SignalBackend[T]): def __init__( self, - datatype: Optional[Type[T]], - read_trl: Optional[str] = "", - write_trl: Optional[str] = "", - device_proxy: Optional[DeviceProxy] = None, + datatype: type[T] | None, + read_trl: str | None = "", + write_trl: str | None = "", + device_proxy: DeviceProxy | None = None, ): self.device_proxy = device_proxy self.datatype = datatype self.read_trl = read_trl self.write_trl = write_trl - self.proxies: Dict[str, TangoProxy] = { + self.proxies: dict[str, TangoProxy] = { read_trl: self.device_proxy, write_trl: self.device_proxy, } - self.trl_configs: Dict[str, AttributeInfoEx] = {} + self.trl_configs: dict[str, AttributeInfoEx] = {} self.descriptor: Descriptor = {} # type: ignore self._polling = (False, 0.1, None, 0.1) self.support_events = True self.status = None - def set_trl(self, read_trl: str, write_trl: Optional[str] = ""): + def set_trl(self, read_trl: str, write_trl: str | None = ""): self.read_trl = read_trl self.write_trl = write_trl if write_trl else read_trl - self.proxies: Dict[str, TangoProxy] = { + self.proxies: dict[str, TangoProxy] = { read_trl: self.device_proxy, write_trl: self.device_proxy, } @@ -615,8 +617,8 @@ async def _connect_and_store_config(self, trl): await self.proxies[trl].connect() self.trl_configs[trl] = await self.proxies[trl].get_config() self.proxies[trl].support_events = self.support_events - except CancelledError: - raise NotConnected(f"Could not connect to {trl}") + except CancelledError as ce: + raise NotConnected(f"Could not connect to {trl}") from ce async def connect(self, timeout: float = DEFAULT_TIMEOUT): if self.read_trl != self.write_trl: @@ -633,7 +635,7 @@ async def connect(self, timeout: float = DEFAULT_TIMEOUT): self.datatype, self.read_trl, self.trl_configs ) - async def put(self, value: Optional[T], wait=True, timeout=None): + async def put(self, value: T | None, wait=True, timeout=None): self.status = None put_status = await self.proxies[self.write_trl].put(value, wait, timeout) self.status = put_status @@ -650,7 +652,7 @@ async def get_value(self) -> T: async def get_setpoint(self) -> T: return await self.proxies[self.write_trl].get_w_value() - def set_callback(self, callback: Optional[ReadingValueCallback]) -> None: + def set_callback(self, callback: ReadingValueCallback | None) -> None: if self.support_events is False and self._polling[0] is False: raise RuntimeError( f"Cannot set event for {self.read_trl}. " @@ -662,10 +664,14 @@ def set_callback(self, callback: Optional[ReadingValueCallback]) -> None: try: assert not self.proxies[self.read_trl].has_subscription() self.proxies[self.read_trl].subscribe_callback(callback) - except AssertionError: - raise RuntimeError("Cannot set a callback when one is already set") + except AssertionError as ae: + raise RuntimeError( + "Cannot set a callback when one" " is already set" + ) from ae except RuntimeError as exc: - raise RuntimeError(f"Cannot set callback for {self.read_trl}. {exc}") + raise RuntimeError( + f"Cannot set callback" f" for {self.read_trl}. {exc}" + ) from exc else: self.proxies[self.read_trl].unsubscribe_callback() @@ -679,5 +685,5 @@ def set_polling( ): self._polling = (allow_polling, polling_period, abs_change, rel_change) - def allow_events(self, allow: Optional[bool] = True): + def allow_events(self, allow: bool | None = True): self.support_events = allow diff --git a/tests/tango/test_base_device.py b/tests/tango/test_base_device.py index f2cc2cc168..0d78e68e22 100644 --- a/tests/tango/test_base_device.py +++ b/tests/tango/test_base_device.py @@ -1,7 +1,6 @@ import asyncio import time from enum import Enum, IntEnum -from typing import Optional, Type, Union import bluesky.plan_stubs as bps import bluesky.plans as bp @@ -182,8 +181,8 @@ class TestTangoReadable(TangoReadable): def __init__( self, - trl: Optional[str] = None, - device_proxy: Optional[Union[AsyncDeviceProxy, SyncDeviceProxy]] = None, + trl: str | None = None, + device_proxy: SyncDeviceProxy | None = None, name: str = "", ) -> None: super().__init__(trl, device_proxy, name=name) @@ -248,7 +247,7 @@ async def describe_class(fqtrl): # -------------------------------------------------------------------- -def get_test_descriptor(python_type: Type[T], value: T, is_cmd: bool) -> dict: +def get_test_descriptor(python_type: type[T], value: T, is_cmd: bool) -> dict: if python_type in [bool, int]: return {"dtype": "integer", "shape": []} if python_type in [float]: @@ -325,7 +324,7 @@ async def test_connect(tango_test_device): # -------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.parametrize("proxy", [True, False, None]) -async def test_connect_proxy(tango_test_device, proxy: Optional[bool]): +async def test_connect_proxy(tango_test_device, proxy: bool | None): if proxy is None: test_device = TestTangoReadable(trl=tango_test_device) test_device.proxy = None diff --git a/tests/tango/test_tango_signals.py b/tests/tango/test_tango_signals.py index 136600b6d9..bfc980cc64 100644 --- a/tests/tango/test_tango_signals.py +++ b/tests/tango/test_tango_signals.py @@ -3,7 +3,7 @@ import time from enum import Enum, IntEnum from random import choice -from typing import Any, Optional, Tuple, Type +from typing import Any import numpy as np import numpy.typing as npt @@ -189,7 +189,9 @@ def echo_command(self, arg): # -------------------------------------------------------------------- def assert_enum(initial_value, readout_value): if type(readout_value) in [list, tuple]: - for _initial_value, _readout_value in zip(initial_value, readout_value): + for _initial_value, _readout_value in zip( + initial_value, readout_value, strict=False + ): assert_enum(_initial_value, _readout_value) else: assert initial_value == readout_value @@ -215,7 +217,7 @@ def reset_tango_asyncio(): # -------------------------------------------------------------------- # helpers to run tests # -------------------------------------------------------------------- -def get_test_descriptor(python_type: Type[T], value: T, is_cmd: bool) -> dict: +def get_test_descriptor(python_type: type[T], value: T, is_cmd: bool) -> dict: if python_type in [bool, int]: return {"dtype": "integer", "shape": []} if python_type in [float]: @@ -239,10 +241,10 @@ def get_test_descriptor(python_type: Type[T], value: T, is_cmd: bool) -> dict: # -------------------------------------------------------------------- async def make_backend( - typ: Optional[Type], + typ: type | None, pv: str, connect: bool = True, - allow_events: Optional[bool] = True, + allow_events: bool | None = True, ) -> TangoSignalBackend: backend = TangoSignalBackend(typ, pv, pv) backend.allow_events(allow_events) @@ -260,7 +262,7 @@ async def prepare_device(echo_device: str, pv: str, put_value: T) -> None: # -------------------------------------------------------------------- class MonitorQueue: def __init__(self, backend: SignalBackend): - self.updates: asyncio.Queue[Tuple[Reading, Any]] = asyncio.Queue() + self.updates: asyncio.Queue[tuple[Reading, Any]] = asyncio.Queue() self.backend = backend self.subscription = backend.set_callback(self.add_reading_value) @@ -301,7 +303,7 @@ async def assert_monitor_then_put( initial_value: T, put_value: T, descriptor: dict, - datatype: Optional[Type[T]] = None, + datatype: type[T] | None = None, ): await prepare_device(echo_device, pv, initial_value) source = echo_device + "/" + pv @@ -332,7 +334,7 @@ async def test_backend_get_put_monitor_attr( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, ): @@ -361,7 +363,7 @@ async def assert_put_read( pv: str, put_value: T, descriptor: dict, - datatype: Optional[Type[T]] = None, + datatype: type[T] | None = None, ): source = echo_device + "/" + pv backend = await make_backend(datatype, source) @@ -394,7 +396,7 @@ async def test_backend_get_put_monitor_cmd( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, ): @@ -439,7 +441,7 @@ async def test_tango_signal_r( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, use_proxy: bool, @@ -492,7 +494,7 @@ async def test_tango_signal_w( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, use_proxy: bool, @@ -558,7 +560,7 @@ async def test_tango_signal_rw( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, use_proxy: bool, @@ -634,7 +636,7 @@ async def test_tango_signal_auto_attrs( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, use_proxy: bool, @@ -709,7 +711,7 @@ async def test_tango_signal_auto_cmds( pv: str, tango_type: str, d_format: AttrDataFormat, - py_type: Type[T], + py_type: type[T], initial_value: T, put_value: T, use_dtype: bool,