daemon: fixed issue for CoreError messages in new hooks module, updated new modules to all use non deprecated type hinting

daemon/core/emulator/broadcast.py

@@ -51,6 +51,11 @@ class BroadcastManager:
         :return: nothing
         """
         handlers = self.handlers.setdefault(data_type, set())
+        if handler in handlers:
+            raise CoreError(
+                f"cannot add data({data_type}) handler({repr(handler)}), "
+                f"already exists"
+            )
         handlers.add(handler)
 
     def remove_handler(self, data_type: type[T], handler: Callable[[T], None]) -> None:
@@ -65,6 +70,6 @@ class BroadcastManager:
         if handler not in handlers:
             raise CoreError(
                 f"cannot remove data({data_type}) handler({repr(handler)}), "
-                f"does not exist "
+                f"does not exist"
             )
         handlers.remove(handler)

daemon/core/emulator/controlnets.py

@@ -1,5 +1,5 @@
 import logging
-from typing import TYPE_CHECKING, Optional, Tuple
+from typing import TYPE_CHECKING, Optional
 
 from core import utils
 from core.emulator.data import InterfaceData
@@ -32,7 +32,7 @@ class ControlNetManager:
 
     def _get_server_ifaces(
         self,
-    ) -> Tuple[None, Optional[str], Optional[str], Optional[str]]:
+    ) -> tuple[None, Optional[str], Optional[str], Optional[str]]:
         """
         Retrieve control net server interfaces.
 
@@ -48,7 +48,7 @@ class ControlNetManager:
 
     def _get_prefixes(
         self,
-    ) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
+    ) -> tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
         """
         Retrieve control net prefixes.
 

daemon/core/emulator/hooks.py

@@ -1,7 +1,7 @@
 import logging
 import subprocess
+from collections.abc import Callable
 from pathlib import Path
-from typing import Callable, Dict, List
 
 from core.emulator.enumerations import EventTypes
 from core.errors import CoreError
@@ -18,8 +18,8 @@ class HookManager:
         """
         Create a HookManager instance.
         """
-        self.script_hooks: Dict[EventTypes, Dict[str, str]] = {}
+        self.script_hooks: dict[EventTypes, dict[str, str]] = {}
-        self.callback_hooks: Dict[EventTypes, List[Callable[[], None]]] = {}
+        self.callback_hooks: dict[EventTypes, list[Callable[[], None]]] = {}
 
     def reset(self) -> None:
         """
@@ -43,9 +43,7 @@ class HookManager:
         state_hooks = self.script_hooks.setdefault(state, {})
         if file_name in state_hooks:
             raise CoreError(
-                "adding duplicate state(%s) hook script(%s)",
+                f"adding duplicate state({state.name}) hook script({file_name})",
-                state.name,
-                file_name,
             )
         state_hooks[file_name] = data
 
@@ -60,9 +58,8 @@ class HookManager:
         state_hooks = self.script_hooks.get(state, {})
         if file_name not in state_hooks:
             raise CoreError(
-                "deleting state(%s) hook script(%s) that does not exist",
+                f"deleting state({state.name}) hook script({file_name}) "
-                state.name,
+                "that does not exist",
-                file_name,
             )
         del state_hooks[file_name]
 
@@ -80,9 +77,7 @@ class HookManager:
         if hook in hooks:
             name = getattr(callable, "__name__", repr(hook))
             raise CoreError(
-                "adding duplicate state(%s) hook callback(%s)",
+                f"adding duplicate state({state.name}) hook callback({name})",
-                state.name,
-                name,
             )
         hooks.append(hook)
 
@@ -100,14 +95,13 @@ class HookManager:
         if hook not in hooks:
             name = getattr(callable, "__name__", repr(hook))
             raise CoreError(
-                "deleting state(%s) hook callback(%s) that does not exist",
+                f"deleting state({state.name}) hook callback({name}) "
-                state.name,
+                "that does not exist",
-                name,
             )
         hooks.remove(hook)
 
     def run_hooks(
-        self, state: EventTypes, directory: Path, env: Dict[str, str]
+        self, state: EventTypes, directory: Path, env: dict[str, str]
     ) -> None:
         """
         Run all hooks for the current state.
@@ -137,10 +131,8 @@ class HookManager:
                         )
                 except (IOError, subprocess.CalledProcessError) as e:
                     raise CoreError(
-                        "failure running state(%s) hook script(%s): %s",
+                        f"failure running state({state.name}) "
-                        state.name,
+                        f"hook script({file_name}): {e}",
-                        file_name,
-                        e,
                     )
         for hook in self.callback_hooks.get(state, []):
             try:
@@ -148,8 +140,6 @@ class HookManager:
             except Exception as e:
                 name = getattr(callable, "__name__", repr(hook))
                 raise CoreError(
-                    "failure running state(%s) hook callback(%s): %s",
+                    f"failure running state({state.name}) "
-                    state.name,
+                    f"hook callback({name}): {e}",
-                    name,
-                    e,
                 )