"""Module to handle the subarray configurations"""
from dataclasses import dataclass
import json
from pathlib import Path
from marshmallow import Schema, fields, post_load
STORAGE_PATH = Path(Path(__file__).resolve().parents[0], "static", "subarrays")
[docs]@dataclass
class Subarray:
"""
Data class to store the subarray data structure.
:param name: name of the configuration
:type name: str
:param MeerKAT: list of the MeerKAT antennas
:type MeerKAT: list
:param SKA: list of the SKA antennas
:type SKA: list
The structure of the list of antennas is not yet defined, at the moment we can consider them list of antenna IDs.
There are two convenience properties, n_MeerKAT and n_SKA that return the number of antennas of each type.
"""
# CHECK: This code requires Python 3.7 or higher
name: str
MeerKAT: list # pylint: disable=invalid-name
SKA: list
@property
def n_MeerKAT(self): # pylint: disable=invalid-name
return len(self.MeerKAT)
@property
def n_SKA(self): # pylint: disable=invalid-name
return len(self.SKA)
[docs]class SubarraySchema(Schema):
"""
Schema to de/serialize the data of the Subarray class
"""
name = fields.Str()
MeerKAT = fields.List(fields.Int()) # pylint: disable=invalid-name
SKA = fields.List(fields.Int())
@post_load
def make_subarray(self, data, **kwargs):
return Subarray(**data)
[docs]class SubarrayStorage:
"""
Class to handle the storage of subarrays in JSON files
:param storage_path: path of the storage area
:type storage_path: str
"""
subarray_schema = SubarraySchema()
def __init__(self, storage_path):
"""
Initialize the storage area.
"""
self.storage_path = Path(storage_path)
[docs] def list(self):
"""
List the subarray files stored
"""
list_json_files = self.storage_path.glob("*.json")
return sorted([name.stem for name in list_json_files])
[docs] def load(self, name):
"""
Load one of the subarray files stored
:param name: name of the subarray configuration
:type name: str
"""
filename = self.storage_path / (name + ".json")
if filename.is_file():
with open(filename, "r") as f:
subarray_data = json.load(f)
subarray = self.subarray_schema.load(subarray_data)
return subarray
else:
raise ValueError(f"Subarray {name} not found")
# # Possible future method to store subarrays
# def write(self, subarray, overwrite=False):
# filename = self.storage_path / (subarray.name + ".json")
# if not filename.is_file() or overwrite:
# json_encoded = self.subarray_schema.dumps(subarray)
# with open(filename, "w") as out:
# out.write(json_encoded)
#