Ticket8445 handle removal of defaults from config

BlockServer/component_switcher/component_switcher.py

@@ -2,18 +2,22 @@
 
 import json
 import os
-import types
 from queue import Queue
-from typing import Any, Dict, Iterable, List, Set, Type
+from typing import Any, Callable, Dict, Iterable, List, Optional, Set
 
-from BlockServer.core.config_list_manager import ConfigListManager
+import numpy.typing as npt
 from server_common.channel_access import ChannelAccess
 from server_common.helpers import MACROS, PVPREFIX_MACRO
 from server_common.utilities import SEVERITY
 from server_common.utilities import print_and_log as _common_print_and_log
 
+from BlockServer.core.config_list_manager import ConfigListManager
+
+type PVBaseValue = bool | int | float | str
+type PVValue = PVBaseValue | list[PVBaseValue] | npt.NDArray | None
+
 
-def print_and_log(message: str, *args, **kwargs) -> None:
+def print_and_log(message: str, *args: str, **kwargs: str) -> None:
     _common_print_and_log(f"ComponentSwitcher: {message}", *args, **kwargs)
 
 
@@ -43,9 +47,9 @@ def __init__(
         self,
         config_list: ConfigListManager,
         blockserver_write_queue: Queue,
-        reload_current_config_func: types.FunctionType,
-        file_manager: ComponentSwitcherConfigFileManager = None,
-        channel_access_class: Type[ChannelAccess] = None,
+        reload_current_config_func: Callable[[], None],
+        file_manager: Optional[ComponentSwitcherConfigFileManager] = None,
+        channel_access_class: Optional[ChannelAccess] = None,
     ) -> None:
         self._config_list = config_list
         self._blockserver_write_queue = blockserver_write_queue
@@ -89,7 +93,7 @@ def create_monitors(self) -> None:
 
             print_and_log("Adding monitor to PV {}".format(pv))
 
-            def callback(val: Any, stat: int, sevr: int) -> None:
+            def callback(val: PVValue, stat: int, sevr: int) -> None:
                 """
                 Callback function called when the monitored PV changes.
 

BlockServer/config/block.py

@@ -13,7 +13,7 @@
 # along with this program; if not, you can obtain a copy from
 # https://www.eclipse.org/org/documents/epl-v10.php or
 # http://opensource.org/licenses/eclipse-1.0.php
-from typing import Dict, Union
+from typing import Dict, TypedDict, Union
 
 from server_common.helpers import PVPREFIX_MACRO
 
@@ -41,17 +41,17 @@ def __init__(
         pv: str,
         local: bool = True,
         visible: bool = True,
-        component: str = None,
+        component: str | None = None,
         runcontrol: bool = False,
-        lowlimit: float = None,
-        highlimit: float = None,
+        lowlimit: float | None = None,
+        highlimit: float | None = None,
         suspend_on_invalid: bool = False,
         log_periodic: bool = False,
         log_rate: float = 5,
         log_deadband: float = 0,
         set_block: bool = False,
-        set_block_val: str = None,
-    ):
+        set_block_val: str | None = None,
+    ) -> None:
         """Constructor.
 
         Args:
@@ -92,24 +92,26 @@ def _get_pv(self) -> str:
             pv_name = PVPREFIX_MACRO + self.pv
         return pv_name
 
-    def set_visibility(self, visible: bool):
+    def set_visibility(self, visible: bool) -> None:
         """Toggle the visibility of the block.
 
         Args:
             visible: Whether the block is visible or not
         """
         self.visible = visible
 
-    def __str__(self):
+    def __str__(self) -> str:
         set_block_str = ""
         if self.set_block:
             set_block_str = f", SetBlockVal: {self.set_block_val}"
         return (
-            f"Name: {self.name}, PV: {self.pv}, Local: {self.local}, Visible: {self.visible}, Component: {self.component}"
-            f", RCEnabled: {self.rc_enabled}, RCLow: {self.rc_lowlimit}, RCHigh: {self.rc_highlimit}{set_block_str}"
+            f"Name: {self.name}, PV: {self.pv}, Local: {self.local}, "
+            f"Visible: {self.visible}, Component: {self.component}"
+            f", RCEnabled: {self.rc_enabled}, RCLow: {self.rc_lowlimit}, "
+            f"RCHigh: {self.rc_highlimit}{set_block_str}"
         )
 
-    def to_dict(self) -> Dict[str, Union[str, float, bool]]:
+    def to_dict(self) -> Dict[str, Union[str, float, bool, None]]:
         """Puts the block's details into a dictionary.
 
         Returns:
@@ -131,3 +133,17 @@ def to_dict(self) -> Dict[str, Union[str, float, bool]]:
             "set_block": self.set_block,
             "set_block_val": self.set_block_val,
         }
+
+
+class BlockKwargs(TypedDict, total=False):
+    visible: bool
+    component: str | None
+    runcontrol: bool
+    lowlimit: float | None
+    highlimit: float | None
+    suspend_on_invalid: bool
+    log_periodic: bool
+    log_rate: float
+    log_deadband: float
+    set_block: bool
+    set_block_val: str | None

BlockServer/config/configuration.py

@@ -17,15 +17,16 @@
 """Contains all the code for defining a configuration or component"""
 
 from collections import OrderedDict
-from typing import Dict
+from typing import Dict, Unpack
 
-from BlockServer.config.block import Block
+from server_common.helpers import PVPREFIX_MACRO
+from server_common.utilities import print_and_log
+
+from BlockServer.config.block import Block, BlockKwargs
 from BlockServer.config.group import Group
 from BlockServer.config.ioc import IOC
 from BlockServer.config.metadata import MetaData
 from BlockServer.core.constants import GRP_NONE
-from server_common.helpers import PVPREFIX_MACRO
-from server_common.utilities import print_and_log
 
 
 class Configuration:
@@ -41,7 +42,7 @@ class Configuration:
         is_component (bool): Whether it is actually a component
     """
 
-    def __init__(self, macros: Dict):
+    def __init__(self, macros: Dict) -> None:
         """Constructor.
 
         Args:
@@ -56,7 +57,14 @@ def __init__(self, macros: Dict):
         self.components = OrderedDict()
         self.is_component = False
 
-    def add_block(self, name: str, pv: str, group: str = GRP_NONE, local: bool = True, **kwargs):
+    def add_block(
+        self,
+        name: str,
+        pv: str,
+        group: str = GRP_NONE,
+        local: bool = True,
+        **kwargs: Unpack[BlockKwargs],
+    ) -> None:
         """Add a block to the configuration.
 
         Args:
@@ -85,15 +93,15 @@ def add_block(self, name: str, pv: str, group: str = GRP_NONE, local: bool = Tru
     def add_ioc(
         self,
         name: str,
-        component: str = None,
-        autostart: bool = None,
-        restart: bool = None,
-        macros: Dict = None,
-        pvs: Dict = None,
-        pvsets: Dict = None,
-        simlevel: str = None,
-        remotePvPrefix: str = None,
-    ):
+        component: str | None = None,
+        autostart: bool | None = None,
+        restart: bool | None = None,
+        macros: Dict | None = None,
+        pvs: Dict | None = None,
+        pvsets: Dict | None = None,
+        simlevel: str | None = None,
+        remote_pv_prefix: str | None = None,
+    ) -> None:
         """Add an IOC to the configuration.
 
         Args:
@@ -115,7 +123,15 @@ def add_ioc(
             )
         else:
             self.iocs[name.upper()] = IOC(
-                name, autostart, restart, component, macros, pvs, pvsets, simlevel, remotePvPrefix
+                name,
+                autostart if autostart is not None else True,
+                restart if restart is not None else True,
+                component,
+                macros,
+                pvs,
+                pvsets,
+                simlevel,
+                remote_pv_prefix,
             )
 
     def get_name(self) -> str:
@@ -128,7 +144,7 @@ def get_name(self) -> str:
             self.meta.name.decode("utf-8") if isinstance(self.meta.name, bytes) else self.meta.name
         )
 
-    def set_name(self, name: str):
+    def set_name(self, name: str) -> None:
         """Sets the configuration's name.
 
         Args:

BlockServer/config/group.py

@@ -25,7 +25,7 @@ class Group:
         component (string): The component the group belongs to
     """
 
-    def __init__(self, name: str, component: str = None):
+    def __init__(self, name: str, component: str | None = None) -> None:
         """Constructor.
 
         Args:
@@ -39,7 +39,7 @@ def __init__(self, name: str, component: str = None):
     def __str__(self) -> str:
         return f"Name: {self.name}, COMPONENT: {self.component}, Blocks: {self.blocks}"
 
-    def to_dict(self) -> Dict[str, Union[str, List]]:
+    def to_dict(self) -> Dict[str, Union[str, List, None]]:
         """Puts the group's details into a dictionary.
 
         Returns:

BlockServer/config/ioc.py

@@ -23,8 +23,10 @@ class IOC:
 
     Attributes:
         name (string): The name of the IOC
-        autostart (bool): Whether the IOC should automatically start/restart when the configuration is loaded/changed
-        restart (bool): If auto start is true, then proc serv will restart the IOC if it terminates unexpectedly
+        autostart (bool): Whether the IOC should automatically
+            start/restart when the configuration is loaded/changed
+        restart (bool): If auto start is true, then proc serv will
+            restart the IOC if it terminates unexpectedly
         component (string): The component the IOC belongs to
         macros (dict): The IOC's macros
         pvs (dict): The IOC's PVs
@@ -37,43 +39,50 @@ def __init__(
         name: str,
         autostart: bool = True,
         restart: bool = True,
-        component: str = None,
-        macros: Dict = None,
-        pvs: Dict = None,
-        pvsets: Dict = None,
-        simlevel: str = None,
-        remotePvPrefix: str = None,
-    ):
+        component: str | None = None,
+        macros: Dict | None = None,
+        pvs: Dict | None = None,
+        pvsets: Dict | None = None,
+        simlevel: str | None = None,
+        remotePvPrefix: str | None = None,  # noqa: N803
+    ) -> None:
         """Constructor.
 
         Args:
             name: The name of the IOC
-            autostart: Whether the IOC should automatically start/restart when the configuration is
+            autostart: Whether the IOC should automatically
+                start/restart when the configuration is
             loaded/changed
-            restart: If auto start is true, then proc serv will restart the IOC if it terminates unexpectedly
+            restart: If auto start is true, then proc serv will
+                restart the IOC if it terminates unexpectedly
             component: The component the IOC belongs to
             macros: The IOC's macros
             pvs: The IOC's PVs
             pvsets: The IOC's PV sets
             simlevel: The level of simulation
-            remotePvPrefix: The remote pv prefix
+            remotePvPrefix: The remote pv prefix,
+                has to be formatted like this to be read in properly.
         """
         self.name = name
         self.autostart = autostart
         self.restart = restart
         self.component = component
-        self.remotePvPrefix = remotePvPrefix
+        self.remote_pv_prefix = remotePvPrefix
 
         if simlevel is None:
             self.simlevel = "none"
         else:
             self.simlevel = simlevel.lower()
-
-        if macros is None:
-            self.macros = {}
-        else:
-            self.macros = macros
-
+        self.macros = {}
+        if macros is not None:
+            for name, data in macros.items():
+                if "useDefault" in data and (
+                    (not data["useDefault"]) or data["useDefault"] == "False"
+                ):
+                    self.macros.update({name: data})
+                    self.macros[name].pop("useDefault")
+                elif "useDefault" not in data:
+                    self.macros.update({name: data})
         if pvs is None:
             self.pvs = {}
         else:
@@ -107,7 +116,7 @@ def _dict_to_list(in_dict: Dict[str, Any]) -> List[Any]:
     def __str__(self) -> str:
         return f"{self.__class__.__name__}(name={self.name}, component={self.component})"
 
-    def to_dict(self) -> Dict[str, Union[str, bool, List[Any]]]:
+    def to_dict(self) -> Dict[str, Union[str, bool, List[Any], None]]:
         """Puts the IOC's details into a dictionary.
 
         Returns:
@@ -122,11 +131,11 @@ def to_dict(self) -> Dict[str, Union[str, bool, List[Any]]]:
             "pvsets": self._dict_to_list(self.pvsets),
             "macros": self._dict_to_list(self.macros),
             "component": self.component,
-            "remotePvPrefix": self.remotePvPrefix,
+            "remotePvPrefix": self.remote_pv_prefix,
         }
 
-    def get(self, name):
+    def get(self, name: str) -> bool | str | Dict | None:
         return self.__getattribute__(name)
 
-    def __getitem__(self, name):
+    def __getitem__(self, name: str) -> bool | str | Dict | None:
         return self.__getattribute__(name)

BlockServer/config/json_converter.py

@@ -15,11 +15,13 @@
 # http://opensource.org/licenses/eclipse-1.0.php
 
 import json
-from collections import OrderedDict
-from typing import Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List
 
 from BlockServer.core.constants import GRP_NONE
 
+if TYPE_CHECKING:
+    from BlockServer.config.group import Group
+
 
 class ConfigurationJsonConverter:
     """Helper class for converting configuration data to and from JSON.
@@ -29,7 +31,7 @@ class ConfigurationJsonConverter:
     """
 
     @staticmethod
-    def _groups_to_list(groups: OrderedDict) -> List[Dict[str, Any]]:
+    def _groups_to_list(groups: Dict[str, "Group"]) -> List[Dict[str, Any]]:
         grps = []
         if groups is not None:
             for group in groups.values():
@@ -46,7 +48,7 @@ def _groups_to_list(groups: OrderedDict) -> List[Dict[str, Any]]:
         return grps
 
     @staticmethod
-    def groups_to_json(groups: OrderedDict) -> str:
+    def groups_to_json(groups: Dict[str, "Group"]) -> str:
         """Converts the groups dictionary to a JSON list
 
         Args: