sls/hla_framework_bd
0
1
Fork 0
Code Issues 28 Pull Requests 2 Actions Packages Projects 1 Releases Wiki Activity

refactor: add paths, services ctx, and extract models to own files

Browse Source
This commit is contained in:
Benjamin Labrecque
2026-07-22 12:04:55 +02:00
parent e324e3abed
commit 7e690a907e
16 changed files with 474 additions and 524 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/core/models.py
-307
View File
@@ -1,307 +0,0 @@
import re
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Any, ClassVar
import yaml
from pydantic import BaseModel, ConfigDict, field_validator
from config.paths import paths
from core.enums import ServiceStatus
class Service(BaseModel):
name: str
ioc_port: int
status: ServiceStatus
@property
def name_upper(self):
return self.name.upper()
@property
def name_lower(self):
return self.name.lower()
@property
def name_lower_underscores(self):
return self.name.replace("-", "_").lower()
@property
def dir_name(self):
return self.name.lower()
def to_dict(self):
return {"name": self.name_lower, "ioc_port": self.ioc_port, "status": str(self.status)}
class Config:
use_enum_values = True
@field_validator("name")
@classmethod
def validate_name_chars(cls, value: str) -> str:
if not re.match(r"^[a-z0-9-]+$", value):
raise ValueError("name can only contain lowercase letters, numbers, and dashes")
return value
class ServiceRegistry(BaseModel):
config_path: ClassVar[Path] = paths.service_registry_file
next_available_ioc_port: int
services: list[dict]
@classmethod
def read_from_config(cls) -> "ServiceRegistry":
with open(cls.config_path) as f:
d = yaml.safe_load(f)
registry = ServiceRegistry.model_validate(d)
return registry
def get_service(self, name: str) -> Service:
for svc_data in self.services:
service = Service(**svc_data)
if name == service.name_lower:
return service
raise ValueError(f"Could not find service with name: {name}")
def get_services(self) -> list[Service]:
return [Service(**svc_data) for svc_data in self.services]
def add_service(self, name: str) -> Service:
svc = Service(
name=name,
ioc_port=self.next_available_ioc_port,
status=ServiceStatus.ACTIVE,
)
self.services.append(svc.to_dict())
self.next_available_ioc_port += 1
return svc
def write_to_config(self) -> None:
with open(self.config_path, "w") as f:
yaml.safe_dump(self.model_dump(), f, sort_keys=False)
class MasterHLANames(BaseModel):
config_path: ClassVar[Path] = paths.master_hla_names_file
hla_apps: list[str]
@classmethod
def read_from_config(cls) -> "MasterHLANames":
with open(cls.config_path) as f:
d = yaml.safe_load(f)
hla_names = MasterHLANames.model_validate(d)
return hla_names
def add_hla_name(self, name: str):
self.hla_apps.append(name)
def write_to_config(self) -> None:
with open(self.config_path, "w") as f:
yaml.safe_dump(self.model_dump(), f, indent=4)
class IocsOverview(BaseModel):
config_path: ClassVar[Path] = paths.iocs_overview_filename
content: str
@classmethod
def read_from_file(cls) -> "IocsOverview":
with open(cls.config_path, "r", encoding="utf-8") as f:
s = f.read()
return IocsOverview(content=s)
def add_ioc(self, service: Service, description: str):
new_row = f"| AGEBD-CPCL-{service.name_upper} | {description} |\n"
# Ensuring it starts on a clean new line if the file didn't end with one
if not self.content.endswith("\n"):
self.content += "\n"
self.content += new_row
def write_to_file(self):
with open(self.config_path, "w") as f:
f.write(self.content)
class MasterIocSubs(BaseModel):
config_path: ClassVar[Path] = paths.master_ioc_subs_file
content: str
@classmethod
def read_from_file(cls) -> "MasterIocSubs":
with open(cls.config_path) as f:
file_content = f.read()
return MasterIocSubs(content=file_content)
def update_content(
self,
service_name: str,
starton: str = "0", # TODO: ok defaults?
autooff: str = "0", # TODO: ok defaults?
):
"""
Parses the configuration text and appends a new row before the closing bracket.
"""
quoted_service = f'"{service_name}"'
new_row = f' {{ "{{{{ agebd_env }}}}", {quoted_service:<22} , "{starton}", "{autooff}" }}'
# Use regex to find the last closing curly brace of the configuration block
# This targets the line that contains only a lone closing brace, optional spaces, and maybe a comma/semicolon.
pattern = r"(\s*\n\s*\}(?:;|,)?\s*$)"
if re.search(pattern, self.content, re.MULTILINE):
# Insert our new row right before that closing brace line
self.content = re.sub(
pattern, f"\n{new_row}\\1", self.content, count=1, flags=re.MULTILINE
)
else:
raise ValueError(
"Could not locate the closing structure format '}' inside the configuration."
)
def write_to_file(self) -> None:
with open(self.config_path, "w") as f:
f.write(self.content)
# TODO: maybe separate models into their own files
class ServiceManagerUI(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
config_path: ClassVar[Path] = paths.service_manager_ui_file
tree: Any
@classmethod
def read_from_file(cls) -> "ServiceManagerUI":
tree = ET.parse(cls.config_path)
return ServiceManagerUI(tree=tree)
def update_xml(self, service: Service, ui_name: str, ui_filename: str) -> None:
num_macro_elements = self._update_macro(
service=service, ui_name=ui_name, ui_filename=ui_filename
)
self._update_heights(num_macro_elements=num_macro_elements)
def _update_macro(self, service: Service, ui_name: str, ui_filename: str) -> int:
new_macro_element = _ServiceManagerUIMacroElement(
service_name_upper=service.name_upper, ui_name=ui_name, ui_filename=ui_filename
)
num_macro_elements = None
root = self.tree.getroot()
for widget in root.iter("widget"):
if widget.get("class") == "caInclude" and widget.get("name") == "cainclude":
for prop in widget.findall("property"):
# Update the macro data string
if prop.get("name") == "macro":
string_elem = prop.find("string")
if string_elem is not None and string_elem.text is not None:
string_elem.text = (
string_elem.text + ";" + new_macro_element.to_string()
)
# Dynamically update the count match so caInclude renders all rows
if prop.get("name") == "numberOfItems":
number_elem = prop.find("number")
assert number_elem is not None and number_elem.text is not None
num_macro_elements = int(number_elem.text)
number_elem.text = str(num_macro_elements + 1)
assert num_macro_elements is not None, "could not find number of macro elements in xml"
return num_macro_elements
def _update_heights(self, num_macro_elements: int) -> None:
"""
There are 3 heights to be adjusted:
1. Main window
2. Tab widget
3. Widget inside the tab widget
"""
root = self.tree.getroot()
ca_include_height = self._get_height_cainclude()
height_to_add = int(
(ca_include_height / num_macro_elements * (num_macro_elements + 1)) - ca_include_height
)
# 1. Main window
mainwindow = root.find(".//widget[@class='QMainWindow']")
if mainwindow is not None:
rect = mainwindow.find("./property[@name='geometry']/rect")
assert rect is not None
height = rect.find("height")
assert height is not None and height.text is not None
height.text = str(int(height.text) + height_to_add)
for widget in root.iter("widget"):
widget_name = widget.get("name")
# 2. Tab widget
if widget_name == "tabWidget":
rect = widget.find("./property[@name='geometry']/rect")
assert rect is not None
height = rect.find("height")
assert height is not None and height.text is not None
height.text = str(int(height.text) + height_to_add)
# 3. Widget inside the tab widget
elif widget_name == "cainclude":
rect = widget.find("./property[@name='geometry']/rect")
assert rect is not None
height = rect.find("height")
assert height is not None and height.text is not None
height.text = str(int(height.text) + height_to_add)
def _get_height_cainclude(self) -> int:
"""
Get the height of the area that includes all the services
"""
root = self.tree.getroot()
cainclude_height = None
for widget in root.iter("widget"):
widget_name = widget.get("name")
# Target the caInclude Widget of Required Services
if widget_name == "cainclude":
rect = widget.find("./property[@name='geometry']/rect")
assert rect is not None
height = rect.find("height")
assert height is not None and height.text is not None
cainclude_height = int(height.text)
assert cainclude_height, "could not get height of cainclude widget"
return cainclude_height
def write_to_file(self) -> None:
self.tree.write(self.config_path, encoding="utf-8", xml_declaration=True)
class _ServiceManagerUIMacroElement(BaseModel):
service_name_upper: str
ui_name: str
panel: int = 1
svc_exist: int = 1
ui_filename: str
def to_string(self) -> str:
return (
f"IOC=AGEBD-CPCL-{self.service_name_upper},"
f"service={self.service_name_upper},"
f"name={self.ui_name},"
f"panel={self.panel},"
f"svc_exist={self.svc_exist},"
f"panelcmd=A_BD_{self.ui_filename}.ui"
)
cli/src/core/service_creator.py
+31 -157
View File
@@ -6,7 +6,6 @@ from pathlib import Path
import copier
import git
from config.paths import paths
from core.exceptions import UVError
from core.git import (
assert_clean_repo,
@@ -14,15 +13,11 @@ from core.git import (
git_push_changes,
switch_branch,
)
from core.models import (
IocsOverview,
MasterHLANames,
MasterIocSubs,
Service,
ServiceManagerUI,
ServiceRegistry,
)
from core.utils import get_git_root
from models import (
Service,
)
from models.context import ServicesContext
logger = logging.getLogger(__name__)
@@ -32,34 +27,29 @@ SERVICE_DEST_DIR = REPO_ROOT / "services"
class ServiceCreator:
def __init__(self, user: str, ioc_description: str, ui_name: str, ui_filename: str) -> None:
self.user = user
self.ioc_description = ioc_description
self.ui_name = ui_name
self.ui_filename = ui_filename
def __init__(self, ctx: ServicesContext) -> None:
self.ctx = ctx
def add_service(self, name: str):
def add_service(
self,
name: str,
user: str,
ioc_description: str,
ui_name: str,
ui_filename: str,
):
repo = git.Repo(REPO_ROOT)
assert_clean_repo(repo)
original_branch = repo.active_branch.name
# Keep old configs to revert in case of Exception
registry_old = ServiceRegistry.read_from_config()
registry = ServiceRegistry.read_from_config()
service = registry.add_service(name)
hla_names_old = MasterHLANames.read_from_config()
hla_names = MasterHLANames.read_from_config()
hla_names.add_hla_name(service.name_upper)
iocs_overview_old = IocsOverview.read_from_file()
iocs_overview = IocsOverview.read_from_file()
iocs_overview.add_ioc(service, description=self.ioc_description)
master_ioc_subs_old = MasterIocSubs.read_from_file()
master_ioc_subs = MasterIocSubs.read_from_file()
master_ioc_subs.update_content(service_name=service.name_upper)
service_manager_old = ServiceManagerUI.read_from_file()
service_manager = ServiceManagerUI.read_from_file()
service_manager.update_xml(
service=service, ui_name=self.ui_name, ui_filename=self.ui_filename
backup = self.ctx.snapshot()
service = self.ctx.service_registry.add_service(name)
self.ctx.master_hla_names.add_hla_name(service.name_upper)
self.ctx.iocs_overview.add_ioc(service, description=ioc_description)
self.ctx.master_ioc_subs.update_content(service_name=service.name_upper)
self.ctx.service_manager_ui.update_xml(
service=service, ui_name=ui_name, ui_filename=ui_filename
)
branch_name = f"feature/add-service-{service.dir_name}"
@@ -69,23 +59,15 @@ class ServiceCreator:
try:
self._render_new_service_templates(
service=service,
user=self.user,
user=user,
ui_filename=ui_filename,
)
self._generate_uv_lock(service=service)
registry.write_to_config()
hla_names.write_to_config()
iocs_overview.write_to_file()
master_ioc_subs.write_to_file()
service_manager.write_to_file()
files_to_stage = [
paths.relative(paths.get_service_dir(service.dir_name)),
paths.relative(paths.service_registry_file),
paths.relative(paths.master_hla_names_file),
paths.relative(paths.master_ioc_subs_file),
paths.relative(paths.iocs_overview_filename),
paths.relative(paths.service_manager_ui_file),
]
self.ctx.write_all()
files_to_stage = self.ctx.paths.get_files_to_stage(service_dir_name=service.dir_name)
# TODO: can't push if branch already exists in remote
git_push_changes(
repo=repo,
@@ -96,119 +78,11 @@ class ServiceCreator:
switch_branch(repo, original_branch)
except Exception as e:
shutil.rmtree(Path(SERVICE_DEST_DIR / service.dir_name), ignore_errors=True)
registry_old.write_to_config()
hla_names_old.write_to_config()
iocs_overview_old.write_to_file()
master_ioc_subs_old.write_to_file()
service_manager_old.write_to_file()
backup.write_all()
switch_branch(repo, original_branch)
raise e
# TODO: remove
def migrate_service(self, name: str):
repo = git.Repo(REPO_ROOT)
assert_clean_repo(repo)
original_branch = repo.active_branch.name
# Keep old configs to revert in case of Exception
registry_old = ServiceRegistry.read_from_config()
registry = ServiceRegistry.read_from_config()
service = registry.add_service(name)
iocs_overview_old = IocsOverview.read_from_file()
iocs_overview = IocsOverview.read_from_file()
iocs_overview.add_ioc(service, description=self.ioc_description)
branch_name = f"feature/add-new-service-all"
# delete_local_branch(repo, branch_name)
switch_branch(repo, branch_name)
def copy_files_from_current_service():
# ui
# CURRENT_FILENAME = (
# REPO_ROOT / ".." / "sls_bd_qt" / f"A_BD_{self.ui_filename}.ui"
# ).resolve()
# DEST_FILENAME = str(
# REPO_ROOT
# / "services"
# / service.dir_name
# / "current"
# / "qt"
# / f"A_BD_{self.ui_filename}.ui"
# )
# shutil.copyfile(CURRENT_FILENAME, DEST_FILENAME)
# ioc subs
CURRENT_FILENAME = (
REPO_ROOT
/ ".."
/ "A_BD"
/ f"AGEBD-CPCL-{service.name_upper}"
/ f"AGEBD-CPCL-{service.name_upper}_main.subs"
).resolve()
DEST_FILENAME = str(
REPO_ROOT
/ "services"
/ service.dir_name
/ "current"
/ "ioc"
/ f"AGEBD-CPCL-{service.name_upper}_main.subs"
)
shutil.copyfile(CURRENT_FILENAME, DEST_FILENAME)
# ioc template
# CURRENT_FILENAME = (
# REPO_ROOT
# / ".."
# / "A_BD"
# / f"AGEBD-CPCL-{service.name_upper}"
# / f"{service.name_upper}.template"
# ).resolve()
# DEST_FILENAME = str(
# REPO_ROOT
# / "services"
# / service.dir_name
# / "current"
# / "ioc"
# / f"{service.name_upper}.template"
# )
shutil.copyfile(CURRENT_FILENAME, DEST_FILENAME)
try:
self._render_new_service_templates(
service=service,
user=self.user,
)
self._generate_uv_lock(service=service)
registry_old.write_to_config()
iocs_overview.write_to_file()
copy_files_from_current_service()
files_to_stage = [
f"services/{service.dir_name}/current",
# "config/services_registry.yml",
"docs/ioc/user/ioc_overview.md",
]
# TODO: can't push if branch already exists in remote
git_push_changes(
repo=repo,
branch_name=branch_name,
files_to_stage=files_to_stage,
commit_msg=f"feature: add new service {service.name_lower}",
)
switch_branch(repo, original_branch)
except Exception as e:
shutil.rmtree(Path(SERVICE_DEST_DIR / service.dir_name), ignore_errors=True)
registry_old.write_to_config()
iocs_overview_old.write_to_file()
switch_branch(repo, original_branch)
raise e
def _render_new_service_templates(
self,
service: Service,
user: str,
):
def _render_new_service_templates(self, service: Service, user: str, ui_filename: str):
"""
Render template files for a new service
"""
@@ -220,7 +94,7 @@ class ServiceCreator:
"service_name_lower": service.name_lower,
"service_name_lower_underscores": service.name_lower_underscores,
"user": user,
"ui_filename": self.ui_filename,
"ui_filename": ui_filename,
},
)