Move util functions from i22 branch of ophyd_async that may be generally useful #10
Conversation
Codecov Report
@@ Coverage Diff @@
## main #10 +/- ##
==========================================
+ Coverage 87.17% 91.47% +4.29%
==========================================
Files 9 13 +4
Lines 78 129 +51
==========================================
+ Hits 68 118 +50
- Misses 10 11 +1
📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
|
May want to turn off test coverage for the plans and stubs module. |
DiamondJoseph
force-pushed
the
i22_functions
branch
from
5659bc4
to
ced4b8f
Compare
- get_constant_duration renamed to be clearer intended function, always return None instead of throwing - in_micros to throw exception when working in negative time - Add tests for intended behaviour of get_constant_duration
Co-authored-by: Tom C (DLS) <101418278+coretl@users.noreply.github.com>
|
Not to be merged until bluesky/ophyd-async#24 has been merged and ophyd_async pin is removed. |
| @pytest.fixture(scope="function") | ||
| def RE(request): | ||
| loop = asyncio.new_event_loop() | ||
| loop.set_debug(True) | ||
| RE = RunEngine({}, call_returns_result=True, loop=loop) | ||
|
|
||
| def clean_event_loop(): | ||
| if RE.state not in ("idle", "panicked"): | ||
| try: | ||
| RE.halt() | ||
| except TransitionError: | ||
| pass | ||
| loop.call_soon_threadsafe(loop.stop) | ||
| RE._th.join() | ||
| loop.close() | ||
|
|
||
| request.addfinalizer(clean_event_loop) | ||
| return RE |
There was a problem hiding this comment.
A general comment for future reference; I've started peppering this into every repo that needs to run plans with ophyd-async devices, such as dodal, and ophyd-async. And now here! Seems like this bit of code really ought to belong somewhere more common but I'm not sure where. This is just a general comment though. Happy for this to be a 'we will sort it after I22' type deal.
There was a problem hiding this comment.
Is it possible to have fixtures available from the library (ophyd_async) to libraries (this, dodal) and applications (BlueAPI?) that use it?
Does ophyd_async depend on bluesky for things other than the RunEngine?
There was a problem hiding this comment.
Ophyd-async mainly depends on bluesky for its protocols (as well as the run engine).
For now I'm thinking the best thing is to have a testutils module in bluesky which contains this, i.e.
bluesky/bluesky/utils.py::
def RE(request):
loop = asyncio.new_event_loop()
loop.set_debug(True)
RE = RunEngine({}, call_returns_result=True, loop=loop)
def clean_event_loop():
if RE.state not in ("idle", "panicked"):
try:
RE.halt()
except TransitionError:
pass
loop.call_soon_threadsafe(loop.stop)
RE._th.join()
loop.close()
request.addfinalizer(clean_event_loop)
return REAnd then here,
from bluesky.utils import RE
@pytest.fixture(scope="function")
def RunEngine(request):
return RE(request)but idk if RE is a good name for it...
| @pytest.fixture(scope="function") | ||
| def RE(request): | ||
| loop = asyncio.new_event_loop() | ||
| loop.set_debug(True) | ||
| RE = RunEngine({}, call_returns_result=True, loop=loop) | ||
|
|
||
| def clean_event_loop(): | ||
| if RE.state not in ("idle", "panicked"): | ||
| try: | ||
| RE.halt() | ||
| except TransitionError: | ||
| pass | ||
| loop.call_soon_threadsafe(loop.stop) | ||
| RE._th.join() | ||
| loop.close() | ||
|
|
||
| request.addfinalizer(clean_event_loop) | ||
| return RE |
No description provided.