# -*- coding: utf-8 -*-
#
# This file is part of the ska-mid-cbf-mcs project
#
# Distributed under the terms of the BSD 3-Clause license.
# See LICENSE for more info.
from __future__ import annotations
import random
from pathlib import Path
from typing import Any, Dict, List
from ska_control_model import ObsMode
from ska_mid_cbf_mcs.commons.dish_utils import DISHUtils
from ska_mid_cbf_mcs.commons.global_enum import supported_values
from ska_mid_cbf_mcs.testing.param_gen.parameter_generation import (
assign_or_release_resources_gen,
configure_scan_config_gen,
delay_model_gen,
scan_config_gen,
)
from ska_mid_cbf_mcs.testing.param_gen.test_param_gen import ParameterGen
__all__ = ["SubarrayParameterGen"]
[docs]
class SubarrayParameterGen(ParameterGen):
"""A class for Mid.CBF subarray test parameter generation."""
def __init__(
self: SubarrayParameterGen,
*args: Any,
configure_scan_template_path: Path,
scan_template_path: Path,
end_scan_template_path: Path,
delay_model_template_path: Path,
assign_resources_template_path: Path,
release_resources_template_path: Path,
**kwargs: Any,
):
"""
Initialize an instance of SubarrayParameterGen().
:param configure_scan_template_path: path to template file for ConfigureScan input
:param scan_template_path: path to template file for Scan input
:param delay_model_template_path: path to template file for delay model
"""
super().__init__(*args, **kwargs)
self.configure_scan_template_path = configure_scan_template_path
self.scan_template_path = scan_template_path
self.end_scan_template_path = end_scan_template_path
self.delay_model_template_path = delay_model_template_path
self.assign_resources_template_path = assign_resources_template_path
self.release_resources_template_path = release_resources_template_path
# --- Parameter generation --- #
[docs]
def generate_parameter_dict(
self: SubarrayParameterGen,
sub_id: int,
frequency_band: str,
vcc_ids: List[int],
fsp_modes: Dict[int, ObsMode],
scan: bool,
end_scan: bool,
delay_model: bool,
assign_resources: bool,
release_resources: bool,
) -> Dict[str, Any]:
"""
Generate one set of all subarray test input parameters.
:param sub_id: a subarray ID to use in parameter generation
:param frequency_band: a frequency band to use in parameter generation
:param vcc_ids: a list of VCCs to use in parameter generation
:param fsp_modes: a dict of FSP IDs and ObsModes to use in parameter generation
:param scan: set to True to generate Scan command data
:param end_scan: set to True to generate Scan command data
:param delay_model: set to True to generate delay model data
:param assign_resources: set to True to generate delay model data
:param release_resources: set to True to generate delay model data
"""
params = {}
params["test_seed"] = self.seed
# Subarray
# Limit outgoing "sub_id" based on max number of available devices.
sub_id = sub_id if sub_id in range(1, self.max_subarrays + 1) else 1
params["sub_id"] = sub_id
# VCC
# Limit outgoing "vcc_ids" based on max number of available devices.
vcc_ids = list(vcc_ids[: self.max_vcc_all_bands])
params["vcc_ids"] = vcc_ids
# DISH
dish_map = DISHUtils(self.sys_param_dict)
dish_ids = [dish_map.vcc_id_to_dish_id[vcc_id] for vcc_id in vcc_ids]
params["dish_ids"] = dish_ids
# FSP
# Limit outgoing "fsp_modes" based on max number of available devices.
fsp_modes = dict(list(fsp_modes.items())[: self.max_fsp_units])
params["fsp_modes"] = fsp_modes
# ConfigureScan
if frequency_band == "random":
frequency_band = random.choice(
list(supported_values["frequency_band"])
)
configure_scan_dict = configure_scan_config_gen(
template_path=self.configure_scan_template_path,
subarray_id=sub_id,
frequency_band=frequency_band,
fsp_modes=fsp_modes,
dish_ids=dish_ids,
)
params["frequency_band"] = frequency_band
params["configure_scan_dict"] = configure_scan_dict
# Test value for FHS FSP subarray<x>ScanStartTimeRounded override
params["subarray_scan_start_time_rounded"] = random.randrange(
1, 1000, 1
)
# Scan
if scan:
scan_dict = scan_config_gen(template_path=self.scan_template_path)
params["scan_dict"] = scan_dict
# EndScan
if end_scan:
end_scan_dict = scan_config_gen(
template_path=self.end_scan_template_path
)
params["end_scan_dict"] = end_scan_dict
# Delay models
if delay_model:
delay_model_dict = delay_model_gen(
template_path=self.delay_model_template_path,
subarray_id=sub_id,
dish_ids=dish_ids,
)
params["delay_model_dict"] = delay_model_dict
# AssignResources
if assign_resources:
assign_resources_dict = assign_or_release_resources_gen(
template_path=self.assign_resources_template_path,
dish_ids=dish_ids,
)
params["assign_resources_dict"] = assign_resources_dict
# ReleaseResources
if release_resources:
release_resources_dict = assign_or_release_resources_gen(
template_path=self.release_resources_template_path,
dish_ids=dish_ids,
)
params["release_resources_dict"] = release_resources_dict
return params
[docs]
def generate_parameters(
self: SubarrayParameterGen,
pytest_params: Dict[str, Any],
scan: bool = False,
end_scan: bool = False,
delay_model: bool = False,
assign_resources: bool = False,
release_resources: bool = False,
) -> Dict[str, Any]:
"""
Generate all subarray test input parameters.
:param pytest_params: the dict of pytest params that defines the basic parameter generation values
contains: "sub_id", "frequency_band", "vcc_ids" and "fsp_modes"
:param delay_model: set to True to generate delay model data
"""
self.parameter_output = self.generate_parameter_dict(
sub_id=pytest_params["sub_id"],
frequency_band=pytest_params["frequency_band"],
vcc_ids=pytest_params["vcc_ids"],
fsp_modes=pytest_params["fsp_modes"],
scan=scan,
end_scan=end_scan,
delay_model=delay_model,
assign_resources=assign_resources,
release_resources=release_resources,
)
return self.parameter_output