ska_pst.testutils.dada
Submodule for providing utilities working with DADA files.
- class ska_pst.testutils.dada.DadaFileReader(file: Path, logger: Optional[Logger] = None)[source]
Class that can be used to read a PSR DADA file.
- property data_size: int
Get the size of the data.
- property file_name: str
Get name of file.
- property file_number: int
Get the FILE_NUMBER value from header.
- property file_size: int
Get size of file in bytes.
- property header: Dict[str, str]
Get header for file.
- property obs_offset: int
Get the OBS_OFFSET value.
- property scan_id: int
Get the SCAN_ID value from header.
- class ska_pst.testutils.dada.DspDataAnalyser(scan_config: dict, local_path: Path, staging_path: Path, dlm_path: Path, subsystem_id: str, scan_id: int, eb_id: Optional[str] = None, logger: Optional[Logger] = None)[source]
Class for analysing files generated by DSP.CORE.
- assert_expected_file_count(file_type: str, expected_num_files: int) None[source]
Assert that the number of files matches the expected count.
- Parameters
file_type – Type of file, either ‘data’ or ‘weights’.
expected_num_files – Number of files expected based on metadata.
- assert_file_data_sizes(file_type: str, expected_num_files: int, bytes_per_file: int) None[source]
Assert that each file has the correct data size based on its position.
- Parameters
file_type – Type of file, either ‘data’ or ‘weights’.
expected_num_files – Number of expected files.
bytes_per_file – Expected size of each file in bytes.
- assert_full_file_integrity(scanlen: float, file_type: str, bytes_per_second: float, bytes_per_file: int, expected_num_files: int, **kwargs: Any) None[source]
Perform all integrity checks on the given file type.
- Parameters
scanlen (float) – expected scan duration in seconds.
file_type (str) – Type of file to validate (e.g., ‘data’, ‘meta’).
bytes_per_second (float) – expected throughput in bytes per second.
bytes_per_file (int) – expected size of each file in bytes.
expected_num_files (int) – number of files expected.
- assert_obs_offsets(file_type: str, expected_num_files: int, bytes_per_file: int) None[source]
Assert that the obs_offset for each file matches its expected position.
- Parameters
file_type – Type of file, either ‘data’ or ‘weights’.
expected_num_files – Number of expected files.
bytes_per_file – Expected size of each file in bytes.
- assert_scan_duration_match(file_type: str, scanlen: float, bytes_per_second: float) None[source]
Assert that the total data duration matches the expected scan length.
- Parameters
file_type – Type of file, either ‘data’ or ‘weights’.
scanlen – Expected scan length in seconds.
bytes_per_second – Expected bytes per second for the file type.
- check_contiguous_files(scanlen: float, calculated_resources: dict, file_type: str) None[source]
Check that the data or weight files generated during the scan are contiguous and consistent.
- Parameters
scanlen – Length of the scan in seconds.
calculated_resources – Dictionary of calculated UDP generator resources.
file_type – The type of files to analyse, either ‘data’ or ‘weights’.
- check_dada_files_exist(dsp_subpath: str) None[source]
Assert that DADA files exist for a given subpath.
This will check that *.dada files exist in the given dsp_subpath. Files are expected to be found at one of the following locations: <local_path> / <eb_id> / <subsystem_id> / <scan_id> / dsp_subpath <staging_path> / <eb_id>_<subsystem_id>_<scan_id> / dsp_subpath <dlm_path> / <eb_id>_<subsystem_id>_<scan_id> / dsp_subpath
- check_sinusoid_frequency(expected_frequency: float, eps: float = 0.1) None[source]
Analyse DSP artefacts.
This will find data and weights file for the given scan and assert that the frequency from data is within eps of the expected frequency.
- check_weights_contain_dropped_packets(expected_dropped_packets: List[int]) None[source]
Analyse DSP weights files.
This will parse all weights files in the local, staging or dlm scan paths and check that the specified packets are flagged as dropped.
- check_weights_zeroed(expected_zeroed_packets: List[int]) None[source]
Analyse DSP weights files for zero weights.
This will parse all weights files in the local, staging or dlm scan paths and check that the specified packets have the weights being zeroed out.
- get_all_dada_files(dsp_subpath: str) List[Path][source]
Return a list of all dada files in the local, staging or dlm scan paths and the dsp_subpath.
- Parameters
dsp_sub_path – subdirectory to look for dada files, i.e. data or weights
- class ska_pst.testutils.dada.ScanFileMetadata(*, name: str, obs_offset: int, file_number: int, data_size: int, file_size: int, scan_id: int)[source]
A data class to store the metadata of a scan file.
- data_size: int
- file_number: int
- file_size: int
- name: str
- obs_offset: int
- scan_id: int
- class ska_pst.testutils.dada.WeightsFileReader(file: Path, unpack_scales: bool = True, unpack_weights: bool = True, logger: Optional[Logger] = None)[source]
Class that can be used to read a Weights PSRDADA file generated by ska_pst_dsp_disk.
- property dropped_packets: numpy.ndarray
Return a list of the dropped packets by inspecting NaNs in the scales.
- property packet_offset: int
Get the package offset for current file.
This converts the obs_offset a packet offset by dividing the value by the weights_packet_stride. This will assert that the obs_offset is a multiple of weights_packet_stride
- property packets_weights: nptyping.NDArray.(typing.Any, nptyping.UShort)
Get the unpacked weights per packet.
The
self.weightsproperty returns the weights in heaps but this property will use packets in the first dimension. The shape of the resulting array is:(num_packets, nchan_per_packet)- Returns
the weights grouped by packets
- Return type
WeightsType
- property scales: nptyping.NDArray.(typing.Any, nptyping.Single)
Return the unpacked scales.
- property weights: nptyping.NDArray.(typing.Any, nptyping.UShort)
Return the unpacked weights.
- property zeroed_packets: numpy.ndarray
Return a list of the zeroed out packets by inspecting the weights.