diff --git a/BlockServer/config/configuration.py b/BlockServer/config/configuration.py index 93fa3055..b88b8ed0 100644 --- a/BlockServer/config/configuration.py +++ b/BlockServer/config/configuration.py @@ -1,5 +1,5 @@ # This file is part of the ISIS IBEX application. -# Copyright (C) 2012-2016 Science & Technology Facilities Council. +# Copyright (C) 2012-2025 Science & Technology Facilities Council. # All rights reserved. # # This program is distributed in the hope that it will be useful. @@ -19,13 +19,14 @@ from collections import OrderedDict from typing import Dict +from server_common.helpers import PVPREFIX_MACRO +from server_common.utilities import print_and_log + from BlockServer.config.block import Block 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: @@ -39,9 +40,10 @@ class Configuration: meta (MetaData): The meta-data for the configuration components (OrderedDict): The components which are part of the configuration is_component (bool): Whether it is actually a component + globalmacros (OrderedDict): The globalmacros for the configuration """ - def __init__(self, macros: Dict): + def __init__(self, macros: Dict) -> None: """Constructor. Args: @@ -55,8 +57,16 @@ def __init__(self, macros: Dict): self.meta = MetaData("") self.components = OrderedDict() self.is_component = False + self.globalmacros = OrderedDict() - 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: bool | float | None, + ) -> None: """Add a block to the configuration. Args: @@ -92,8 +102,8 @@ def add_ioc( pvs: Dict = None, pvsets: Dict = None, simlevel: str = None, - remotePvPrefix: str = None, - ): + remotePvPrefix: str = None, # Has to match the mapped Java attribute #noqa: N803 + ) -> None: """Add an IOC to the configuration. Args: @@ -128,7 +138,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: diff --git a/BlockServer/config/globalmacros.py b/BlockServer/config/globalmacros.py new file mode 100644 index 00000000..39b3b886 --- /dev/null +++ b/BlockServer/config/globalmacros.py @@ -0,0 +1,131 @@ +# This file is part of the ISIS IBEX application. +# Copyright (C) 2012-2025 Science & Technology Facilities Council. +# All rights reserved. +# +# This program is distributed in the hope that it will be useful. +# This program and the accompanying materials are made available under the +# terms of the Eclipse Public License v1.0 which accompanies this distribution. +# EXCEPT AS EXPRESSLY SET FORTH IN THE ECLIPSE PUBLIC LICENSE V1.0, THE PROGRAM +# AND ACCOMPANYING MATERIALS ARE PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES +# OR CONDITIONS OF ANY KIND. See the Eclipse Public License v1.0 for more details. +# +# You should have received a copy of the Eclipse Public License v1.0 +# 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 + +import copy +from collections import OrderedDict +from typing import Any + +from server_common.utilities import print_and_log + + +class Globalmacro: + """Represents an IOC with its global macros. + + Attributes: + name (string): The name of the IOC + macros (dict): The IOC's macros + """ + + def __init__( + self, + name: str, + macros: dict[str, str] | None, + ) -> None: + """Constructor. + + Args: + name: The name of the IOC + macros: The IOC's macros + """ + self.name = name + + if macros is None: + self.macros = {} + else: + self.macros = macros + + @staticmethod + def _dict_to_list(in_dict: dict[str, Any]) -> list[Any]: + """Converts into a format better for the GUI to parse, namely a list. + + Args: + in_dict: The dictionary to be converted + + Returns: + The newly created list + """ + out_list = [] + if in_dict: + for k, v in in_dict.items(): + # Take a copy as we do not want to modify the original + c = copy.deepcopy(v) + c["name"] = k + out_list.append(c) + return out_list + + def __str__(self) -> str: + return f"{self.__class__.__name__}(name={self.name})" + + def to_dict(self) -> dict[str, str | dict[str, str]]: + """Puts the IOC-globalmacro's details into a dictionary. + + Returns: + The IOC-Global Macros' details + """ + return { + "name": self.name, + "macros": self.macros, + } + + +class GlobalmacroHelper: + """Converts global macro data to Globalmacro Object.""" + + globalmacros = OrderedDict() + + @staticmethod + def row_to_globalmacro(globalmacros: dict, row: str) -> None: + """converts a row from the globals file to globalmacro data. + + Args: + globalmacros: The current list of global macros + row: The IOC's (or All IOCs) global macro record + """ + ioc_separator = "__" + equal_to = "=" + all_iocs = "" + # Each record is of the form IOC__MACRO=VALUE + # Where there is no __ the Macro is applicable for all IOCs + if equal_to in row: + ioc_macro, value = row.rsplit(equal_to, maxsplit=1) + to_add_ioc = {} + if ioc_separator in ioc_macro: + ioc, macro = ioc_macro.split(ioc_separator, maxsplit=1) + else: + ioc = all_iocs + macro = ioc_macro + + if ioc in globalmacros: + to_add_ioc = globalmacros[ioc] + to_add_ioc[macro] = value.strip() + globalmacros[ioc] = to_add_ioc + + @staticmethod + def add_globalmacro(name: str, macros: dict) -> None: + """Add an IOC with its global macros to the configuration. + + Args: + name (string): The name of the IOC to add + macros: The macro sets relating to the IOC + + """ + # Only add it if it has not been added before + if name.upper() in GlobalmacroHelper.globalmacros.keys(): + print_and_log( + f"Warning: IOC '{name}' is already part of the configuration. Not adding it again." + ) + else: + GlobalmacroHelper.globalmacros[name.upper()] = Globalmacro(name, macros) diff --git a/BlockServer/core/config_holder.py b/BlockServer/core/config_holder.py index 9ec2651b..ac34356f 100644 --- a/BlockServer/core/config_holder.py +++ b/BlockServer/core/config_holder.py @@ -1,5 +1,5 @@ # This file is part of the ISIS IBEX application. -# Copyright (C) 2012-2016 Science & Technology Facilities Council. +# Copyright (C) 2012-2025 Science & Technology Facilities Council. # All rights reserved. # # This program is distributed in the hope that it will be useful. @@ -16,20 +16,20 @@ """Contains the code for the ConfigHolder class""" +# ruff: noqa: I001 import copy import re from collections import OrderedDict from typing import Any, Dict, List -from server_common.file_path_manager import FILEPATH_MANAGER -from server_common.helpers import PVPREFIX_MACRO -from server_common.utilities import print_and_log - from BlockServer.config.configuration import Configuration from BlockServer.config.group import Group from BlockServer.config.metadata import MetaData from BlockServer.core.constants import DEFAULT_COMPONENT, GRP_NONE from BlockServer.fileIO.file_manager import ConfigurationFileManager +from server_common.file_path_manager import FILEPATH_MANAGER +from server_common.helpers import PVPREFIX_MACRO +from server_common.utilities import print_and_log class ConfigHolder: @@ -127,7 +127,7 @@ def get_blocknames(self) -> List[str]: names.append(block.name) return names - def get_block_details(self): + def get_block_details(self) -> OrderedDict: """Get the configuration details for all the blocks including any in components. Returns: @@ -183,7 +183,7 @@ def get_group_details(self) -> Dict[str, Group]: return groups - def _set_group_details(self, redefinition) -> None: + def _set_group_details(self, redefinition: List) -> None: # Any redefinition only affects the main configuration homeless_blocks = self.get_blocknames() for grp in redefinition: @@ -245,7 +245,7 @@ def get_ioc_names(self, include_base: bool = False) -> List[str]: iocs.extend(cv.iocs) return iocs - def get_ioc_details(self): + def get_ioc_details(self) -> OrderedDict: """Get the details of the IOCs in the configuration. Returns: @@ -253,7 +253,7 @@ def get_ioc_details(self): """ return copy.deepcopy(self._config.iocs) - def get_component_ioc_details(self): + def get_component_ioc_details(self) -> Dict: """Get the details of the IOCs in any components. Returns: @@ -266,7 +266,7 @@ def get_component_ioc_details(self): iocs[ioc_name] = ioc return iocs - def get_all_ioc_details(self): + def get_all_ioc_details(self) -> Dict: """Get the details of the IOCs in the configuration and any components. Returns: @@ -325,6 +325,9 @@ def _add_ioc( f"Can't add IOC '{name}' to component '{component}': component does not exist" ) + def _globalmacros_to_list(self) -> List[Any]: + return [globalmacro.to_dict() for globalmacro in self._config.globalmacros.values()] + def get_config_details(self) -> Dict[str, Any]: """Get the details of the configuration. @@ -332,6 +335,7 @@ def get_config_details(self) -> Dict[str, Any]: A dictionary containing all the details of the configuration """ return { + "globalmacros": self._globalmacros_to_list(), "blocks": self._blocks_to_list(True), "groups": self._groups_to_list(), "iocs": self._iocs_to_list(), @@ -364,7 +368,7 @@ def is_dynamic(self) -> bool: """ return self._config.meta.isDynamic - def configures_block_gateway_and_archiver(self): + def configures_block_gateway_and_archiver(self) -> bool: """ Returns: (bool): Whether this config has a gwblock.pvlist and block_config.xml to configure the @@ -372,14 +376,14 @@ def configures_block_gateway_and_archiver(self): """ return self._config.meta.configuresBlockGWAndArchiver - def _comps_to_list(self): + def _comps_to_list(self) -> List: comps = [] for component_name, component_value in self._components.items(): if component_name.lower() != DEFAULT_COMPONENT.lower(): comps.append({"name": component_value.get_name()}) return comps - def _blocks_to_list(self, expand_macro: bool = False): + def _blocks_to_list(self, expand_macro: bool = False) -> List: blocks = self.get_block_details() blks = [] if blocks is not None: @@ -391,7 +395,7 @@ def _blocks_to_list(self, expand_macro: bool = False): blks.append(b) return blks - def _groups_to_list(self): + def _groups_to_list(self) -> List: groups = self.get_group_details() grps = [] if groups is not None: @@ -404,10 +408,10 @@ def _groups_to_list(self): grps.append(groups[GRP_NONE.lower()].to_dict()) return grps - def _iocs_to_list(self): + def _iocs_to_list(self) -> List: return [ioc.to_dict() for ioc in self._config.iocs.values()] - def _iocs_to_list_with_components(self): + def _iocs_to_list_with_components(self) -> List: ioc_list = self._iocs_to_list() for component in self._components.values(): @@ -415,7 +419,7 @@ def _iocs_to_list_with_components(self): ioc_list.append(ioc.to_dict()) return ioc_list - def _to_dict(self, data_list): + def _to_dict(self, data_list: List) -> dict | None: return None if data_list is None else {item["name"]: item for item in data_list} def set_config(self, config: Configuration, is_component: bool = False) -> None: diff --git a/BlockServer/core/constants.py b/BlockServer/core/constants.py index 3620c54d..b5528dd9 100644 --- a/BlockServer/core/constants.py +++ b/BlockServer/core/constants.py @@ -1,5 +1,5 @@ # This file is part of the ISIS IBEX application. -# Copyright (C) 2012-2016 Science & Technology Facilities Council. +# Copyright (C) 2012-2025 Science & Technology Facilities Council. # All rights reserved. # # This program is distributed in the hope that it will be useful. @@ -79,5 +79,6 @@ FILENAME_META = "meta.xml" FILENAME_SCREENS = "screens.xml" FILENAME_BANNER = "banner.xml" +FILENAME_GLOBALS = "globals.txt" SCHEMA_FOR = [FILENAME_BLOCKS, FILENAME_GROUPS, FILENAME_IOCS, FILENAME_COMPONENTS, FILENAME_META] diff --git a/BlockServer/fileIO/file_manager.py b/BlockServer/fileIO/file_manager.py index e612c59d..566e894c 100644 --- a/BlockServer/fileIO/file_manager.py +++ b/BlockServer/fileIO/file_manager.py @@ -1,5 +1,5 @@ # This file is part of the ISIS IBEX application. -# Copyright (C) 2012-2016 Science & Technology Facilities Council. +# Copyright (C) 2012-2025 Science & Technology Facilities Council. # All rights reserved. # # This program is distributed in the hope that it will be useful. @@ -13,13 +13,19 @@ # 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 + import os import re import shutil from collections import OrderedDict from xml.etree import ElementTree +from server_common.common_exceptions import MaxAttemptsExceededException +from server_common.file_path_manager import FILEPATH_MANAGER +from server_common.utilities import print_and_log, retry + from BlockServer.config.configuration import Configuration, MetaData +from BlockServer.config.globalmacros import GlobalmacroHelper from BlockServer.config.group import Group from BlockServer.config.xml_converter import ConfigurationXmlConverter from BlockServer.core.constants import ( @@ -28,18 +34,16 @@ FILENAME_BANNER, FILENAME_BLOCKS, FILENAME_COMPONENTS, + FILENAME_GLOBALS, FILENAME_GROUPS, FILENAME_IOCS, FILENAME_META, GRP_NONE, ) -from server_common.file_path_manager import FILEPATH_MANAGER from BlockServer.fileIO.schema_checker import ( ConfigurationIncompleteException, ConfigurationSchemaChecker, ) -from server_common.common_exceptions import MaxAttemptsExceededException -from server_common.utilities import print_and_log, retry RETRY_MAX_ATTEMPTS = 20 RETRY_INTERVAL = 0.5 @@ -51,7 +55,7 @@ class ConfigurationFileManager: Contains utilities to save and load configurations. """ - def find_ci(self, root_path, name): + def find_ci(self, root_path: str, name: str) -> str: """find a file with a case insensitive match""" res = "" for f in os.listdir(root_path): @@ -59,7 +63,7 @@ def find_ci(self, root_path, name): res = f return res - def load_config(self, name, macros, is_component): + def load_config(self, name: str, macros: dict, is_component: bool) -> Configuration: """Loads the configuration from the specified folder. Args: @@ -167,17 +171,19 @@ def load_config(self, name, macros, is_component): configuration.iocs = iocs configuration.components = components configuration.meta = meta + configuration.globalmacros = self.get_global_macros() + print_and_log(f"Configuration ('{name}') loaded.") return configuration @staticmethod - def _check_against_schema(xml, filename): + def _check_against_schema(xml: bytes, filename: str) -> None: regex = re.compile(re.escape(".xml"), re.IGNORECASE) name = regex.sub(".xsd", filename) schema_path = os.path.join(FILEPATH_MANAGER.schema_dir, name) ConfigurationSchemaChecker.check_xml_data_matches_schema(schema_path, xml) - def save_config(self, configuration, is_component): + def save_config(self, configuration: Configuration, is_component: bool) -> None: """Saves the current configuration with the specified name. Args: @@ -198,7 +204,7 @@ def save_config(self, configuration, is_component): meta_xml = ConfigurationXmlConverter.meta_to_xml(configuration.meta) try: components_xml = ConfigurationXmlConverter.components_to_xml(configuration.components) - except: + except Exception: # Is a component, so no components components_xml = ConfigurationXmlConverter.components_to_xml(dict()) @@ -223,14 +229,14 @@ def save_config(self, configuration, is_component): self._write_to_file(current_file, meta_xml) @retry(RETRY_MAX_ATTEMPTS, RETRY_INTERVAL, (OSError, IOError)) - def delete(self, name, is_component): + def delete(self, name: str, is_component: bool) -> None: path = self.get_path(name, is_component) if not os.path.exists(path): print_and_log(f"Directory {path} not found on filesystem.", "MINOR") return shutil.rmtree(path) - def component_exists(self, root_path, name): + def component_exists(self, root_path: str, name: str) -> None: """Checks to see if a component exists. root_path (string): The root folder where components are stored @@ -243,7 +249,7 @@ def component_exists(self, root_path, name): raise Exception("Component does not exist") @staticmethod - def copy_default(dest_path): + def copy_default(dest_path: str) -> None: """Copies the default/base component in if it does exist. Args: @@ -255,7 +261,7 @@ def copy_default(dest_path): ) @staticmethod - def _read_element_tree(file_path): + def _read_element_tree(file_path: str) -> ElementTree.Element: try: return ConfigurationFileManager._attempt_read(file_path) except MaxAttemptsExceededException: @@ -264,7 +270,7 @@ def _read_element_tree(file_path): f"is not in use by another process." ) - def _write_to_file(self, file_path, data): + def _write_to_file(self, file_path: str, data: str) -> None: try: return self._attempt_write(file_path, data) except MaxAttemptsExceededException: @@ -275,7 +281,7 @@ def _write_to_file(self, file_path, data): @staticmethod @retry(RETRY_MAX_ATTEMPTS, RETRY_INTERVAL, (OSError, IOError)) - def _attempt_read(file_path): + def _attempt_read(file_path: str) -> ElementTree.Element: """Read and return the element tree from a given xml file. Args: @@ -285,7 +291,7 @@ def _attempt_read(file_path): @staticmethod @retry(RETRY_MAX_ATTEMPTS, RETRY_INTERVAL, (OSError, IOError)) - def _attempt_write(file_path, data): + def _attempt_write(file_path: str, data: str) -> None: """Write xml data to a given configuration file. Args: @@ -311,7 +317,7 @@ def get_files_in_directory(self, path: str) -> list[str]: return files @staticmethod - def get_path(name, is_component): + def get_path(name: str, is_component: bool) -> str: if is_component: path = FILEPATH_MANAGER.get_component_path(name) else: @@ -320,10 +326,10 @@ def get_path(name, is_component): return path @staticmethod - def get_banner_config(): + def get_banner_config() -> dict: """ - Parses the banner config file into a dictionary of lists of dictionaries containing the items and buttons. - + Parses the banner config file into a dictionary of lists of dictionaries + containing the items and buttons. Returns: Dictionary containing information about banner items and buttons, empty dictionary if it doesn't exist or fails to parse. @@ -348,3 +354,15 @@ def get_banner_config(): else: banner = {} return banner + + def get_global_macros(self) -> dict: + # Import the Global macros + globals_path = os.path.join(FILEPATH_MANAGER.config_root_dir, FILENAME_GLOBALS) + globalmacros = {} + if os.path.isfile(globals_path): + with open(globals_path, "r") as file: + for line in file: + GlobalmacroHelper.row_to_globalmacro(globalmacros, line.strip()) + for key, value in globalmacros.items(): + GlobalmacroHelper.add_globalmacro(key, value) + return GlobalmacroHelper.globalmacros