import json
from typing import List
from ska_ser_skallop.utils.nrgen import get_id
standard_composition_template = {
"subarrayID": 1,
"dish": {"receptorIDList": ["0001", "0002"]},
"sdp": {
"id": "sbi-mvp01-20200325-58005",
"max_length": 100.0,
"scan_types": [
{
"id": "science_A",
"coordinate_system": "ICRS",
"ra": "21:08:47.92",
"dec": "-88:57:22.9",
"channels": [
{
"count": 744,
"start": 0,
"stride": 2,
"freq_min": 350000000.0,
"freq_max": 368000000.0,
"link_map": [[0, 0], [200, 1], [744, 2], [944, 3]],
},
{
"count": 744,
"start": 2000,
"stride": 1,
"freq_min": 360000000.0,
"freq_max": 368000000.0,
"link_map": [[2000, 4], [2200, 5]],
},
],
},
{
"id": "calibration_B",
"coordinate_system": "ICRS",
"ra": "21:12:47.92",
"dec": "-88:57:22.9",
"channels": [
{
"count": 744,
"start": 0,
"stride": 2,
"freq_min": 350000000.0,
"freq_max": 368000000.0,
"link_map": [[0, 0], [200, 1], [744, 2], [944, 3]],
},
{
"count": 744,
"start": 2000,
"stride": 1,
"freq_min": 360000000.0,
"freq_max": 368000000.0,
"link_map": [[2000, 4], [2200, 5]],
},
],
},
],
"processing_blocks": [
{
"id": "pb-mvp01-20200325-58005",
"workflow": {
"type": "realtime",
"id": "test_receive_addresses",
"version": "0.3.2",
},
"parameters": {},
}
],
},
}
[docs]def generate_standard_comp_to_file(
subarray_id: int, dish_ids: List[int], location: str, sb_id: str
) -> str:
template = standard_composition_template.copy()
template["subarrayID"] = subarray_id
template["dish"]["receptorIDList"] = [f"{id:0>4}" for id in dish_ids]
template["sdp"]["id"] = sb_id
for pb in template["sdp"]["processing_blocks"]: # pylint: disable=invalid-name
pb["id"] = get_id("pb-mvp01-********-*****")
file_name = f"{location}/standard_comp.json"
with open(file_name, "w") as file:
json.dump(template, file)
return file_name
[docs]def generate_standard_comp(subarray_id: int, dish_ids: List[int], sb_id: str) -> str:
template = standard_composition_template.copy()
template["subarrayID"] = subarray_id
template["dish"]["receptorIDList"] = [f"{id:0>4}" for id in dish_ids]
template["sdp"]["id"] = sb_id
for pb in template["sdp"]["processing_blocks"]: # pylint: disable=invalid-name
pb["id"] = get_id("pb-mvp01-********-*****")
return json.dumps(template)
[docs]def generate_tear_down_all_resources(subarray_id: int) -> str:
configuration = {"subarrayID": subarray_id, "releaseALL": True, "receptorIDList": []}
return json.dumps(configuration)