Remove assert in production code

pyproject.toml

@@ -156,6 +156,7 @@ lint.select = [
     "SLF",     # self - https://docs.astral.sh/ruff/settings/#lintflake8-self
     "PLC2701", # private import - https://docs.astral.sh/ruff/rules/import-private-name/
     "LOG015",  # root logger call - https://docs.astral.sh/ruff/rules/root-logger-call/
+    "S101",    # assert = https://docs.astral.sh/ruff/rules/assert/
 ]
 lint.ignore = [
     "B901", # Return in a generator is needed for plans
@@ -166,7 +167,10 @@ lint.preview = true # so that preview mode PLC2701 is enabled
 # By default, private member access is allowed in tests
 # See https://github.com/DiamondLightSource/python-copier-template/issues/154
 # Remove this line to forbid private member access in tests
-"tests/**/*" = ["SLF001"]
+"tests/**/*" = ["SLF001", "S101"]
+"src/ophyd_async/testing/**/*" = ["SLF001", "S101"]
+"system_tests/**/*" = ["SLF001", "S101"]
+"docs/examples/**/*" = ["SLF001", "S101"]
 
 
 [tool.importlinter]

src/ophyd_async/core/_detector.py

@@ -173,6 +173,13 @@ def hints(self) -> Hints:
 DetectorWriterT = TypeVar("DetectorWriterT", bound=DetectorWriter)
 
 
+def _ensure_trigger_info_exists(trigger_info: TriggerInfo | None) -> TriggerInfo:
+    # make absolute sure we realy have a valid TriggerInfo ... mostly for pylance
+    if trigger_info is None:
+        raise RuntimeError("Trigger info must be set before calling this method.")
+    return trigger_info
+
+
 class StandardDetector(
     Device,
     Stageable,
@@ -279,8 +286,12 @@ async def trigger(self) -> None:
                     frame_timeout=None,
                 )
             )
-        assert self._trigger_info
-        assert self._trigger_info.trigger is DetectorTrigger.INTERNAL
+
+        self._trigger_info = _ensure_trigger_info_exists(self._trigger_info)
+        if self._trigger_info.trigger is not DetectorTrigger.INTERNAL:
+            msg = "The trigger method can only be called with INTERNAL triggering"
+            raise ValueError(msg)
+
         # Arm the detector and wait for it to finish.
         indices_written = await self._writer.get_indices_written()
         await self._controller.arm()
@@ -311,16 +322,17 @@ async def prepare(self, value: TriggerInfo) -> None:
         Args:
             value: TriggerInfo describing how to trigger the detector
         """
-        if value.trigger != DetectorTrigger.INTERNAL:
-            assert value.deadtime, (
-                "Deadtime must be supplied when in externally triggered mode"
-            )
-        if value.deadtime:
-            required = self._controller.get_deadtime(value.livetime)
-            assert required <= value.deadtime, (
-                f"Detector {self._controller} needs at least {required}s deadtime, "
-                f"but trigger logic provides only {value.deadtime}s"
+        if value.trigger != DetectorTrigger.INTERNAL and not value.deadtime:
+            msg = "Deadtime must be supplied when in externally triggered mode"
+            raise ValueError(msg)
+        required_deadtime = self._controller.get_deadtime(value.livetime)
+        if value.deadtime and required_deadtime > value.deadtime:
+            msg = (
+                f"Detector {self._controller} needs at least {required_deadtime}s "
+                f"deadtime, but trigger logic provides only {value.deadtime}s"
             )
+            raise ValueError(msg)
+
         self._trigger_info = value
         self._number_of_triggers_iter = iter(
             self._trigger_info.number_of_triggers
@@ -350,7 +362,7 @@ async def kickoff(self):
 
     @WatchableAsyncStatus.wrap
     async def complete(self):
-        assert self._trigger_info
+        self._trigger_info = _ensure_trigger_info_exists(self._trigger_info)
         indices_written = self._writer.observe_indices_written(
             self._trigger_info.frame_timeout
             or (

src/ophyd_async/core/_device.py

@@ -158,10 +158,12 @@ async def connect(
         timeout:
             Time to wait before failing with a TimeoutError.
         """
-        assert hasattr(self, "_connector"), (
-            f"{self}: doesn't have attribute `_connector`,"
-            " did you call `super().__init__` in your `__init__` method?"
-        )
+        if not hasattr(self, "_connector"):
+            msg = (
+                f"{self}: doesn't have attribute `_connector`,"
+                " did you call `super().__init__` in your `__init__` method?"
+            )
+            raise RuntimeError(msg)
         if mock:
             # Always connect in mock mode serially
             if isinstance(mock, LazyMock):
@@ -182,7 +184,9 @@ async def connect(
                 self._mock = None
                 coro = self._connector.connect_real(self, timeout, force_reconnect)
                 self._connect_task = asyncio.create_task(coro)
-            assert self._connect_task, "Connect task not created, this shouldn't happen"
+            if not self._connect_task:
+                msg = "Connect task not created, this shouldn't happen"
+                raise RuntimeError(msg)
             # Wait for it to complete
             await self._connect_task
 
@@ -232,8 +236,12 @@ def __getitem__(self, key: int) -> DeviceT:
     def __setitem__(self, key: int, value: DeviceT) -> None:
         # Check the types on entry to dict to make sure we can't accidentally
         # make a non-integer named child
-        assert isinstance(key, int), f"Expected int, got {key}"
-        assert isinstance(value, Device), f"Expected Device, got {value}"
+        if not isinstance(key, int):
+            msg = f"Expected int, got {key}"
+            raise TypeError(msg)
+        if not isinstance(value, Device):
+            msg = f"Expected Device, got {value}"
+            raise TypeError(msg)
         self._children[key] = value
         value.parent = self
 
@@ -271,14 +279,20 @@ def _caller_locals(self) -> dict[str, Any]:
             raise ValueError
         except ValueError:
             _, _, tb = sys.exc_info()
-            assert tb, "Can't get traceback, this shouldn't happen"
+            if not tb:
+                msg = "Can't get traceback, this shouldn't happen"
+                raise RuntimeError(msg)  # noqa: B904
             caller_frame = tb.tb_frame
             while caller_frame.f_locals.get("self", None) is self:
                 caller_frame = caller_frame.f_back
-                assert caller_frame, (
-                    "No previous frame to the one with self in it, this shouldn't "
-                    "happen"
-                )
+                if not caller_frame:
+                    msg = (
+                        "No previous frame to the one with self in it, "
+                        "this shouldn't happen"
+                    )
+                    raise RuntimeError(  # noqa: B904
+                        msg
+                    )
             return caller_frame.f_locals.copy()
 
     def __enter__(self) -> DeviceProcessor:

src/ophyd_async/core/_device_filler.py

@@ -39,6 +39,13 @@ def _logical(name: UniqueName) -> LogicalName:
     return LogicalName(name.rstrip("_"))
 
 
+def _check_device_annotation(annotation: Any) -> DeviceAnnotation:
+    if not isinstance(annotation, DeviceAnnotation):
+        msg = f"Annotation {annotation} is not a DeviceAnnotation"
+        raise TypeError(msg)
+    return annotation
+
+
 @runtime_checkable
 class DeviceAnnotation(Protocol):
     @abstractmethod
@@ -150,8 +157,8 @@ def create_signals_from_annotations(
             yield backend, extras
             signal = child_type(backend)
             for anno in extras:
-                assert isinstance(anno, DeviceAnnotation), anno
-                anno(self._device, signal)
+                device_annotation = _check_device_annotation(annotation=anno)
+                device_annotation(self._device, signal)
             setattr(self._device, name, signal)
             dest = self._filled_backends if filled else self._unfilled_backends
             dest[_logical(name)] = (backend, child_type)
@@ -167,15 +174,17 @@ def create_devices_from_annotations(
             yield connector, extras
             device = child_type(connector=connector)
             for anno in extras:
-                assert isinstance(anno, DeviceAnnotation), anno
-                anno(self._device, device)
+                device_annotation = _check_device_annotation(annotation=anno)
+                device_annotation(self._device, device)
             setattr(self._device, name, device)
             dest = self._filled_connectors if filled else self._unfilled_connectors
             dest[_logical(name)] = connector
 
     def create_device_vector_entries_to_mock(self, num: int):
         for name, cls in self._vector_device_type.items():
-            assert cls, "Shouldn't happen"
+            if not cls:
+                msg = "Malformed device vector"
+                raise TypeError(msg)
             for i in range(1, num + 1):
                 if issubclass(cls, Signal):
                     self.fill_child_signal(name, cls, i)
@@ -254,9 +263,9 @@ def fill_child_device(
             # We need to add a new entry to a DeviceVector
             vector = self._ensure_device_vector(name)
             vector_device_type = self._vector_device_type[name] or device_type
-            assert issubclass(vector_device_type, Device), (
-                f"{vector_device_type} is not a Device"
-            )
+            if not issubclass(vector_device_type, Device):
+                msg = f"{vector_device_type} is not a Device"
+                raise TypeError(msg)
             connector = self._device_connector_factory()
             vector[vector_index] = vector_device_type(connector=connector)
         elif child := getattr(self._device, name, None):

src/ophyd_async/core/_readable.py

@@ -123,29 +123,31 @@ def hints(self) -> Hints:
             # we want to combine them when they are Sequences, and ensure they are
             # identical when string values.
             for key, value in new_hint.hints.items():
+                # fail early for unkwon types
                 if isinstance(value, str):
                     if key in hints:
-                        assert (
-                            hints[key] == value  # type: ignore[literal-required]
-                        ), f"Hints key {key} value may not be overridden"
+                        if hints[key] != value:
+                            msg = f"Hints key {key} value may not be overridden"
+                            raise RuntimeError(msg)
                     else:
                         hints[key] = value  # type: ignore[literal-required]
                 elif isinstance(value, Sequence):
                     if key in hints:
                         for new_val in value:
-                            assert (
-                                new_val not in hints[key]  # type: ignore[literal-required]
-                            ), f"Hint {key} {new_val} overrides existing hint"
+                            if new_val in hints[key]:
+                                msg = f"Hint {key} {new_val} overrides existing hint"
+                                raise RuntimeError(msg)
                         hints[key] = (  # type: ignore[literal-required]
                             hints[key] + value  # type: ignore[literal-required]
                         )
                     else:
                         hints[key] = value  # type: ignore[literal-required]
                 else:
-                    raise TypeError(
-                        f"{new_hint.name}: Unknown type for value '{value}' "
+                    msg = (
+                        f"{new_hint.name}: Unknown type for value '{value}'"
                         f" for key '{key}'"
                     )
+                    raise TypeError(msg)
 
         return hints
 
@@ -204,6 +206,11 @@ def add_readables(
             `StandardReadableFormat` documentation
         """
 
+        def assert_device_is_signalr(device: Device) -> SignalR:
+            if not isinstance(device, SignalR):
+                raise TypeError(f"{device} is not a SignalR")
+            return device
+
         for device in devices:
             match format:
                 case StandardReadableFormat.CHILD:
@@ -218,24 +225,24 @@ def add_readables(
                     if isinstance(device, HasHints):
                         self._has_hints += (device,)
                 case StandardReadableFormat.CONFIG_SIGNAL:
-                    assert isinstance(device, SignalR), f"{device} is not a SignalR"
-                    self._describe_config_funcs += (device.describe,)
-                    self._read_config_funcs += (device.read,)
+                    signalr_device = assert_device_is_signalr(device=device)
+                    self._describe_config_funcs += (signalr_device.describe,)
+                    self._read_config_funcs += (signalr_device.read,)
                 case StandardReadableFormat.HINTED_SIGNAL:
-                    assert isinstance(device, SignalR), f"{device} is not a SignalR"
-                    self._describe_funcs += (device.describe,)
-                    self._read_funcs += (device.read,)
-                    self._stageables += (device,)
-                    self._has_hints += (_HintsFromName(device),)
+                    signalr_device = assert_device_is_signalr(device=device)
+                    self._describe_funcs += (signalr_device.describe,)
+                    self._read_funcs += (signalr_device.read,)
+                    self._stageables += (signalr_device,)
+                    self._has_hints += (_HintsFromName(signalr_device),)
                 case StandardReadableFormat.UNCACHED_SIGNAL:
-                    assert isinstance(device, SignalR), f"{device} is not a SignalR"
-                    self._describe_funcs += (device.describe,)
-                    self._read_funcs += (_UncachedRead(device),)
+                    signalr_device = assert_device_is_signalr(device=device)
+                    self._describe_funcs += (signalr_device.describe,)
+                    self._read_funcs += (_UncachedRead(signalr_device),)
                 case StandardReadableFormat.HINTED_UNCACHED_SIGNAL:
-                    assert isinstance(device, SignalR), f"{device} is not a SignalR"
-                    self._describe_funcs += (device.describe,)
-                    self._read_funcs += (_UncachedRead(device),)
-                    self._has_hints += (_HintsFromName(device),)
+                    signalr_device = assert_device_is_signalr(device=device)
+                    self._describe_funcs += (signalr_device.describe,)
+                    self._read_funcs += (_UncachedRead(signalr_device),)
+                    self._has_hints += (_HintsFromName(signalr_device),)
 
 
 class _UncachedRead:

src/ophyd_async/core/_signal.py

@@ -97,30 +97,35 @@ def source(self) -> str:
 
 
 class _SignalCache(Generic[SignalDatatypeT]):
-    def __init__(self, backend: SignalBackend[SignalDatatypeT], signal: Signal):
-        self._signal = signal
+    def __init__(self, backend: SignalBackend[SignalDatatypeT], signal: Signal) -> None:
+        self._signal: Signal[Any] = signal
         self._staged = False
         self._listeners: dict[Callback, bool] = {}
         self._valid = asyncio.Event()
         self._reading: Reading[SignalDatatypeT] | None = None
-        self.backend = backend
+        self.backend: SignalBackend[SignalDatatypeT] = backend
         signal.log.debug(f"Making subscription on source {signal.source}")
         backend.set_callback(self._callback)
 
-    def close(self):
+    def close(self) -> None:
         self.backend.set_callback(None)
         self._signal.log.debug(f"Closing subscription on source {self._signal.source}")
 
+    def _ensure_reading(self) -> Reading[SignalDatatypeT]:
+        if not self._reading:
+            msg = "Monitor not working"
+            raise RuntimeError(msg)
+        return self._reading
+
     async def get_reading(self) -> Reading[SignalDatatypeT]:
         await self._valid.wait()
-        assert self._reading is not None, "Monitor not working"
-        return self._reading
+        return self._ensure_reading()
 
     async def get_value(self) -> SignalDatatypeT:
-        reading = await self.get_reading()
+        reading: Reading[SignalDatatypeT] = await self.get_reading()
         return reading["value"]
 
-    def _callback(self, reading: Reading[SignalDatatypeT]):
+    def _callback(self, reading: Reading[SignalDatatypeT]) -> None:
         self._signal.log.debug(
             f"Updated subscription: reading of source {self._signal.source} changed "
             f"from {self._reading} to {reading}"
@@ -134,12 +139,10 @@ def _notify(
         self,
         function: Callback[dict[str, Reading[SignalDatatypeT]] | SignalDatatypeT],
         want_value: bool,
-    ):
-        assert self._reading, "Monitor not working"
-        if want_value:
-            function(self._reading["value"])
-        else:
-            function({self._signal.name: self._reading})
+    ) -> None:
+        function(self._ensure_reading()["value"]) if want_value else function(
+            {self._signal.name: self._ensure_reading()}
+        )
 
     def subscribe(self, function: Callback, want_value: bool) -> None:
         self._listeners[function] = want_value
@@ -150,7 +153,7 @@ def unsubscribe(self, function: Callback) -> bool:
         self._listeners.pop(function)
         return self._staged or bool(self._listeners)
 
-    def set_staged(self, staged: bool):
+    def set_staged(self, staged: bool) -> bool:
         self._staged = staged
         return self._staged or bool(self._listeners)
 
@@ -167,7 +170,10 @@ def _backend_or_cache(
         if cached is None:
             cached = self._cache is not None
         if cached:
-            assert self._cache, f"{self.source} not being monitored"
+            if not self._cache:
+                msg = f"{self.source} not being monitored"
+                raise RuntimeError(msg)
+            # assert self._cache, f"{self.source} not being monitored"
             return self._cache
         else:
             return self._connector.backend

src/ophyd_async/core/_soft_signal_backend.py

@@ -175,7 +175,8 @@ async def get_setpoint(self) -> SignalDatatypeT:
         return self.reading["value"]
 
     def set_callback(self, callback: Callback[Reading[SignalDatatypeT]] | None) -> None:
+        if callback and self.callback:
+            raise RuntimeError("Cannot set a callback when one is already set")
         if callback:
-            assert not self.callback, "Cannot set a callback when one is already set"
             callback(self.reading)
         self.callback = callback