diff --git a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/antennafield_device.py b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/antennafield_device.py
index b8dc08d6852340177ce102caa10a5f7282de6fe3..e71044b738b84503db1746fe6272ae012b8f8156 100644
--- a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/antennafield_device.py
+++ b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/antennafield_device.py
@@ -47,8 +47,8 @@ from tangostationcontrol.devices.base_device_classes.mapper import (
MappedAttribute,
AntennaToSdpMapper,
RecvDeviceWalker,
- HBAToRecvMapper,
- LBAToRecvMapper,
+ AFHToRecvMapper,
+ AFLToRecvMapper,
)
from tangostationcontrol.devices.types import DeviceTypes
@@ -813,13 +813,13 @@ class AF(LOFARDevice):
MappingKeys.POWER: power_mapping,
}
if self.ANTENNA_TYPE == "HBA":
- self.__recv_mapper = HBAToRecvMapper(
+ self.__recv_mapper = AFHToRecvMapper(
recv_mapping,
self.get_mapped_dimensions("RECV"),
number_of_receivers,
)
elif self.ANTENNA_TYPE == "LBA":
- self.__recv_mapper = LBAToRecvMapper(
+ self.__recv_mapper = AFLToRecvMapper(
recv_mapping,
self.get_mapped_dimensions("RECV"),
number_of_receivers,
diff --git a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
index 79b459eac2bb278694aac09256241da0a2f0afd9..b86fb0fab924dc581e1757d10482eacbc98178b9 100644
--- a/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
+++ b/tangostationcontrol/tangostationcontrol/devices/base_device_classes/mapper.py
@@ -4,6 +4,7 @@
""" Mapper for TANGO devices that share a common attribute
"""
+from abc import ABC, abstractmethod
from enum import Enum
from typing import List, Optional, Dict
from collections.abc import Callable
@@ -29,12 +30,12 @@ __all__ = [
"AntennaMapper",
"AntennaToRecvMapper",
"AntennaToSdpMapper",
- "HBAToRecvMapper",
- "LBAToRecvMapper",
+ "AFHToRecvMapper",
+ "AFLToRecvMapper",
]
-class MappingKeys(Enum):
+class MappingKeys(str, Enum):
"""Types of Mapping"""
POWER = "POWER"
@@ -90,7 +91,6 @@ class AntennaMapper:
self,
mapping_dict: Dict[str, numpy.ndarray],
num_devices: Optional[int] = 1,
- antenna_type: str = "HBA",
) -> None:
self._mapping_dict = mapping_dict
self._num_devices = num_devices
@@ -101,10 +101,11 @@ class AntennaMapper:
self._reshape_attributes_out = {}
self._fill_attributes_in = {}
self._empty_attributes_out = {}
- self._antenna_type = antenna_type
self._pwr_mapping = []
- def map_read(self, mapped_attribute: str, device_values: List[any]) -> List[any]:
+ def map_read(
+ self, mapped_attribute: str, device_values: List[any]
+ ) -> numpy.ndarray:
"""Perform a mapped read for the attribute using the device_results
:param mapped_attribute: attribute identifier as present in
@@ -113,9 +114,24 @@ class AntennaMapper:
:return: device_results as mapped given attribute dimensions and control mapping
"""
default_values = self._default_value_mapping_read[mapped_attribute]
+ device_values = self._reshape_r_values(mapped_attribute, device_values)
+ mapped_values = numpy.array(default_values)
+ value_mapping = self._value_mapper[mapped_attribute]
+ # If mapping is based on only one map, then insert the value
+ # provided by the map
+ mapping = value_mapping[0]
+ for idx, (dev_conn, dev_input) in enumerate(mapping):
+ # Differentiate operations between RECVs and FPGAs
+ if dev_conn > 0 and MappingKeys.CONTROL in list(self._mapping_dict.keys()):
+ mapped_values[idx] = device_values[dev_conn - 1][dev_input]
+ elif dev_input > -1 and MappingKeys.FPGA in list(self._mapping_dict.keys()):
+ mapped_values[idx] = device_values[dev_conn]
- # Attributes which need to be reshaped with a copy of their values,
- # because Device original dimension < AntennaField mapped dimension
+ return mapped_values
+
+ def _reshape_r_values(self, mapped_attribute: str, device_values: List[any]):
+ """Attributes which need to be reshaped with a copy of their values,
+ because Device original dimension < AntennaField mapped dimension"""
if mapped_attribute in self._fill_attributes_in:
device_values = [
[x] * self._fill_attributes_in[mapped_attribute] for x in device_values
@@ -126,16 +142,9 @@ class AntennaMapper:
device_values,
(self._num_devices,) + self._reshape_attributes_in[mapped_attribute],
)
+ return device_values
- return self._mapped_r_values(
- device_values,
- default_values,
- self._value_mapper[mapped_attribute],
- self._antenna_type,
- mapped_attribute in self._pwr_mapping,
- )
-
- def map_write(self, mapped_attribute: str, set_values: List[any]) -> List[any]:
+ def map_write(self, mapped_attribute: str, set_values: List[any]) -> numpy.ndarray:
"""Perform a mapped write for the attribute using the set_values
:param mapped_attribute: attribute identifier as present in
@@ -144,99 +153,8 @@ class AntennaMapper:
:return: set_values as mapped given attribute dimensions and control mapping
"""
default_values = self._masked_value_mapping_write[mapped_attribute]
+ value_mapping = self._value_mapper[mapped_attribute]
- mapped_values = self._mapped_rw_values(
- set_values,
- default_values,
- self._value_mapper[mapped_attribute],
- self._antenna_type,
- mapped_attribute in self._pwr_mapping,
- )
-
- # Attributes which need to be reshaped with removing a part of their duplicated
- # values because Device original dimension < Antennafield mapped dimension
- if mapped_attribute in self._empty_attributes_out:
- mapped_values = numpy.reshape(
- mapped_values,
- (self._num_devices,) + self._empty_attributes_out[mapped_attribute],
- )
- # Remove second (last) dimension
- mapped_values = mapped_values[:, :, 0]
-
- # Only RECV mapping may have more than one devices by now
- if mapped_attribute in self._reshape_attributes_out:
- mapped_values = numpy.reshape(
- mapped_values,
- (self._num_devices,) + self._reshape_attributes_out[mapped_attribute],
- )
-
- return mapped_values
-
- def _mapped_r_values(
- self,
- device_values: List[any],
- default_values: List[any],
- value_mapping: List[any],
- antenna_type: str,
- pwr_attribute: bool,
- ):
- """Mapping for read operation"""
- mapped_values = numpy.array(default_values)
-
- # If mapping is based on only one map, then insert the value
- # provided by the map
- if len(value_mapping) == 1:
- mapping = value_mapping[0]
- for idx, (dev_conn, dev_input) in enumerate(mapping):
- # Differentiate operations between RECVs and FPGAs
- if dev_conn > 0 and MappingKeys.CONTROL in list(
- self._mapping_dict.keys()
- ):
- mapped_values[idx] = device_values[dev_conn - 1][dev_input]
- elif dev_input > -1 and MappingKeys.FPGA in list(
- self._mapping_dict.keys()
- ):
- mapped_values[idx] = device_values[dev_conn]
-
- # If mapping is based on 2 maps
- # f.e. power and control maps in AntennaToRecvMapper
- if len(value_mapping) == 2:
- # Assuming mapper lists are always in the following order:
- # [Power_Mapping, Control_Mapping]
- [power_mapping, control_mapping] = value_mapping
- for idx, (
- (dev_power, dev_input_power),
- (dev_control, dev_input_control),
- ) in enumerate(zip(power_mapping, control_mapping)):
- if antenna_type == "LBA" and pwr_attribute:
- # We should report the value of both mappings for each antenna (2 values),
- # but for now we pick only one to keep the resulting array of similar
- # dimension compared to HBA
- if dev_power > 0 and dev_control > 0:
- mapped_values[idx] = device_values[dev_power - 1][
- dev_input_power
- ]
- else:
- # Insert the two values in the mapped array
- if dev_power > 0:
- mapped_values[idx][0] = device_values[dev_power - 1][
- dev_input_power
- ]
- if dev_control > 0:
- mapped_values[idx][1] = device_values[dev_control - 1][
- dev_input_control
- ]
-
- return mapped_values
-
- def _mapped_rw_values(
- self,
- set_values: List[any],
- default_values: List[any],
- value_mapping: List[any],
- antenna_type: str,
- pwr_attribute: bool,
- ):
if MappingKeys.CONTROL in list(self._mapping_dict.keys()):
mapped_values = []
for _ in range(self._num_devices):
@@ -258,27 +176,26 @@ class AntennaMapper:
):
mapped_values[dev_conn] = set_values[idx]
- if len(value_mapping) == 2:
- [power_mapping, control_mapping] = value_mapping
- for idx, (
- (dev_power, dev_input_power),
- (dev_control, dev_input_control),
- ) in enumerate(zip(power_mapping, control_mapping)):
- if antenna_type == "LBA" and pwr_attribute:
- if dev_power > 0 and dev_control > 0:
- mapped_values[dev_power - 1, dev_input_power] = set_values[idx]
- mapped_values[dev_control - 1, dev_input_control] = set_values[
- idx
- ]
- else:
- if dev_power > 0:
- mapped_values[dev_power - 1, dev_input_power] = set_values[idx][
- 0
- ]
- if dev_control > 0:
- mapped_values[dev_control - 1, dev_input_control] = set_values[
- idx
- ][1]
+ return self._reshape_rw_values(mapped_attribute, mapped_values)
+
+ def _reshape_rw_values(self, mapped_attribute: str, mapped_values: numpy.ndarray):
+ """Attributes which need to be reshaped with removing a part of their duplicated
+ values because Device original dimension < Antennafield mapped dimension"""
+ if mapped_attribute in self._empty_attributes_out:
+ mapped_values = numpy.reshape(
+ mapped_values,
+ (self._num_devices,) + self._empty_attributes_out[mapped_attribute],
+ )
+ # Remove second (last) dimension
+ mapped_values = mapped_values[:, :, 0]
+
+ # Only RECV mapping may have more than one devices by now
+ if mapped_attribute in self._reshape_attributes_out:
+ mapped_values = numpy.reshape(
+ mapped_values,
+ (self._num_devices,) + self._reshape_attributes_out[mapped_attribute],
+ )
+
return mapped_values
def _init_reshape_out(self, mapped_attributes: List[str], device_type: str):
@@ -331,7 +248,7 @@ class AntennaToSdpMapper(AntennaMapper):
] = AntennaToSdpMapper._VALUE_MAP_NONE_16
-class AntennaToRecvMapper(AntennaMapper):
+class AntennaToRecvMapper(ABC, AntennaMapper):
"""Class that sets a mapping between RECVs and AntennaField attributes"""
_VALUE_MAP_NONE_96 = numpy.full(MAX_ANTENNA, None)
@@ -343,9 +260,8 @@ class AntennaToRecvMapper(AntennaMapper):
mapping_dict: Dict[str, numpy.ndarray],
mapped_dimensions: Dict[str, tuple],
number_of_receivers: Optional[int] = 1,
- antenna_type: str = "HBA",
):
- super().__init__(mapping_dict, number_of_receivers, antenna_type)
+ super().__init__(mapping_dict, number_of_receivers)
self._power_mapping = self._mapping_dict[MappingKeys.POWER]
self._control_mapping = self._mapping_dict[MappingKeys.CONTROL]
@@ -363,8 +279,6 @@ class AntennaToRecvMapper(AntennaMapper):
self._mapped_attributes = list(mapped_dimensions)
- self._value_mapper = self._init_value_mapper(antenna_type)
-
self._default_value_mapping_read = {
"ANT_mask_RW": value_map_ant_bool,
"HBAT_BF_delay_steps_R": value_map_ant_32_int,
@@ -403,6 +317,16 @@ class AntennaToRecvMapper(AntennaMapper):
mapped_dimensions
)
+ self._double_mapping = [
+ "RCU_attenuator_dB_R",
+ "RCU_attenuator_dB_RW",
+ "RCU_band_select_R",
+ "RCU_band_select_RW",
+ "RCU_PCB_ID_R",
+ "RCU_PCB_version_R",
+ ]
+ self._pwr_mapping = ["RCU_PWR_ANT_on_R", "RCU_PWR_ANT_on_RW"]
+
# Dict of reshaped mapped attribute dimensions with special cases
self._reshape_attributes_in = mapped_dimensions
self._reshape_attributes_in["RCU_attenuator_dB_R"] = ((MAX_ANTENNA * N_pol),)
@@ -436,8 +360,16 @@ class AntennaToRecvMapper(AntennaMapper):
"RCU_attenuator_dB_RW": (N_rcu * N_rcu_inp, N_pol),
}
- def _init_masked_value_write(self, mapped_dimensions: dict):
- """Create the masked value mapping dictionary for write attributes"""
+ def _init_masked_value_write(
+ self, mapped_dimensions: Dict[str, tuple]
+ ) -> Dict[str, numpy.ndarray]:
+ """Create the masked value mapping dictionary for write attributes
+
+ :param mapped_dimensions: a dictionary[str,tuple] which maps each
+ attribute with its RECV dimension
+ :return a dictionary[str, numpy.ndarray] which maps each attribute with
+ its AntennaField masked array dimension
+ """
special_cases = ["RCU_PCB_ID_R", "RCU_PCB_version_R"]
masked_write = {}
for attr, dim in mapped_dimensions.items():
@@ -450,32 +382,122 @@ class AntennaToRecvMapper(AntennaMapper):
masked_write[attr] = AntennaToRecvMapper._VALUE_MAP_NONE_96_32
return masked_write
- def _init_value_mapper(self, antenna_type: str):
- """Create the value mapping dictionary"""
- value_mapper = {}
- self._double_mapping = [
- "RCU_attenuator_dB_R",
- "RCU_attenuator_dB_RW",
- "RCU_band_select_R",
- "RCU_band_select_RW",
- "RCU_PCB_ID_R",
- "RCU_PCB_version_R",
- ]
- self._pwr_mapping = ["RCU_PWR_ANT_on_R", "RCU_PWR_ANT_on_RW"]
- for attr in self._mapped_attributes:
- if attr in self._double_mapping:
- value_mapper[attr] = [self._power_mapping, self._control_mapping]
- elif attr in self._pwr_mapping:
- if antenna_type == "LBA":
- value_mapper[attr] = [self._power_mapping, self._control_mapping]
- elif antenna_type == "HBA":
- value_mapper[attr] = [self._power_mapping]
- else:
- value_mapper[attr] = [self._control_mapping]
- return value_mapper
+ def map_read(
+ self, mapped_attribute: str, device_values: List[any]
+ ) -> numpy.ndarray:
+ """Perform a mapped read for the attribute using the device_results
+
+ :param mapped_attribute: attribute identifier as present in
+ py:attribute:`~_default_value_mapping_read`
+ :param device_values: Results as gathered by appending attributes from device
+ :return: device_results as mapped given attribute dimensions and control mapping
+ """
+ if len(self._value_mapper[mapped_attribute]) == 1:
+ return super().map_read(mapped_attribute, device_values)
+ if len(self._value_mapper[mapped_attribute]) == 2:
+ return self.map_read_double(mapped_attribute, device_values)
+ raise ValueError(
+ f"Mapping dims must be <=2, while it is {len(self._value_mapper[mapped_attribute])}"
+ )
+
+ def map_write(self, mapped_attribute: str, set_values: List[any]) -> numpy.ndarray:
+ """Perform a mapped write for the attribute using the set_values
+
+ :param mapped_attribute: attribute identifier as present in
+ py:attribute:`~_default_value_mapping_write`
+ :param set_values: The values to be set for the specified attribute
+ :return: set_values as mapped given attribute dimensions and control mapping
+ """
+ if len(self._value_mapper[mapped_attribute]) == 1:
+ return super().map_write(mapped_attribute, set_values)
+ if len(self._value_mapper[mapped_attribute]) == 2:
+ return self.map_write_double(mapped_attribute, set_values)
+ raise ValueError(
+ f"Mapping dims must be <=2, while it is {len(self._value_mapper[mapped_attribute])}"
+ )
+ def map_read_double(
+ self, mapped_attribute: str, device_values: List[any]
+ ) -> numpy.ndarray:
+ """Perform a mapped read for the attribute using the device_results,
+ only when the mapping is based on TWO maps (CTRL and PWR)
-class LBAToRecvMapper(AntennaToRecvMapper):
+ :param mapped_attribute: attribute identifier as present in
+ py:attribute:`~_default_value_mapping_read`
+ :param device_values: Results as gathered by appending attributes from device
+ :return: mapped_values as a numpy array containing the mapped values for the read
+ operation based on both CTRL and PWR mapping
+ """
+ default_values = self._default_value_mapping_read[mapped_attribute]
+ device_values = super()._reshape_r_values(mapped_attribute, device_values)
+ value_mapping = self._value_mapper[mapped_attribute]
+ mapped_values = numpy.array(default_values)
+ return self._mapped_r_values(
+ value_mapping, mapped_values, mapped_attribute, device_values
+ )
+
+ def map_write_double(
+ self, mapped_attribute: str, set_values: List[any]
+ ) -> numpy.ndarray:
+ """Perform a mapped write for the attribute using the set_values
+ only when the mapping is based on TWO maps (CTRL and PWR)
+
+ :param mapped_attribute: attribute identifier as present in
+ py:attribute:`~_default_value_mapping_read`
+ :param set_values: The values to be set for the specified attribute
+ :return: mapped_values as a numpy array containing the mapped values for the write
+ operation based on both CTRL and PWR mapping
+ """
+ default_values = self._masked_value_mapping_write[mapped_attribute]
+ mapped_values = []
+ for _ in range(self._num_devices):
+ mapped_values.append(default_values)
+ mapped_values = numpy.array(mapped_values)
+ mapped_values = self._mapped_rw_values(
+ mapped_values, mapped_attribute, set_values
+ )
+ return super()._reshape_rw_values(mapped_attribute, mapped_values)
+
+ @abstractmethod
+ def _mapped_r_values(
+ self,
+ value_mapping: List[any],
+ mapped_values: numpy.ndarray,
+ mapped_attribute: str,
+ device_values: List[any],
+ ) -> numpy.ndarray:
+ """Insert in the mapped values array the read attribute values following
+ the rules of the specific mapping
+
+ :param value_mapping: list of the available mappings (e.g. PWR and CTRL)
+ :param mapped_values: array with default values to be updated with the mapped ones
+ :param mapped_attribute: attribute name to be mapped
+ :param device_values: results as gathered by appending attribute values from device
+ :return: mapped_values as a numpy array containing the mapped values for the read
+ operation
+ """
+ raise NotImplementedError("Subclasses must implement this method")
+
+ @abstractmethod
+ def _mapped_rw_values(
+ self,
+ mapped_values: numpy.ndarray,
+ mapped_attribute: str,
+ set_values: List[any],
+ ) -> numpy.ndarray:
+ """Insert in the mapped values array the read-write attribute values following
+ the rules of the specific mapping
+
+ :param mapped_values: array with default values to be updated with the mapped ones
+ :param mapped_attribute: attribute name to be mapped
+ :param set_values: the values to be set for the specified attribute
+ :return: mapped_values as a numpy array containing the mapped values for the write
+ operation
+ """
+ raise NotImplementedError("Subclasses must implement this method")
+
+
+class AFLToRecvMapper(AntennaToRecvMapper):
"""Class that sets a mapping between RECVs and LBA attributes"""
def __init__(
@@ -483,14 +505,104 @@ class LBAToRecvMapper(AntennaToRecvMapper):
mapping_dict: Dict[str, numpy.ndarray],
mapped_dimensions: Dict[str, tuple],
number_of_receivers: Optional[int] = 1,
- antenna_type="LBA",
):
super().__init__(
- mapping_dict, mapped_dimensions, number_of_receivers, antenna_type
+ mapping_dict,
+ mapped_dimensions,
+ number_of_receivers,
)
+ self._value_mapper = self._init_value_mapper()
-class HBAToRecvMapper(AntennaToRecvMapper):
+ def _init_value_mapper(self):
+ """Create the value mapping dictionary"""
+ value_mapper = {}
+ for attr in self._mapped_attributes:
+ if attr in self._double_mapping:
+ value_mapper[attr] = [self._power_mapping, self._control_mapping]
+ elif attr in self._pwr_mapping:
+ value_mapper[attr] = [self._power_mapping, self._control_mapping]
+ else:
+ value_mapper[attr] = [self._control_mapping]
+ return value_mapper
+
+ def _mapped_rw_values(
+ self,
+ mapped_values: numpy.ndarray,
+ mapped_attribute: str,
+ set_values: List[any],
+ ) -> numpy.ndarray:
+ """Insert in the mapped values array the read-write attribute values following
+ the rules of the specific mapping for AntennaField-Low
+
+ :param mapped_values: array with default values to be updated with the mapped ones
+ :param mapped_attribute: attribute name to be mapped
+ :param set_values: the values to be set for the specified attribute
+ :return: mapped_values as a numpy array containing the mapped values for the write
+ operation
+ """
+ [power_mapping, control_mapping] = self._value_mapper[mapped_attribute]
+ for idx, (
+ (dev_power, dev_input_power),
+ (dev_control, dev_input_control),
+ ) in enumerate(zip(power_mapping, control_mapping)):
+ if mapped_attribute in self._pwr_mapping:
+ if dev_power > 0 and dev_control > 0:
+ mapped_values[dev_power - 1, dev_input_power] = set_values[idx]
+ mapped_values[dev_control - 1, dev_input_control] = set_values[idx]
+ else:
+ if dev_power > 0:
+ mapped_values[dev_power - 1, dev_input_power] = set_values[idx][0]
+ if dev_control > 0:
+ mapped_values[dev_control - 1, dev_input_control] = set_values[idx][
+ 1
+ ]
+ return mapped_values
+
+ def _mapped_r_values(
+ self,
+ value_mapping: List[any],
+ mapped_values: numpy.ndarray,
+ mapped_attribute: str,
+ device_values: List[any],
+ ) -> numpy.ndarray:
+ """Insert in the mapped values array the read attribute values following
+ the rules of the specific mapping for the AntennaField-Low
+
+ :param value_mapping: list of the available mappings (e.g. PWR and CTRL)
+ :param mapped_values: array with default values to be updated with the mapped ones
+ :param mapped_attribute: attribute name to be mapped
+ :param device_values: results as gathered by appending attribute values from device
+ :return: mapped_values as a numpy array containing the mapped values for the read
+ operation
+ """
+ # Assuming mapper lists are always in the following order:
+ # [Power_Mapping, Control_Mapping]
+ [power_mapping, control_mapping] = value_mapping
+ for idx, (
+ (dev_power, dev_input_power),
+ (dev_control, dev_input_control),
+ ) in enumerate(zip(power_mapping, control_mapping)):
+ if mapped_attribute in self._pwr_mapping:
+ # We should report the value of both mappings for each antenna (2 values),
+ # but for now we pick only one to keep the resulting array of similar
+ # dimension compared to HBA
+ if dev_power > 0 and dev_control > 0:
+ mapped_values[idx] = device_values[dev_power - 1][dev_input_power]
+ else:
+ # Insert the two values in the mapped array
+ if dev_power > 0:
+ mapped_values[idx][0] = device_values[dev_power - 1][
+ dev_input_power
+ ]
+ if dev_control > 0:
+ mapped_values[idx][1] = device_values[dev_control - 1][
+ dev_input_control
+ ]
+ return mapped_values
+
+
+class AFHToRecvMapper(AntennaToRecvMapper):
"""Class that sets a mapping between RECVs and HBA attributes"""
def __init__(
@@ -498,14 +610,88 @@ class HBAToRecvMapper(AntennaToRecvMapper):
mapping_dict: Dict[str, numpy.ndarray],
mapped_dimensions: Dict[str, tuple],
number_of_receivers: Optional[int] = 1,
- antenna_type="HBA",
):
super().__init__(
- mapping_dict, mapped_dimensions, number_of_receivers, antenna_type
+ mapping_dict,
+ mapped_dimensions,
+ number_of_receivers,
)
+ self._value_mapper = self._init_value_mapper()
+
+ def _init_value_mapper(self):
+ """Create the value mapping dictionary"""
+ value_mapper = {}
+ for attr in self._mapped_attributes:
+ if attr in self._double_mapping:
+ value_mapper[attr] = [self._power_mapping, self._control_mapping]
+ elif attr in self._pwr_mapping:
+ value_mapper[attr] = [self._power_mapping]
+ else:
+ value_mapper[attr] = [self._control_mapping]
+ return value_mapper
+
+ def _mapped_rw_values(
+ self,
+ mapped_values: numpy.ndarray,
+ mapped_attribute: str,
+ set_values: List[any],
+ ) -> numpy.ndarray:
+ """Insert in the mapped values array the read-write attribute values following
+ the rules of the specific mapping for AntennaField-High
+
+ :param mapped_values: array with default values to be updated with the mapped ones
+ :param mapped_attribute: attribute name to be mapped
+ :param set_values: the values to be set for the specified attribute
+ :return: mapped_values as a numpy array containing the mapped values for the write
+ operation
+ """
+ [power_mapping, control_mapping] = self._value_mapper[mapped_attribute]
+ for idx, (
+ (dev_power, dev_input_power),
+ (dev_control, dev_input_control),
+ ) in enumerate(zip(power_mapping, control_mapping)):
+ if dev_power > 0:
+ mapped_values[dev_power - 1, dev_input_power] = set_values[idx][0]
+ if dev_control > 0:
+ mapped_values[dev_control - 1, dev_input_control] = set_values[idx][1]
+ return mapped_values
+
+ def _mapped_r_values(
+ self,
+ value_mapping: List[any],
+ mapped_values: numpy.ndarray,
+ mapped_attribute: str,
+ device_values: List[any],
+ ) -> numpy.ndarray:
+ """Insert in the mapped values array the read attribute values following
+ the rules of the specific mapping for the AntennaField-High
+
+ :param value_mapping: list of the available mappings (e.g. PWR and CTRL)
+ :param mapped_values: array with default values to be updated with the mapped ones
+ :param mapped_attribute: attribute name to be mapped
+ :param device_values: results as gathered by appending attribute values from device
+ :return: mapped_values as a numpy array containing the mapped values for the read
+ operation
+ """
+ # Assuming mapper lists are always in the following order:
+ # [Power_Mapping, Control_Mapping]
+ [power_mapping, control_mapping] = value_mapping
+ for idx, (
+ (dev_power, dev_input_power),
+ (dev_control, dev_input_control),
+ ) in enumerate(zip(power_mapping, control_mapping)):
+ # Insert the two values in the mapped array
+ if dev_power > 0:
+ mapped_values[idx][0] = device_values[dev_power - 1][dev_input_power]
+ if dev_control > 0:
+ mapped_values[idx][1] = device_values[dev_control - 1][
+ dev_input_control
+ ]
+ return mapped_values
+
-class RecvDeviceWalker(object):
+class RecvDeviceWalker:
"""Walks over child devices with the appropriate mask set to affect the hardware
under the parent's control.."""
diff --git a/tangostationcontrol/test/devices/antennafield/test_afh_device.py b/tangostationcontrol/test/devices/antennafield/test_afh_device.py
index 36aa5cf9011c0d746676c8fbdaf02cadbae36112..7b0603a484eae914346d4cb6ba271b259fc09b49 100644
--- a/tangostationcontrol/test/devices/antennafield/test_afh_device.py
+++ b/tangostationcontrol/test/devices/antennafield/test_afh_device.py
@@ -26,7 +26,7 @@ from tangostationcontrol.devices.base_device_classes.antennafield_device import
from tangostationcontrol.devices.antennafield.afh import AFH
from tangostationcontrol.devices.base_device_classes.mapper import (
MappingKeys,
- HBAToRecvMapper,
+ AFHToRecvMapper,
AntennaToSdpMapper,
)
@@ -150,7 +150,9 @@ class TestAntennaToSdpMapper(base.TestCase):
numpy.testing.assert_equal(expected, actual)
-class TestHBAToRecvMapper(base.TestCase):
+class TestAFHToRecvMapper(base.TestCase):
+ """Test class for AntennaField-HBA to RECV Mapper"""
+
# A mapping where Antennas are all not mapped to power RCUs
POWER_NOT_CONNECTED = [[-1, -1]] * DEFAULT_N_HBA_TILES
# A mapping where Antennas are all not mapped to control RCUs
@@ -174,7 +176,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[False] * MAX_ANTENNA,
@@ -190,7 +192,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[False, True, False] + [False, False, False] * (N_rcu - 1),
@@ -207,7 +209,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [[0] * MAX_ANTENNA, [0] * MAX_ANTENNA, [0] * MAX_ANTENNA]
expected = [[0, 0]] * DEFAULT_N_HBA_TILES
actual = mapper.map_read("RCU_band_select_RW", receiver_values)
@@ -218,7 +220,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[0] * N_rcu] * MAX_ANTENNA,
@@ -234,7 +236,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[2] * N_rcu, [1] * N_rcu] + [[0] * N_rcu] * (MAX_ANTENNA - 2),
@@ -253,7 +255,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[0] * N_rcu] * MAX_ANTENNA,
@@ -269,7 +271,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[2] * N_rcu, [1] * N_rcu] + [[0] * N_rcu] * (MAX_ANTENNA - 2),
@@ -288,7 +290,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False] * N_rcu] * MAX_ANTENNA,
@@ -304,7 +306,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False, True] * 16, [True, False] * 16]
@@ -324,7 +326,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False] * N_rcu] * MAX_ANTENNA,
@@ -340,7 +342,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False, True] * 16, [True, False] * 16]
@@ -360,7 +362,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False] * N_rcu] * MAX_ANTENNA,
@@ -376,7 +378,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False, True] * 16, [True, False] * 16]
@@ -396,7 +398,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False] * N_rcu] * MAX_ANTENNA,
@@ -412,7 +414,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False, True] * 16, [True, False] * 16]
@@ -432,7 +434,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False] * N_rcu] * MAX_ANTENNA,
@@ -448,7 +450,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False, True] * 16, [True, False] * 16]
@@ -468,7 +470,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False] * N_rcu] * MAX_ANTENNA,
@@ -484,7 +486,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
[[False, True] * 16, [True, False] * 16]
@@ -505,7 +507,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [list(range(32)), [0] * 32, [0] * 32]
expected = [[0] * 2] * DEFAULT_N_HBA_TILES
actual = mapper.map_read("RCU_PCB_ID_R", receiver_values)
@@ -518,7 +520,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_3_AND_2_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [list(range(32)), [0] * 32, [0] * 32]
expected = [[0, 3], [0, 2]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
actual = mapper.map_read("RCU_PCB_ID_R", receiver_values)
@@ -531,7 +533,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [list(range(32)), [0] * 32, [0] * 32]
expected = [[1, 0], [0, 0]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
actual = mapper.map_read("RCU_PCB_ID_R", receiver_values)
@@ -544,7 +546,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_3_AND_2_OF_RECV_1,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [list(range(32)), [0] * 32, [0] * 32]
expected = [[1, 3], [0, 2]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
actual = mapper.map_read("RCU_PCB_ID_R", receiver_values)
@@ -556,7 +558,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 3)
receiver_values = [
list(range(MAX_ANTENNA)),
[0] * MAX_ANTENNA,
@@ -573,7 +575,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
receiver_values = [list(range(MAX_ANTENNA))]
expected = [[0, 1], [0, 0]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
actual = mapper.map_read("RCU_attenuator_dB_R", receiver_values)
@@ -586,7 +588,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
receiver_values = [list(range(MAX_ANTENNA))]
expected = [[1, 0], [0, 0]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
actual = mapper.map_read("RCU_attenuator_dB_R", receiver_values)
@@ -599,7 +601,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_3_AND_2_OF_RECV_1,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
receiver_values = [list(range(MAX_ANTENNA))]
expected = [[1, 3], [0, 2]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
actual = mapper.map_read("RCU_attenuator_dB_R", receiver_values)
@@ -613,7 +615,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [None] * DEFAULT_N_HBA_TILES
expected = [[[None, None, None]] * N_rcu]
@@ -626,7 +628,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [None] * DEFAULT_N_HBA_TILES
expected = [[[None, None, None]] * N_rcu] * 2
@@ -638,7 +640,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [True, False] + [None] * (DEFAULT_N_HBA_TILES - 2)
expected = [[[False, True, None]] + [[None, None, None]] * (N_rcu - 1)]
@@ -650,7 +652,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [None] * DEFAULT_N_HBA_TILES
expected = [[[None, None, None]] * N_rcu]
@@ -662,7 +664,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [None] * DEFAULT_N_HBA_TILES
expected = [[[None, None, None]] * N_rcu] * 2
@@ -674,7 +676,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [1, 0] + [None] * (DEFAULT_N_HBA_TILES - 2)
expected = [[[0, 1, None]] + [[None, None, None]] * (N_rcu - 1)]
@@ -686,7 +688,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [None] * DEFAULT_N_HBA_TILES
expected = [[[None, None, None]] * N_rcu]
@@ -698,7 +700,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [None] * DEFAULT_N_HBA_TILES
expected = [[[None, None, None]] * N_rcu] * 2
@@ -710,7 +712,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[1, 1], [0, 0]] + [[None, None]] * (DEFAULT_N_HBA_TILES - 2)
expected = [[[0, 1, None]] + [[None, None, None]] * (N_rcu - 1)]
@@ -722,7 +724,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[1] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA]
@@ -734,7 +736,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [[1] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA, [[None] * N_rcu] * MAX_ANTENNA]
@@ -746,7 +748,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[1] * N_rcu, [2] * N_rcu] + [[None] * N_rcu] * (
DEFAULT_N_HBA_TILES - 2
@@ -760,7 +762,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[None] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA]
@@ -772,7 +774,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [[None] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA, [[None] * N_rcu] * MAX_ANTENNA]
@@ -784,7 +786,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[False, True] * 16, [True, False] * 16] + [[None] * N_rcu] * (
DEFAULT_N_HBA_TILES - 2
@@ -801,7 +803,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[None] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA]
@@ -813,7 +815,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [[None] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA, [[None] * N_rcu] * MAX_ANTENNA]
@@ -825,7 +827,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[False, True] * 16, [True, False] * 16] + [[None] * N_rcu] * (
DEFAULT_N_HBA_TILES - 2
@@ -842,7 +844,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[None] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA]
actual = mapper.map_write("HBAT_PWR_on_RW", set_values)
@@ -853,7 +855,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 2)
set_values = [[None] * N_rcu] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu] * MAX_ANTENNA, [[None] * N_rcu] * MAX_ANTENNA]
@@ -865,7 +867,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[False, True] * 16, [True, False] * 16] + [[None] * N_rcu] * (
DEFAULT_N_HBA_TILES - 2
@@ -883,7 +885,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[1, 1]] * DEFAULT_N_HBA_TILES
expected = [[[None] * N_rcu_inp] * N_rcu]
@@ -898,7 +900,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = [[1, 2], [3, 4]] + [[0, 0]] * (DEFAULT_N_HBA_TILES - 2)
expected = [[[4, 2, None]] + [[None] * N_rcu_inp] * (N_rcu - 1)]
@@ -913,7 +915,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = set_values = [[1, 2], [3, 4]] + [[0, 0]] * (
DEFAULT_N_HBA_TILES - 2
@@ -930,7 +932,7 @@ class TestHBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_HBA_0_AND_1_ON_RCU_3_AND_2_OF_RECV_1,
MappingKeys.POWER: self.POWER_HBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = HBAToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
+ mapper = AFHToRecvMapper(recv_mapping, RECV_MAPPED_ATTRS, 1)
set_values = set_values = [[1, 2], [3, 4]] + [[0, 0]] * (
DEFAULT_N_HBA_TILES - 2
diff --git a/tangostationcontrol/test/devices/antennafield/test_afl_device.py b/tangostationcontrol/test/devices/antennafield/test_afl_device.py
index de2b1e5db0083f0cfa697c3486b50b43ebc80028..b8a602d0f0320e243959455f0c13269dbdad4802 100644
--- a/tangostationcontrol/test/devices/antennafield/test_afl_device.py
+++ b/tangostationcontrol/test/devices/antennafield/test_afl_device.py
@@ -17,7 +17,7 @@ from tangostationcontrol.common.constants import (
from tangostationcontrol.devices.antennafield.afl import AFL
from tangostationcontrol.devices.base_device_classes.mapper import (
MappingKeys,
- LBAToRecvMapper,
+ AFLToRecvMapper,
)
logger = logging.getLogger()
@@ -26,8 +26,8 @@ RECV_MAPPED_ATTRS = AFL.get_mapped_dimensions("RECV")
DEFAULT_N_LBA = DEFAULT_N_HBA_TILES # 48 antennas
-class TestLBAToRecvMapper(base.TestCase):
- """Test class for AntennaToRecvMapper with LBA AntennaType"""
+class TestAFLToRecvMapper(base.TestCase):
+ """Test class for AntennaField-LBA to RECV Mapper"""
# A mapping where Antennas are all not mapped to power RCUs
POWER_NOT_CONNECTED = [[-1, -1]] * 2 * DEFAULT_N_LBA
@@ -51,7 +51,7 @@ class TestLBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = LBAToRecvMapper(
+ mapper = AFLToRecvMapper(
recv_mapping,
RECV_MAPPED_ATTRS,
1,
@@ -67,7 +67,7 @@ class TestLBAToRecvMapper(base.TestCase):
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_NOT_CONNECTED,
}
- mapper = LBAToRecvMapper(
+ mapper = AFLToRecvMapper(
recv_mapping,
RECV_MAPPED_ATTRS,
2,
@@ -78,14 +78,14 @@ class TestLBAToRecvMapper(base.TestCase):
actual = mapper.map_write("RCU_PWR_ANT_on_RW", set_values)
numpy.testing.assert_equal(expected, actual)
- def test_map_write_rcu_pwr_ant_on_hba_0_and_1_on_rcu_1_and_0_of_recv_1_and_no_ctrl(
+ def test_map_write_rcu_pwr_ant_on_lba_0_and_1_on_rcu_1_and_0_of_recv_1_and_no_ctrl(
self,
):
recv_mapping = {
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_LBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = LBAToRecvMapper(
+ mapper = AFLToRecvMapper(
recv_mapping,
RECV_MAPPED_ATTRS,
1,
@@ -96,12 +96,12 @@ class TestLBAToRecvMapper(base.TestCase):
actual = mapper.map_write("RCU_PWR_ANT_on_RW", set_values)
numpy.testing.assert_equal(expected, actual)
- def test_map_read_rcu_pwr_ant_on_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self):
+ def test_map_read_rcu_pwr_ant_on_lba_0_and_1_on_rcu_1_and_0_of_recv_1(self):
recv_mapping = {
MappingKeys.CONTROL: self.CONTROL_LBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
MappingKeys.POWER: self.POWER_LBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = LBAToRecvMapper(
+ mapper = AFLToRecvMapper(
recv_mapping,
RECV_MAPPED_ATTRS,
2,
@@ -112,12 +112,12 @@ class TestLBAToRecvMapper(base.TestCase):
actual = mapper.map_read("RCU_PWR_ANT_on_R", receiver_values)
numpy.testing.assert_equal(expected, actual)
- def test_map_write_rcu_pwr_ant_on_hba_0_and_1_on_rcu_1_and_0_of_recv_1(self):
+ def test_map_write_rcu_pwr_ant_on_lba_0_and_1_on_rcu_1_and_0_of_recv_1(self):
recv_mapping = {
MappingKeys.CONTROL: self.CONTROL_NOT_CONNECTED,
MappingKeys.POWER: self.POWER_LBA_0_AND_1_ON_RCU_1_AND_0_OF_RECV_1,
}
- mapper = LBAToRecvMapper(
+ mapper = AFLToRecvMapper(
recv_mapping,
RECV_MAPPED_ATTRS,
1,