Cleanup if/else chain

frigate/comms/dispatcher.py

@@ -44,43 +44,22 @@ class Dispatcher:
         for comm in self.comms:
             comm.subscribe(self._receive)
 
+        self._camera_settings_handlers: dict[str, Callable] = {
+            "detect": self._on_detect_command,
+            "improve_contrast": self._on_motion_improve_contrast_command,
+            "motion": self._on_motion_command,
+            "motion_contour_area": self._on_motion_contour_area_command,
+            "motion_threshold": self._on_motion_threshold_command,
+            "recording": self._on_recordings_command,
+            "snapshots": self._on_snapshots_command,
+        }
+
     def _receive(self, topic: str, payload: str) -> None:
         """Handle receiving of payload from communicators."""
-        if "detect/set" in topic:
+        if topic.endswith("set"):
             camera_name = topic.split("/")[-3]
-            self._on_detect_command(camera_name, payload)
+            command = topic.split("/")[-2]
-        elif "improve_contrast/set" in topic:
+            self._camera_settings_handlers[command](camera_name, payload)
-            camera_name = topic.split("/")[-3]
-            self._on_motion_improve_contrast_command(camera_name, payload)
-        elif "motion/set" in topic:
-            camera_name = topic.split("/")[-3]
-            self._on_motion_command(camera_name, payload)
-        elif "motion_contour_area/set":
-            camera_name = topic.split("/")[-3]
-
-            try:
-                value = int(payload)
-            except ValueError:
-                f"Received unsupported value at {topic}: {payload}"
-                return
-
-            self._on_motion_contour_area_command(camera_name, value)
-        elif "motion_threshold/set":
-            camera_name = topic.split("/")[-3]
-
-            try:
-                value = int(payload)
-            except ValueError:
-                f"Received unsupported value at {topic}: {payload}"
-                return
-
-            self._on_motion_threshold_command(camera_name, value)
-        elif "recordings/set" in topic:
-            camera_name = topic.split("/")[-3]
-            self._on_recordings_command(camera_name, payload)
-        elif "snapshots/set" in topic:
-            camera_name = topic.split("/")[-3]
-            self._on_snapshots_command(camera_name, payload)
         elif topic == "restart":
             restart_frigate()
 
@@ -157,6 +136,12 @@ class Dispatcher:
 
     def _on_motion_contour_area_command(self, camera_name: str, payload: int) -> None:
         """Callback for motion contour topic."""
+        try:
+            payload = int(payload)
+        except ValueError:
+            f"Received unsupported value for motion contour area: {payload}"
+            return
+
         motion_settings = self.config.cameras[camera_name].motion
         logger.info(f"Setting motion contour area for {camera_name}: {payload}")
         self.camera_metrics[camera_name]["motion_contour_area"].value = payload
@@ -165,6 +150,12 @@ class Dispatcher:
 
     def _on_motion_threshold_command(self, camera_name: str, payload: int) -> None:
         """Callback for motion threshold topic."""
+        try:
+            payload = int(payload)
+        except ValueError:
+            f"Received unsupported value for motion threshold: {payload}"
+            return
+
         motion_settings = self.config.cameras[camera_name].motion
         logger.info(f"Setting motion threshold for {camera_name}: {payload}")
         self.camera_metrics[camera_name]["motion_threshold"].value = payload

frigate/comms/mqtt.py

@@ -82,7 +82,7 @@ class MqttClient(Communicator):  # type: ignore[misc]
         self.publish("available", "online", retain=True)
 
     def on_mqtt_command(
-        self, client: mqtt.Client, userdata: mqtt._UserData, message: mqtt.MQTTMessage
+        self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage
     ) -> None:
         self._dispatcher(
             message.topic.replace(f"{self.mqtt_config.topic_prefix}/", ""),
@@ -92,7 +92,7 @@ class MqttClient(Communicator):  # type: ignore[misc]
     def _on_connect(
         self,
         client: mqtt.Client,
-        userdata: mqtt._UserData,
+        userdata: Any,
         flags: Any,
         rc: mqtt.ReasonCodes,
     ) -> None:
@@ -121,7 +121,7 @@ class MqttClient(Communicator):  # type: ignore[misc]
         self._set_initial_topics()
 
     def _on_disconnect(
-        self, client: mqtt.Client, userdata: mqtt._UserData, flags: Any, rc: mqtt
+        self, client: mqtt.Client, userdata: Any, flags: Any, rc: mqtt
     ) -> None:
         """Mqtt disconnection callback."""
         self.connected = False

frigate/mypy.ini

@@ -12,7 +12,8 @@ enable_error_code = ignore-without-code
 check_untyped_defs = true
 disallow_incomplete_defs = true
 disallow_subclassing_any = true
-disallow_untyped_calls = false # https://github.com/python/mypy/issues/10757
+# https://github.com/python/mypy/issues/10757
+disallow_untyped_calls = false
 disallow_untyped_decorators = true
 disallow_untyped_defs = true
 no_implicit_optional = true