Source code for

import copy
import logging
import os
from typing import Union

from ska_oso_pdm.entities.common.sb_definition import SBDefinition
from ska_oso_pdm.schemas import CODEC as pdm_CODEC
from ska_ser_skuid.client import SkuidClient
from import SKAEntityTypeService

LOGGER = logging.getLogger(__name__)

[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. """ # TODO inject this dependency uid_client = SkuidClient(os.environ["SKUID_URL"]) # Create a new ID for the SBI sbi_map = { sbd.sbd_id: uid_client.fetch_skuid( SKAEntityTypeService.DefaultEntityTypes.SchedulingBlockInstance ) }"New SBI ID mapping: {sbd.sbd_id} -> {sbi_map[sbd.sbd_id]}") # Create a mapping of old processing block IDs to new IDs if sbd.sdp_configuration is not None: sdp_config = sbd.sdp_configuration pb_map = { pb.pb_id: uid_client.fetch_skuid( SKAEntityTypeService.DefaultEntityTypes.ProcessingBlock ) for pb in sdp_config.processing_blocks if sdp_config is not None } for old_id, new_id in pb_map.items():"New PB ID mapping: {old_id} -> {new_id}") sbi = copy.deepcopy(sbd) # update the top-level ID. This is the SBI ID. sbi.sbd_id = sbi_map[sbd.sbd_id] if sbd.sdp_configuration is not None: sdp_config = sbi.sdp_configuration # Get EB ID for SDP Execution Block # # Currently, SBIs are effectively straight copies of SBDs. Eventually, # SBIs will become real entities with an entity ID, stored in the ODA. # These entities will be lightweight, *referencing* a specific SBD # version, rather than including all the SBD content in the SBI. # Once this is implemented, copying of EB ID into the SDP config would # be done in the PDM transform rather than when creating an SBI. sdp_config.execution_block.eb_id = uid_client.fetch_skuid( SKAEntityTypeService.DefaultEntityTypes.ExecutionBlock ) # Trawl through the SDP Processing Blocks replacing old IDs with new for pb in sdp_config.processing_blocks: pb.pb_id = pb_map[pb.pb_id] # Expect the mapping of PBs to SBDs and (or??) SBIs to become # more complicated. For a commensal SB built from a host SB and # guest SB (e.g., interferometric observing SB plus a pulsar # search SB), certain processing blocks might only reference their # originating SBs. That's moot at the moment as we only operate on # one SB. if sbi.sbd_id not in pb.sbi_ids: pb.sbi_ids.append(sbi.sbd_id) if pb.dependencies: for dependency in pb.dependencies: dependency.pb_id = pb_map[dependency.pb_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) return pdm_CODEC.load_from_file(SBDefinition, path)
[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 """ sbi_json = pdm_CODEC.dumps(sbi) with open(path, "w") as outfile: outfile.write(sbi_json) return path