import collections
from typing import List, Tuple, Union
[docs]class NameBase:
_shim = ""
def _name(self, name: str) -> str:
return f"{self._shim}{name}"
[docs]class FSPSubelement(NameBase):
type = ""
def __init__(self, tag: str, index: int) -> None:
self._shim = tag
self.index = index
def __str__(self) -> str:
return self._name(f"_{self.type}/{self.index:0>2}")
[docs] def subarray(self, nr: int) -> str: # pylint: disable=invalid-name
return self._name(f"{self.type}subarray/{self.index:0>2}_{nr:0>2}")
[docs]class Correlator(FSPSubelement):
type = "corr"
[docs]class PSS(FSPSubelement):
type = "pss"
[docs]class FSP(NameBase):
def __init__(self, tag: str, index: int) -> None:
self._shim = f"{tag}fsp"
self._index = index
self.correlator = Correlator(self._shim, index)
self.pulsar_search = PSS(self._shim, index)
def __str__(self) -> str:
return self._name(f"/s{self._index:0>2}")
[docs] def pulsar_timing(self, index: int) -> str:
return self._name(f"_pst/{index:0>2}")
[docs] def vlbi(self, index: int) -> str:
return self._name(f"_vlbi/{index:0>2}")
[docs]class VCC(NameBase):
band_mappings = {1: "12", 2: "3", 3: "4", 4: "5"}
def __init__(self, tag: str, index: int) -> None:
self._index = index
self._shim = f"{tag}vcc"
def __str__(self) -> str:
return self._name(f"/{self._index:0>3}")
[docs] def sw(self, index: int) -> str: # pylint: disable=invalid-name
return self._name(f"_sw{index}/{self._index:0>3}")
[docs] def band(self, index: int) -> str:
band_nr = self.band_mappings[index]
assert index < 5, "Only 4 bands currently allowed"
return self._name(f"_band{band_nr}/{self._index:0>3}")
[docs]class CBF(NameBase):
def __init__(self, tag) -> None:
self._shim = f"{tag}_cbf/sub_elt/"
self.tag = tag
self.master = self._name("master")
[docs] def fsp(self, index: int) -> FSP:
return FSP(f"{self.tag}_cbf/", index)
[docs] def subarray(self, index: int) -> str:
return self._name(f"subarray_{index:0>2}")
[docs] def vcc(self, index: int) -> VCC:
return VCC(f"{self.tag}_cbf/", index)
[docs]class SDP(NameBase):
def __init__(self, tag) -> None:
self._shim = f"{tag}_sdp"
self.master = self._name("/elt/master")
[docs] def subarray(self, index: int) -> str:
return self._name(f"/elt/subarray_{index}")
[docs]class CSP(NameBase):
def __init__(self, tag) -> None:
self._shim = f"{tag}_csp"
self.cbf = CBF(self._shim)
self.master = self._name("/elt/master")
[docs] def subarray(self, index: int) -> str:
return self._name(f"/elt/subarray_{index:0>2}")
[docs]class TMSubarray(NameBase):
def __init__(self, index: int, tag: str) -> None:
self.index = index
self._shim = tag
self.sdp_leaf_node = self._name(f"leaf_node/sdp_subarray{self.index:0>2}")
self.csp_leaf_node = self._name(f"leaf_node/csp_subarray{self.index:0>2}")
def __str__(self) -> str:
return self._name(f"subarray_node/{self.index}")
[docs]class TM(NameBase):
def __init__(self, tag) -> None:
self._shim = f"ska_{tag}/tm_"
self.central_node = self._name("central/central_node")
self.sdp_leaf_node = self._name("leaf_node/sdp_master")
self.csp_leaf_node = self._name("leaf_node/csp_master")
[docs] def dish_leafnode(self, index: int):
return self._name(f"leaf_node/d{index:0>4}")
[docs] def subarray(self, index: int) -> TMSubarray:
return TMSubarray(index, self._shim)
[docs]class Mid(NameBase):
tag = "mid"
csp = CSP(tag)
sdp = SDP(tag)
tm = TM(tag)
[docs] @staticmethod
def dish(index: int) -> str:
return f"{Mid.tag}_d{index:0>4}/elt/master"
[docs] @staticmethod
def dishes(ids: List[int]) -> List[str]:
return [Mid.dish(id) for id in ids]
[docs]class DeviceName:
name: str
tags: Tuple
def __init__(self, name: str, *tags: str) -> None:
self.name = name
self.tags = tags
[docs]class DomainList:
def __init__(self, token: Union[int, List[DeviceName], None]) -> None:
self._list: List[DeviceName] = []
if isinstance(token, int):
self._generate_list(token)
elif isinstance(token, list):
self._list = token
elif not token:
self._list = []
def _generate_list(self, index: int) -> None:
_ = index
self._list: List[DeviceName] = []
[docs] def get_inner_list(self) -> List[DeviceName]:
return self._list
def __getitem__(self, key: int) -> str:
return self._list[key].name
def __len__(self) -> int:
return len(self._list)
def __add__(self, other: "DomainList") -> "DomainList":
new_list = list({*self.get_inner_list(), *other.get_inner_list()})
new_domain_list = DomainList(0)
new_domain_list._list = new_list
return new_domain_list
def __iter__(self):
output = [item.name for item in self._list]
return output.__iter__()
[docs] def filter(self, *tags: str):
self._list = [item for item in self._list if any(tag in list(item.tags) for tag in tags)]
return self
[docs] def filterables(self) -> collections.Counter:
tags = [tag for item in self._list for tag in item.tags]
return collections.Counter(tags)
[docs] def subtract(self, tag: str):
self._list = [item for item in self._list if tag not in list(item.tags)]
return self
@property
def list(self) -> List[str]:
return [item.name for item in self._list]
[docs]class Masters(DomainList):
def __init__(self) -> None:
super().__init__(0)
self._generate_list(0)
def _generate_list(self, index: int):
_ = index
self._list = [
DeviceName(Mid.csp.master, "csp", "master domain", "csp master"),
DeviceName(Mid.csp.cbf.master, "cbf", "csp", "master domain"),
DeviceName(Mid.sdp.master, "sdp", "master domain"),
DeviceName(Mid.tm.central_node, "tm", "master domain"),
DeviceName(Mid.tm.sdp_leaf_node, "tm", "leaf nodes", "master domain", "sdp"),
DeviceName(Mid.tm.csp_leaf_node, "tm", "leaf nodes", "master domain", "csp"),
]
[docs]class Sensors(DomainList):
nr_bands = 4
nr_switches = 2
def _generate_list(self, index: int):
self._list = (
[
DeviceName(Mid.dish(index), "dishes", "sensor domain"),
DeviceName(Mid.tm.dish_leafnode(index), "tm", "leaf nodes", "sensor domain"),
]
+ [
DeviceName(
Mid.csp.cbf.vcc(index).band(band), "cbf", "csp", "inner", "sensor domain", "vcc"
)
for band in range(1, self.nr_bands + 1)
]
+ [
DeviceName(Mid.csp.cbf.vcc(index).sw(nr), "cbf", "inner", "sensor domain", "vcc")
for nr in range(1, self.nr_switches + 1)
]
)
[docs]class SubArrays(DomainList):
nr_of_fsps = 4
def _generate_list(self, index: int):
i = index
self._list = (
[
DeviceName(str(Mid.tm.subarray(i)), "tm", "tm domain"),
DeviceName(Mid.tm.subarray(i).csp_leaf_node, "tm", "leaf nodes", "sdp domain"),
DeviceName(Mid.tm.subarray(i).sdp_leaf_node, "tm", "leaf nodes", "csp domain"),
DeviceName(Mid.csp.subarray(i), "csp", "csp domain"),
DeviceName(Mid.sdp.subarray(i), "sdp", "sdp domain"),
]
+ [
DeviceName(
Mid.csp.cbf.fsp(j).correlator.subarray(i),
"correlator",
"fsp",
"inner",
"csp domain",
"cbf domain",
)
for j in range(1, self.nr_of_fsps + 1)
]
+ [
DeviceName(
Mid.csp.cbf.fsp(j).pulsar_search.subarray(i),
"pulsar search",
"fsp",
"inner",
"csp domain",
"cbf domain",
)
for j in range(1, self.nr_of_fsps + 1)
]
)