diff --git a/README.md b/README.md index 016a5e5..57dc66a 100644 --- a/README.md +++ b/README.md @@ -69,7 +69,27 @@ assert registry.find("my_device") is the_device ``` This greedy behavior can be suppressed with the *auto_register* -parameter. If *auto_register* is false, then a device class can be +parameter. It can also be turned off after initialization by setting +the *auto_register* attribute to ``False``:: + +```python + +registry = Registry(auto_register=False) + +# Make a bunch of devices +... + +# Turn if off for this one +registry.auto_register = False +device = MyDevice(...) +registry.auto_register = True + +# Register some other devices maybe +... + +``` + +If *auto_register* is false, then a device class can be decorated to allow the registry to find it: ```python diff --git a/pyproject.toml b/pyproject.toml index a35a671..9861a89 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,7 +19,7 @@ dependencies = ["ophyd"] [project.optional-dependencies] -dev = ["black", "isort", "pytest", "build", "twine", "flake8", "caproto", "ruff"] +dev = ["black", "isort", "pytest", "build", "twine", "flake8", "ruff"] [project.urls] "Homepage" = "https://github.com/spc-group/ophyd-registry" diff --git a/src/ophydregistry/registry.py b/src/ophydregistry/registry.py index 5b7239e..4ca7576 100644 --- a/src/ophydregistry/registry.py +++ b/src/ophydregistry/registry.py @@ -107,6 +107,7 @@ class Registry: """ use_typhos: bool = False + _auto_register: bool # components: Sequence _objects_by_name: Mapping @@ -119,11 +120,28 @@ def __init__(self, auto_register: bool = True, use_typhos: bool = False): # Set up empty lists and things for registering components self.clear() self.use_typhos = use_typhos - # Add a callback to get notified of new objects - if auto_register: + self.auto_register = auto_register + + @property + def auto_register(self): + return self._auto_register + + @auto_register.setter + def auto_register(self, val): + """Turn on or off the automatic registration of new devices.""" + self._auto_register = val + if val: + # Add a callback to get notified of new objects ophydobj.OphydObject.add_instantiation_callback( self.register, fail_if_late=False ) + else: + try: + ophydobj.OphydObject._OphydObject__instantiation_callbacks.remove( + self.register + ) + except ValueError: + pass def __getitem__(self, key): """Retrieve the object from the dicionary. @@ -454,11 +472,14 @@ def register(self, component: ophydobj.OphydObject) -> ophydobj.OphydObject: # (Needed for some sub-components that are just readback # values of the parent) # Check that we're not adding a duplicate component name + if name == "I0_center": + print(self._objects_by_name.keys(), self) if name in self._objects_by_name.keys(): old_obj = self._objects_by_name[name] is_readback = component in [ getattr(old_obj, "readback", None), getattr(old_obj, "user_readback", None), + getattr(old_obj, "val", None), ] if is_readback: msg = f"Ignoring readback with duplicate name: '{name}'" diff --git a/src/ophydregistry/tests/test_instrument_registry.py b/src/ophydregistry/tests/test_instrument_registry.py index e8feae6..86b2599 100644 --- a/src/ophydregistry/tests/test_instrument_registry.py +++ b/src/ophydregistry/tests/test_instrument_registry.py @@ -1,3 +1,4 @@ +import gc import logging import pytest @@ -8,13 +9,14 @@ @pytest.fixture() def registry(): - reg = Registry(auto_register=False) - return reg + reg = Registry(auto_register=False, use_typhos=False) + try: + yield reg + finally: + del reg def test_register_component(registry): - # Prepare registry - registry = Registry(auto_register=False) # Create an unregistered component cpt = sim.SynGauss( "I0", @@ -40,9 +42,8 @@ def test_register_component(registry): assert cpt in results -def test_find_missing_components(): +def test_find_missing_components(registry): """Test that registry raises an exception if no matches are found.""" - reg = Registry() cpt = sim.SynGauss( "I0", sim.motor, @@ -52,25 +53,23 @@ def test_find_missing_components(): sigma=1, labels={"ion_chamber"}, ) - reg.register(cpt) + registry.register(cpt) # Now make sure a different query still returns no results with pytest.raises(ComponentNotFound): - reg.findall(label="spam") + registry.findall(label="spam") -def test_find_allow_missing_components(): +def test_find_allow_missing_components(registry): """Test that registry tolerates missing components with the *allow_none* argument. """ - reg = Registry() # Get some non-existent devices and check that the right nothing is returned - assert list(reg.findall(label="spam", allow_none=True)) == [] - assert reg.find(name="eggs", allow_none=True) is None + assert list(registry.findall(label="spam", allow_none=True)) == [] + assert registry.find(name="eggs", allow_none=True) is None def test_exceptions(registry): - registry = Registry() registry.register(Device("", name="It")) # Test if a non-existent labels throws an exception with pytest.raises(ComponentNotFound): @@ -137,9 +136,7 @@ def test_find_name_by_dot_notation(registry): assert result is cptA.val -def test_find_labels_by_dot_notation(): - # Prepare registry - reg = Registry() +def test_find_labels_by_dot_notation(registry): # Create a simulated component cptA = sim.SynGauss( "I0", @@ -150,15 +147,13 @@ def test_find_labels_by_dot_notation(): sigma=1, labels={"ion_chamber"}, ) - reg.register(cptA) + registry.register(cptA) # Only one match should work fine - result = reg.find(label="ion_chamber.val") + result = registry.find(label="ion_chamber.val") assert result is cptA.val -def test_find_any(): - # Prepare registry - reg = Registry() +def test_find_any(registry): # Create an unregistered component cptA = sim.SynGauss( "I0", @@ -196,25 +191,23 @@ def test_find_any(): labels={}, ) # Register the components - reg.register(cptA) - reg.register(cptB) - reg.register(cptC) - reg.register(cptD) + registry.register(cptA) + registry.register(cptB) + registry.register(cptC) + registry.register(cptD) # Only one match should work fine - result = reg.findall(any_of="ion_chamber") + result = registry.findall(any_of="ion_chamber") assert cptA in result assert cptB in result assert cptC in result assert cptD not in result -def test_find_by_device(): +def test_find_by_device(registry): """The registry should just return the device itself if that's what is passed.""" - # Prepare registry - reg = Registry() # Register a component cptD = sim.SynGauss( - "sample motor", + "sample_motor", sim.motor, "motor", center=-0.5, @@ -222,16 +215,14 @@ def test_find_by_device(): sigma=1, labels={}, ) - reg.register(cptD) + registry.register(cptD) # Pass the device itself to the find method - result = reg.find(cptD) + result = registry.find(cptD) assert result is cptD -def test_find_by_list_of_names(): +def test_find_by_list_of_names(registry): """Will the findall() method handle lists of things to look up.""" - # Prepare registry - reg = Registry() # Register a component cptA = sim.SynGauss( "sample motor A", @@ -260,11 +251,11 @@ def test_find_by_list_of_names(): sigma=1, labels={}, ) - reg.register(cptA) - reg.register(cptB) - reg.register(cptC) + registry.register(cptA) + registry.register(cptB) + registry.register(cptC) # Pass the device names into the findall method - result = reg.findall(["sample motor A", "sample motor B"]) + result = registry.findall(["sample motor A", "sample motor B"]) assert cptA in result assert cptB in result assert cptC not in result @@ -272,7 +263,9 @@ def test_find_by_list_of_names(): def test_user_readback(registry): """Edge case where EpicsMotor.user_readback is named the same as the motor itself.""" - device = EpicsMotor("255idVME:m1", name="epics_motor") + device = sim.instantiate_fake_device( + EpicsMotor, prefix="255idVME:m1", name="epics_motor" + ) registry.register(device) # See if requesting the device.user_readback returns the proper signal registry.find("epics_motor_user_readback") @@ -284,7 +277,8 @@ def test_auto_register(): Uses ophyds instantiation callback mechanism. """ - registry = Registry() + registry = Registry(auto_register=True) + print(f"auto_register: {registry}") sim.SynGauss( "I0", sim.motor, @@ -295,11 +289,12 @@ def test_auto_register(): labels={"ion_chamber"}, ) registry.find("I0") + # Delete this registry to avoid warnings about duplicate entries + registry.auto_register = False -def test_clear(): +def test_clear(registry): """Can the registry be properly cleared.""" - registry = Registry() cpt = sim.SynGauss( "I0", sim.motor, @@ -309,6 +304,7 @@ def test_clear(): sigma=1, labels={"ion_chamber"}, ) + registry.register(cpt) assert registry.find("I0") is cpt # Clear the registry and confirm that it's gone registry.clear() @@ -316,10 +312,10 @@ def test_clear(): registry.find("I0") -def test_component_properties(): +def test_component_properties(registry): """Check that we can get lists of component and devices.""" - registry = Registry() - sim.SynGauss( + print(f"component_properties: {registry}") + I0 = sim.SynGauss( "I0", sim.motor, "motor", @@ -328,6 +324,7 @@ def test_component_properties(): sigma=1, labels={"ion_chamber"}, ) + registry.register(I0) assert registry.device_names == {"I0"} assert registry.component_names == { "I0", @@ -336,12 +333,10 @@ def test_component_properties(): "I0_noise", "I0_noise_multiplier", "I0_sigma", - "I0_val", } -def test_root_devices(): - registry = Registry() +def test_root_devices(registry): registry.register(sim.motor1) registry.register(sim.motor2) registry.register(sim.motor3) @@ -350,18 +345,16 @@ def test_root_devices(): assert len(root_devices) == 3 -def test_getitem(): +def test_getitem(registry): """Check that the registry can be accessed like a dictionary.""" - registry = Registry() registry.register(sim.motor1) # Check that the dictionary access works result = registry["motor1"] assert result is sim.motor1 -def test_duplicate_device(caplog): +def test_duplicate_device(caplog, registry): """Check that a device doesn't get added twice.""" - registry = Registry() motor = sim.motor # Set up logging so that we can know what caplog.clear() @@ -379,10 +372,9 @@ def test_duplicate_device(caplog): assert "Ignoring component with duplicate name" in caplog.text -def test_delete_by_name(): +def test_delete_by_name(registry): """Check that we can remove an item from the ophyd registry.""" # Add an item to the registry - registry = Registry() motor = sim.motor registry.register(motor) # Delete the item from the registry @@ -392,10 +384,9 @@ def test_delete_by_name(): registry[motor.name] -def test_pop_by_name(): +def test_pop_by_name(registry): """Check that we can remove an item from the ophyd registry.""" # Add an item to the registry - registry = Registry() motor = sim.motor registry.register(motor) # Pop the item from the registry @@ -411,10 +402,9 @@ def test_pop_by_name(): registry[motor.acceleration.name] -def test_pop_by_object(): +def test_pop_by_object(registry): """Check that we can remove an item from the ophyd registry.""" # Add an item to the registry - registry = Registry() motor = sim.motor registry.register(motor) # Pop the item from the registry @@ -427,10 +417,9 @@ def test_pop_by_object(): registry["motors"] -def test_pop_default(): +def test_pop_default(registry): """Check that we get a default object if our key is not present.""" # Add an item to the registry - registry = Registry() motor = sim.motor # Pop the item from the registry popped = registry.pop("gibberish", motor)