ska_pst.testutils.dada
Submodule for providing utilities working with DADA files.
- 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