wait_for_value_interface_change #582
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
ZohebShaikh
force-pushed
the
461-rationalise-interface-for-wait_for_value
branch
from
3751b80
to
caadaca
Compare
coretl
reviewed
Sep 19, 2024
coretl
reviewed
Sep 19, 2024
coretl
reviewed
Sep 26, 2024
…eter for waiting for set completion
…face-for-wait_for_value
coretl
reviewed
Oct 18, 2024
coretl
reviewed
Oct 22, 2024
coretl
reviewed
Oct 24, 2024
coretl
approved these changes
Oct 24, 2024
* Add test for context race condition * Add timeout to wait on connection in test
Rewrite the type support, and use a plugin `Connector` architecture to support reconnection for PVI and Tango. Fixes #472, fixes #562, fixes #535, fixes #505, fixes #373, fixes #601 Required for #551 Structure now read from `.value` rather than `.pvi`. Supported in FastCS. Requires at least PandABlocks-ioc 0.10.0 ```python from enum import Enum class MyEnum(str, Enum): ONE = "one" TWO = "two" from ophyd_async.core import StrictEnum class MyEnum(StrictEnum): ONE = "one" TWO = "two" ``` ```python from ophyd_async.core import SubsetEnum MySubsetEnum = SubsetEnum["one", "two"] class MySubsetEnum(SubsetEnum): ONE = "one" TWO = "two" ``` ```python import numpy as np x = epics_signal_rw(np.int32, "PV") x = epics_signal_rw(int, "PV") ``` ```python import numpy as np import numpy.typing as npt x = epics_signal_rw(npt.NDArray[np.int32], "PV") from ophyd_async.core import Array1D x = epics_signal_rw(Array1D[np.int32], "PV") ``` ```python import numpy as np import numpy.typing as npt x = epics_signal_rw(npt.NDArray[np.str_], "PV") from collections.abc import Sequence x = epics_signal_rw(Sequence[str], "PV") ``` ```python fake_set_signal = SignalRW(MockSignalBackend(float)) fake_set_signal = soft_signal_rw(float) await fake_set_signal.connect(mock=True) ``` ```python get_mock_put(driver.capture).assert_called_once_with(Writing.ON, wait=ANY, timeout=ANY) get_mock_put(driver.capture).assert_called_once_with(Writing.ON, wait=ANY) ``` ```python class MyDevice(Device): def __init__(self, name: str = ""): self.signal, self.backend_put = soft_signal_r_and_setter(int) class MyDevice(Device): def __init__(self, name: str = ""): self.signal, self.backend_put = soft_signal_r_and_setter(int) super().__init__(name=name) ``` The `Table` type has been suitable for everything we have seen so far, if you need an arbitrary `BaseModel` subclass then please make an issue ```python class SourceDevice(Device): def __init__(self, name: str = ""): self.signal = soft_signal_rw(int) super().__init__(name=name) class ReferenceDevice(Device): def __init__(self, signal: SignalRW[int], name: str = ""): self.signal = signal super().__init__(name=name) def set(self, value) -> AsyncStatus: return self.signal.set(value + 1) from ophyd_async.core import Reference class ReferenceDevice(Device): def __init__(self, signal: SignalRW[int], name: str = ""): self._signal_ref = Reference(signal) super().__init__(name=name) def set(self, value) -> AsyncStatus: return self._signal_ref().set(value + 1) ```
* Update copier template to 2.4.0 * Choose a machine accessible URL for license
* Allow unittest.Mock to be passed to Device.connect * use bool | Mock signature for connect everywhere * store Mocks for Devices and MockSignalBackends for Signals in dictionaries * assert mock calls explicitly in epics/demo/test_demo.py
* Fix race condition in mock_signal_backend * Fix numpy datatypes in test_mock_signal_backend * Fix adaravis URI paths * Fix path in test_kinetix * Fix paths in simdetector tests * Loosen timings against race condition in test_sim_detector * Correct paths in test_advimba * Fix paths in test_eiger * Fix broken tests in test_signals * Fix tange test_base_device tests on Windows Missing process=True argument was causing all subsequent tests to fail on windows * Fix race conditions in tango tests * Use tango FQTRL to placate windows unit tests * Correct paths in fastcs tests * Correct path logic for windows compatibility * Slacken timings for race condition in test_sim_motor * Slacken race condition timings in test_motor * Enable windows CI builds * Try allowing asyncio time to clean up tasks * Object IDs on windows are uppercase * Try wait_for_wakeups
Add optional Declarative Device support
Includes:
- An ADR for optional Declarative Devices
- Support for `StandardReadable` Declarative Devices via `StandardReadableFormat` annotations
- Support for EPICS Declarative Devices via an `EpicsDevice` baseclass and `PvSuffic` annotations
- Updates to the EPICS and Tango demo devices to use them
Structure read from `.value` now includes `DeviceVector` support. Requires at least PandABlocks-ioc 0.11.2
`ophyd_async.epics.signal` moves to `ophyd_async.epics.core` with a backwards compat module that emits deprecation warning.
```python
from ophyd_async.epics.signal import epics_signal_rw
from ophyd_async.epics.core import epics_signal_rw
```
`StandardReadable` wrappers change to enum members of `StandardReadableFormat` (normally imported as `Format`)
```python
from ophyd_async.core import ConfigSignal, HintedSignal
class MyDevice(StandardReadable):
def __init__(self):
self.add_readables([sig1], ConfigSignal)
self.add_readables([sig2], HintedSignal)
self.add_readables([sig3], HintedSignal.uncached)
from ophyd_async.core import StandardReadableFormat as Format
class MyDevice(StandardReadable):
def __init__(self):
self.add_readables([sig1], Format.CONFIG_SIGNAL)
self.add_readables([sig2], Format.HINTED_SIGNAL)
self.add_readables([sig3], Format.HINTED_UNCACHED_SIGNAL
```
```python
from ophyd_async.core import ConfigSignal, HintedSignal
from ophyd_async.epics.signal import epics_signal_r, epics_signal_rw
class Sensor(StandardReadable):
def __init__(self, prefix: str, name="") -> None:
with self.add_children_as_readables(HintedSignal):
self.value = epics_signal_r(float, prefix + "Value")
with self.add_children_as_readables(ConfigSignal):
self.mode = epics_signal_rw(EnergyMode, prefix + "Mode")
super().__init__(name=name)
from typing import Annotated as A
from ophyd_async.core import StandardReadableFormat as Format
from ophyd_async.epics.core import EpicsDevice, PvSuffix, epics_signal_r, epics_signal_rw
class Sensor(StandardReadable, EpicsDevice):
value: A[SignalR[float], PvSuffix("Value"), Format.HINTED_SIGNAL]
mode: A[SignalRW[EnergyMode], PvSuffix("Mode"), Format.CONFIG_SIGNAL]
```
- Raise `NotConnected` errors in mock mode too when failing on creation of mock signal because of invalid datatype - Support bare `np.ndarray` in SoftSignalBackend - Support tagged and variant unions for NTNDArray - Check that Devices have unique names in `ensure_connected`
This helps in the very specific case of an observe_value directly or indirectly modifying the signal that is being updated. This creates a busy loop which will not be interrupted by wrapping in asyncio.wait_for. To demonstrate, added test_observe_value_times_out_with_no_external_task
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
closes #461