diff --git a/ophyd_devices/__init__.py b/ophyd_devices/__init__.py index 729f684..a4f6207 100644 --- a/ophyd_devices/__init__.py +++ b/ophyd_devices/__init__.py @@ -1,7 +1,3 @@ -from .ophyd_patch import monkey_patch_ophyd - -monkey_patch_ophyd() - from .devices.sls_devices import SLSInfo, SLSOperatorMessages from .sim.sim_camera import SimCamera from .sim.sim_monitor import SimMonitor, SimMonitorAsync diff --git a/ophyd_devices/ophyd_patch.py b/ophyd_devices/ophyd_patch.py deleted file mode 100644 index 783e0d1..0000000 --- a/ophyd_devices/ophyd_patch.py +++ /dev/null @@ -1,91 +0,0 @@ -import importlib -import importlib.util -import pathlib -import sys -from importlib.abc import Loader, MetaPathFinder -from types import ModuleType - -_patched_status_base = """ -import threading -from unittest.mock import Mock, patch - -_StatusBase = StatusBase - -class StatusBase(_StatusBase): - _bec_patched = True - - def __init__(self, *args, **kwargs): - timeout = kwargs.get("timeout", None) - if not timeout: - with patch("threading.Thread", Mock(spec=threading.Thread)): - super().__init__(*args, **kwargs) - else: - super().__init__(*args, **kwargs) - - def set_finished(self, *args, **kwargs): - super().set_finished(*args, **kwargs) - if isinstance(self._callback_thread, Mock): - if self.settle_time > 0: - - def settle_done(): - self._settled_event.set() - self._run_callbacks() - - threading.Timer(self.settle_time, settle_done).start() - else: - self._run_callbacks() - - def set_exception(self, *args, **kwargs): - super().set_exception(*args, **kwargs) - if isinstance(self._callback_thread, Mock): - self._run_callbacks() - -""" - - -class _CustomLoader(Loader): - def __init__(self, patched_code): - self.patched_code = patched_code - - def load_module(self, fullname): - """Load and execute ophyd.status""" - status_module = ModuleType("ophyd.status") - status_module.__loader__ = self - status_module.__file__ = None - status_module.__name__ = fullname - - exec(self.patched_code, status_module.__dict__) - sys.modules[fullname] = status_module - - return status_module, True - - -class _CustomImporter(MetaPathFinder): - def __init__(self): - origin = pathlib.Path(importlib.util.find_spec("ophyd").origin) - module_file = str(origin.parent / "status.py") - - with open(module_file, "r") as source: - src = source.read() - before, _, after = src.partition("class StatusBase") - orig_status_base, _, final = after.partition("\nclass ") - - self.patched_source = ( - f"{before}class StatusBase{orig_status_base}{_patched_status_base}class {final}" - ) - self.patched_code = compile(self.patched_source, module_file, "exec") - self.loader = _CustomLoader(self.patched_code) - - def find_spec(self, fullname, path, target=None): - # The new import classes are difficult to grasp; - # why the finder needs a .loader member, it could be returned - # from here. And also .name, which name has to correspond to - # the searched module ??? - if fullname == "ophyd.status": - self.name = fullname - return self - return None - - -def monkey_patch_ophyd(): - sys.meta_path.insert(0, _CustomImporter()) diff --git a/pyproject.toml b/pyproject.toml index a30eff5..2a77551 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ classifiers = [ ] dependencies = [ - "ophyd ~= 1.9", + "ophyd ~= 1.10", "typeguard ~= 4.1, >=4.1.5", "prettytable ~= 3.9", "bec_lib >=2.12.2, <=4.0", diff --git a/tests/test_ophyd_status_obj.py b/tests/test_ophyd_status_obj.py index e917c92..2639302 100644 --- a/tests/test_ophyd_status_obj.py +++ b/tests/test_ophyd_status_obj.py @@ -6,11 +6,10 @@ import pytest from ophyd.status import DeviceStatus, StatusBase, StatusTimeoutError -def test_ophyd_status_patch(): - assert StatusBase._bec_patched - - st = DeviceStatus(device="test") - assert st._bec_patched +def test_ophyd_status(): + device = Mock() + device.name.return_value = "test" + st = DeviceStatus(device) assert isinstance(st, StatusBase) cb = Mock() @@ -25,7 +24,7 @@ def test_ophyd_status_patch(): cb.reset_mock() st = StatusBase() - assert isinstance(st._callback_thread, Mock) + assert st._callback_thread is None st.add_callback(cb) st.set_finished() cb.assert_called_once() @@ -34,7 +33,7 @@ def test_ophyd_status_patch(): st = StatusBase(settle_time=1) st.add_callback(cb) - assert isinstance(st._callback_thread, Mock) + assert st._callback_thread is None st.set_finished() assert cb.call_count == 0 time.sleep(0.5)