Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 33 additions & 6 deletions BlockServer/config/configuration.py
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -16,10 +16,12 @@

"""Contains all the code for defining a configuration or component"""

# ruff: noqa: I001
from collections import OrderedDict
from typing import Dict

from BlockServer.config.block import Block
from BlockServer.config.globalmacros import Globalmacro
from BlockServer.config.group import Group
from BlockServer.config.ioc import IOC
from BlockServer.config.metadata import MetaData
Expand All @@ -39,9 +41,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:
Expand All @@ -55,8 +58,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:
Expand Down Expand Up @@ -92,8 +103,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:
Expand Down Expand Up @@ -128,10 +139,26 @@ 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:
name: The new name for the configuration
"""
self.meta.name = name

def add_globalmacro(self, 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 self.globalmacros.keys():
print_and_log(
f"Warning: IOC '{name}' is already part of the configuration. Not adding it again."
)
else:
self.globalmacros[name.upper()] = Globalmacro(name, macros)
118 changes: 118 additions & 0 deletions BlockServer/config/globalmacros.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
# 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 typing import Any, Dict, List


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:
"""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,
}

def get(self, name: str) -> None:
return self.__getattribute__(name)

def __getitem__(self, name: str) -> None:
return self.__getattribute__(name)


class GlobalmacroHelper:
"""Converts global macro data to Globalmacro Object.

Consists of static methods only.
"""

@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 = ioc_separator
# 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
38 changes: 21 additions & 17 deletions BlockServer/core/config_holder.py
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -245,15 +245,15 @@ 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:
A copy of all the configuration IOC details
"""
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:
Expand All @@ -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:
Expand Down Expand Up @@ -325,13 +325,17 @@ 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.

Returns:
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(),
Expand Down Expand Up @@ -364,22 +368,22 @@ 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
block gateway and archiver with.
"""
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:
Expand All @@ -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:
Expand All @@ -404,18 +408,18 @@ 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():
for ioc in component.iocs.values():
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:
Expand Down
3 changes: 2 additions & 1 deletion BlockServer/core/constants.py
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -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]
Loading
Loading