Source code for ska_low_mccs.subarray_beam.beam_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.
"""Beam Metrics builder for subarray beam quality assurance metrics."""
from __future__ import annotations

import logging
from typing import Any




class BeamMetricsBuilder:
    """Builder for constructing valid beam QA metrics structures."""



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

        :param logger: logger
        """
        self._logger = logger
        self._apertures: dict[str, dict[str, Any]] = {}
        self._is_beam_locked: bool = False




    def update_aperture_metrics(
        self,
        aperture_id: str,
        is_beam_locked: bool,
    ) -> BeamMetricsBuilder:
        """
        Update metrics for an aperture.

        :param aperture_id: Identifier for the aperture
        :param is_beam_locked: Whether beam is locked for this aperture
        :return: Self for method chaining
        """
        self._apertures[aperture_id] = {"is_beam_locked": is_beam_locked}
        return self




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

        :return: Valid QA metrics structure
        """
        beam_qa_info = {
            "apertures": self._apertures,
            "is_beam_locked": self._is_beam_locked,
        }

        return beam_qa_info




    def set_beam_locked(self, is_locked: bool) -> BeamMetricsBuilder:
        """
        Set the overall beam locked status.

        :param is_locked: Whether the beam is locked
        :return: Self for method chaining
        """
        self._is_beam_locked = is_locked
        return self




    def check_all_locked(self) -> bool:
        """
        Check if all tracked apertures are locked.

        :return: True if all apertures are locked, False otherwise
        """
        locked_statuses = [
            metrics.get("is_beam_locked", False) for metrics in self._apertures.values()
        ]
        return all(locked_statuses) if locked_statuses else False




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

        :return: Self for method chaining
        """
        self._apertures = {}
        self._is_beam_locked = False
        return self