Source code for ska_low_mccs.subarray.qa_metrics_builder

#  -*- coding: utf-8 -*-
#
# This file is part of the SKA Low MCCS project
#
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""QA Metrics builder for subarray quality assurance metrics."""
from __future__ import annotations

import json
import logging
from importlib import resources
from typing import Any, Final

from jsonschema import ValidationError, validate

SUBARRAY_QA_SCHEMA: Final = json.loads(
    resources.files("ska_low_mccs.schemas.subarray_beam")
    .joinpath("MccsSubarrayBeam_QualityAssuranceMetrics_1_0.json")
    .read_text()
)




class QAMetricsBuilder:
    """Builder for constructing valid QA metrics structures."""



    def __init__(self, logger: logging.Logger) -> None:
        """
        Initialize the builder.

        :param logger: logger
        """
        self._logger = logger
        self._subarray_beams: dict[str, Any] = {}
        self._beam_definitions: dict[str, list[str]] = {}
        self._total_beams = 0




    def define_beam(
        self,
        beam_id: str,
        aperture_ids: list[str],
    ) -> QAMetricsBuilder:
        """
        Add a beam to the metrics.

        :param beam_id: Identifier for the beam
        :param aperture_ids: List of aperture IDs for the beam
        :return: Self for method chaining
        """
        is_new = not self.has_beam(beam_id)
        if is_new:
            self._total_beams += 1
        self._beam_definitions[beam_id] = aperture_ids
        return self




    def has_beam(self, beam_id: str) -> bool:
        """
        Return of beam already defined.

        :param beam_id: the beam id
        :return: true if already defined.
        """
        return beam_id in self._beam_definitions




    def update_beam_metrics(
        self,
        beam_id: str,
        beam_qa_info: dict[str, dict[str, Any]],
    ) -> QAMetricsBuilder:
        """
        Update metrics for a beam.

        :param beam_id: Identifier for the beam
        :param beam_qa_info: QA information for the beam
        :return: Self for method chaining
        """
        if beam_id not in self._beam_definitions:
            self._logger.warning(f"Ignoring metrics for undefined beam {beam_id}")
            return self
        try:
            validate(instance=beam_qa_info, schema=SUBARRAY_QA_SCHEMA)
        except ValidationError as e:
            self._logger.error(f"Invalid beam QA info for {beam_id}: {e}")
            return self

        self._subarray_beams[beam_id] = beam_qa_info
        return self




    def build(self) -> dict[str, Any]:
        """
        Build the complete QA metrics dictionary.

        :return: Valid QA metrics structure
        """
        locked_percent = self._calculate_locked_beam_percent()
        return {
            "subarray_beams": self._subarray_beams,
            "beam_locked_percent": locked_percent,
        }


    def _calculate_locked_beam_percent(self: QAMetricsBuilder) -> float:
        # Collect all allocated beam locked statuses
        locked_beam_statuses = []

        for beam_id, beam_qa in self._subarray_beams.items():
            is_beam_locked = beam_qa.get("is_beam_locked", False)
            locked_beam_statuses.append(is_beam_locked)

        # Calculate percentage
        if locked_beam_statuses:
            locked_count = sum(locked_beam_statuses)
            percent_locked = (locked_count / len(locked_beam_statuses)) * 100
            self._logger.debug(
                "Calculated locked percent: "
                f"{locked_count}/{len(locked_beam_statuses)} "
                f"beams locked = {percent_locked}%"
            )
        else:
            percent_locked = 0.0
            self._logger.debug("No beam metrics available, locked percent = 0.0%")

        return percent_locked



    def reset(self) -> QAMetricsBuilder:
        """
        Reset the builder for reuse.

        :return: Self for method chaining
        """
        self._subarray_beams = {}
        self._beam_definitions = {}
        self._total_beams = 0
        return self