import copy
import logging
import os
from typing import Union
from ska_oso_pdm.sb_definition import SBDefinition
from ska_ser_skuid import EntityType, mint_skuid
LOGGER = logging.getLogger(__name__)
__all__ = [
"create_sbi",
"load_sbd",
"save_sbi",
]
[docs]
def create_sbi(sbd: SBDefinition) -> SBDefinition:
"""
Create a Scheduling Block Instance from a Scheduling Block Definition.
Currently, an SBI is a snapshot of an SBD but with EB and PB IDs replaced.
"""
# Create a new ID for the SBI
sbi_id = mint_skuid(EntityType.SBI)
LOGGER.info(f"New SBI ID mapping: {sbd.sbd_id} -> {sbi_id}")
sbi = copy.deepcopy(sbd)
# update the top-level ID. This is the SBI ID.
sbi.sbd_id = sbi_id
return sbi
[docs]
def load_sbd(path: Union[str, os.PathLike]) -> SBDefinition:
"""
Load an SBDefinition from a JSON file on disk.
:param path: path to SBD.
:return: SBDefinition object
"""
if not os.path.isfile(path):
msg = f"SB file not found: {path}"
LOGGER.error(msg)
raise IOError(msg)
with open(path, "r", encoding="utf-8") as infile:
return SBDefinition.model_validate_json(infile.read())
[docs]
def save_sbi(sbi: SBDefinition, path: str):
"""
Save an SBI to disk.
Saves an SBI (really, an SBD but with fixed IDs) the specified path.
:param sbi: SBI to serialise
:param path: output file to write
"""
with open(path, "w", encoding="utf-8") as outfile:
outfile.write(sbi.model_dump_json())
return path