IO

blipss.io.read_blimpy_data

Utilities for loading blimpy-compatible filterbank and HDF5 data files

read_waterfall_file(file_path, max_memory_gb)

Load a .h5 or .fil file into a blimpy Waterfall object.

Parameters:

Name	Type	Description	Default
file_path	Path | str	Path to the .h5 or .fil data file to load	required
max_memory_gb	float	Maximum data size in GB permitted in memory	required

Returns:

Type	Description
Waterfall	Blimpy Waterfall object containing the data file contents

Source code in blipss/io/read_blimpy_data.py

def read_waterfall_file(file_path: Path | str, max_memory_gb: float) -> Waterfall:
    """
    Load a .h5 or .fil file into a blimpy Waterfall object.

    Args:
        file_path: Path to the .h5 or .fil data file to load
        max_memory_gb: Maximum data size in GB permitted in memory

    Returns:
        Blimpy Waterfall object containing the data file contents
    """
    return Waterfall(str(file_path), max_load=max_memory_gb)

blipss.io.read_candidates

Read utilities for FFA candidate detection CSV files.

read_candidates_csv(csv_path)

Read FFA candidate detections previously written by write_candidates_csv.

Parameters:

Name	Type	Description	Default
csv_path	Path	Path to a candidate CSV file with FFA_CANDIDATE_CSV_COLUMNS columns.	required

Returns:

Type	Description
NDArray[intp]	Tuple of (channels, radiofreqs_MHz, phase_bins, boxcar_widths, periods, snrs, flags)
NDArray[floating]	arrays, one entry per candidate row in file order.

Raises:

Type	Description
ValueError	When the file header is missing one of FFA_CANDIDATE_CSV_COLUMNS.

Source code in blipss/io/read_candidates.py

def read_candidates_csv(
    csv_path: Path,
) -> tuple[
    npt.NDArray[np.intp],
    npt.NDArray[np.floating],
    npt.NDArray[np.uint],
    npt.NDArray[np.uint],
    npt.NDArray[np.floating],
    npt.NDArray[np.floating],
    npt.NDArray[np.str_],
]:
    """
    Read FFA candidate detections previously written by ``write_candidates_csv``.

    Args:
        csv_path: Path to a candidate CSV file with ``FFA_CANDIDATE_CSV_COLUMNS`` columns.

    Returns:
        Tuple of (channels, radiofreqs_MHz, phase_bins, boxcar_widths, periods, snrs, flags)
        arrays, one entry per candidate row in file order.

    Raises:
        ValueError: When the file header is missing one of ``FFA_CANDIDATE_CSV_COLUMNS``.
    """
    df = pd.read_csv(csv_path)
    missing = [name for name in FFA_CANDIDATE_CSV_COLUMNS if name not in df.columns]
    if missing:
        raise ValueError(f"Missing required column(s) in {csv_path}: {missing}")
    channels, radiofreqs, phase_bins, boxcar_widths, periods, snrs, flags = (
        df[name].to_numpy(dtype=dtype)
        for name, (_, dtype) in zip(FFA_CANDIDATE_CSV_COLUMNS, _CANDIDATE_DTYPE, strict=True)
    )
    return channels, radiofreqs, phase_bins, boxcar_widths, periods, snrs, flags

blipss.io.read_compared_candidates

Read utilities for cross-file candidate comparison output files.

read_compared_candidates_csv(csv_path)

Read cross-file candidate comparison results previously written by write_compared_candidates_csv.

Parameters:

Name	Type	Description	Default
csv_path	Path	Path to a comparison CSV file with COMPARE_CANDS_CSV_COLUMNS columns.	required

Returns:

Type	Description
NDArray[intp]	Tuple of (channels, radiofreqs_MHz, phase_bins, boxcar_widths, periods, snrs, codes)
NDArray[floating]	arrays, one entry per candidate row in file order.

Raises:

Type	Description
ValueError	When the file is missing one of COMPARE_CANDS_CSV_COLUMNS.

Source code in blipss/io/read_compared_candidates.py

def read_compared_candidates_csv(
    csv_path: Path,
) -> tuple[
    npt.NDArray[np.intp],
    npt.NDArray[np.floating],
    npt.NDArray[np.uint],
    npt.NDArray[np.uint],
    npt.NDArray[np.floating],
    npt.NDArray[np.floating],
    npt.NDArray[np.str_],
]:
    """
    Read cross-file candidate comparison results previously written by ``write_compared_candidates_csv``.

    Args:
        csv_path: Path to a comparison CSV file with ``COMPARE_CANDS_CSV_COLUMNS`` columns.

    Returns:
        Tuple of (channels, radiofreqs_MHz, phase_bins, boxcar_widths, periods, snrs, codes)
        arrays, one entry per candidate row in file order.

    Raises:
        ValueError: When the file is missing one of ``COMPARE_CANDS_CSV_COLUMNS``.
    """
    df = pd.read_csv(csv_path, usecols=COMPARE_CANDS_CSV_COLUMNS, dtype={"Code": "string"})
    return (
        df["Channel"].to_numpy(dtype=np.intp),
        df["Radio frequency (MHz)"].to_numpy(dtype=np.float64),
        df["Bins"].to_numpy(dtype=np.uint),
        df["Best width"].to_numpy(dtype=np.uint),
        df["Period (s)"].to_numpy(dtype=np.float64),
        df["S/N"].to_numpy(dtype=np.float64),
        df["Code"].to_numpy(dtype=str),
    )

blipss.io.read_yaml_config

Routines for reading YAML config files

load_yaml_config(config_path)

Read a YAML config file and return its contents as a dictionary.

Parameters:

Name	Type	Description	Default
config_path	Path	Path to the YAML config file	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing the parsed YAML config

Raises:

Type	Description
FileNotFoundError	When config_path does not point to an existing file
TypeError	When the file is empty or its top-level structure is not a mapping

Source code in blipss/io/read_yaml_config.py

def load_yaml_config(config_path: Path) -> dict[str, Any]:
    """
    Read a YAML config file and return its contents as a dictionary.

    Args:
        config_path: Path to the YAML config file

    Returns:
        Dictionary containing the parsed YAML config

    Raises:
        FileNotFoundError: When config_path does not point to an existing file
        TypeError: When the file is empty or its top-level structure is not a mapping
    """
    if not config_path.is_file():
        raise FileNotFoundError(f"Config file not found: {config_path}")

    with config_path.open("r") as f:
        config = yaml.safe_load(f)

    if not isinstance(config, dict):
        raise TypeError(f"Expected a YAML mapping at the top level, got {type(config).__name__}")

    return config

blipss.io.write_candidates

Write utilities for FFA candidate detection output files.

write_candidates_csv(output_path, freqs_MHz, cand_channels, cand_periods, cand_snrs, cand_phase_bins, cand_boxcar_widths, cand_flags, start_ch)

Write FFA candidate detections to a CSV file sorted by descending S/N.

Rows are ordered from highest to lowest S/N. Column layout follows FFA_CANDIDATE_CSV_COLUMNS.

Parameters:

Name	Type	Description	Default
output_path	Path	Destination path for the output CSV file.	required
freqs_MHz	NDArray[floating]	Radio frequencies in MHz for all channels in the processed band.	required
cand_channels	NDArray[intp]	Absolute channel indices of each candidate detection.	required
cand_periods	NDArray[floating]	Best-fit periods in seconds for each candidate.	required
cand_snrs	NDArray[floating]	Peak signal-to-noise ratios for each candidate.	required
cand_phase_bins	NDArray[uint]	Number of phase bins in the folded profile for each candidate.	required
cand_boxcar_widths	NDArray[uint]	Best-fit boxcar widths in phase bins for each candidate.	required
cand_flags	NDArray[str_]	Harmonic classification label for each candidate.	required
start_ch	int	Absolute channel index of the first channel in the processed sub-band, used to map cand_channels to entries in freqs_MHz.	required

Source code in blipss/io/write_candidates.py

def write_candidates_csv(
    output_path: Path,
    freqs_MHz: npt.NDArray[np.floating],
    cand_channels: npt.NDArray[np.intp],
    cand_periods: npt.NDArray[np.floating],
    cand_snrs: npt.NDArray[np.floating],
    cand_phase_bins: npt.NDArray[np.uint],
    cand_boxcar_widths: npt.NDArray[np.uint],
    cand_flags: npt.NDArray[np.str_],
    start_ch: int,
) -> None:
    """
    Write FFA candidate detections to a CSV file sorted by descending S/N.

    Rows are ordered from highest to lowest S/N. Column layout follows
    ``FFA_CANDIDATE_CSV_COLUMNS``.

    Args:
        output_path: Destination path for the output CSV file.
        freqs_MHz: Radio frequencies in MHz for all channels in the processed band.
        cand_channels: Absolute channel indices of each candidate detection.
        cand_periods: Best-fit periods in seconds for each candidate.
        cand_snrs: Peak signal-to-noise ratios for each candidate.
        cand_phase_bins: Number of phase bins in the folded profile for each candidate.
        cand_boxcar_widths: Best-fit boxcar widths in phase bins for each candidate.
        cand_flags: Harmonic classification label for each candidate.
        start_ch: Absolute channel index of the first channel in the processed sub-band,
            used to map ``cand_channels`` to entries in ``freqs_MHz``.
    """
    cand_radiofreqs = freqs_MHz[cand_channels - start_ch]
    sort_desc_idx = np.argsort(cand_snrs)[::-1]
    with output_path.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(FFA_CANDIDATE_CSV_COLUMNS)
        for row in zip(
            cand_channels[sort_desc_idx],
            cand_radiofreqs[sort_desc_idx],
            cand_phase_bins[sort_desc_idx],
            cand_boxcar_widths[sort_desc_idx],
            cand_periods[sort_desc_idx],
            cand_snrs[sort_desc_idx],
            cand_flags[sort_desc_idx],
            strict=False,
        ):
            writer.writerow(row)

blipss.io.write_compared_candidates

Write utilities for cross-file candidate comparison output files.

write_compared_candidates_csv(output_path, channels, radiofreqs, phase_bins, boxcar_widths, periods, snrs, codes)

Write cross-file candidate comparison results to a CSV file.

Rows are written in the order given. Column layout follows COMPARE_CANDS_CSV_COLUMNS.

Parameters:

Name	Type	Description	Default
output_path	Path	Destination path for the output CSV file.	required
channels	NDArray[intp]	Spectral channel index of each candidate.	required
radiofreqs	NDArray[floating]	Radio frequency (MHz) of each candidate.	required
phase_bins	NDArray[uint]	Number of phase bins in the folded profile for each candidate.	required
boxcar_widths	NDArray[uint]	Best-fit boxcar widths in phase bins for each candidate.	required
periods	NDArray[floating]	Best-fit periods in seconds for each candidate.	required
snrs	NDArray[floating]	Peak signal-to-noise ratios for each candidate.	required
codes	NDArray[str_]	Per-file binary detection code string for each candidate.	required

Source code in blipss/io/write_compared_candidates.py

def write_compared_candidates_csv(
    output_path: Path,
    channels: npt.NDArray[np.intp],
    radiofreqs: npt.NDArray[np.floating],
    phase_bins: npt.NDArray[np.uint],
    boxcar_widths: npt.NDArray[np.uint],
    periods: npt.NDArray[np.floating],
    snrs: npt.NDArray[np.floating],
    codes: npt.NDArray[np.str_],
) -> None:
    """
    Write cross-file candidate comparison results to a CSV file.

    Rows are written in the order given. Column layout follows ``COMPARE_CANDS_CSV_COLUMNS``.

    Args:
        output_path: Destination path for the output CSV file.
        channels: Spectral channel index of each candidate.
        radiofreqs: Radio frequency (MHz) of each candidate.
        phase_bins: Number of phase bins in the folded profile for each candidate.
        boxcar_widths: Best-fit boxcar widths in phase bins for each candidate.
        periods: Best-fit periods in seconds for each candidate.
        snrs: Peak signal-to-noise ratios for each candidate.
        codes: Per-file binary detection code string for each candidate.
    """
    with output_path.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(COMPARE_CANDS_CSV_COLUMNS)
        for row in zip(channels, radiofreqs, phase_bins, boxcar_widths, periods, snrs, codes, strict=False):
            writer.writerow(row)

blipss.io.write_filterbank

Filterbank write utilities for both synthetic and real-data pipelines.

Two write paths are provided:

• write_filterbank: writes a raw numpy array produced by the simulate_data pipeline directly to a .fil file by serialising a sigproc header followed by the raw float32 samples.

• write_waterfall: writes a blimpy Waterfall object produced by the inject_signal pipeline to either a .fil or .h5 file, delegating to the appropriate blimpy serialiser based on the output file extension.

Helper utilities

build_sigproc_header Constructs the sigproc header dict required by write_filterbank.

build_sigproc_header(sim, header_params)

Construct the sigproc header dictionary from simulation and metadata parameters.

This is a pre-write step for the simulate_data pipeline. The resulting dict is passed directly to write_filterbank.

Parameters:

Name	Type	Description	Default
sim	SimulationProperties	Filterbank dimension and frequency/time axis parameters.	required
header_params	OptionalHeaderParameters	Optional observational metadata (source name, start MJD).	required

Returns:

Type	Description
dict[str, Any]	Dictionary of sigproc header key-value pairs compatible with
dict[str, Any]	blimpy.io.sigproc.generate_sigproc_header.

Source code in blipss/io/write_filterbank.py

def build_sigproc_header(
    sim: SimulationProperties,
    header_params: OptionalHeaderParameters,
) -> dict[str, Any]:
    """
    Construct the sigproc header dictionary from simulation and metadata parameters.

    This is a pre-write step for the *simulate_data* pipeline. The resulting
    dict is passed directly to ``write_filterbank``.

    Args:
        sim: Filterbank dimension and frequency/time axis parameters.
        header_params: Optional observational metadata (source name, start MJD).

    Returns:
        Dictionary of sigproc header key-value pairs compatible with
        ``blimpy.io.sigproc.generate_sigproc_header``.
    """
    return {
        "machine_id": SIGPROC_MACHINE_ID,
        "telescope_id": SIGPROC_TELESCOPE_ID,
        "data_type": SIGPROC_DATA_TYPE,
        "nbits": SIGPROC_N_BITS,
        "nifs": SIGPROC_N_IFS,
        "fch1": sim.fch1,
        "foff": sim.foff,
        "tsamp": sim.t_samp,
        "nchans": sim.n_channels,
        "nsamples": sim.n_samples,
        "source_name": header_params.source_name,
        "tstart": header_params.tstart,
    }

write_filterbank(data, header, output_dir, basename)

Write a numpy data array and a sigproc header to a .fil filterbank file.

Used by the simulate_data pipeline, where the full data array is generated in memory. The array is serialised as contiguous float32 samples immediately after the binary sigproc header.

Parameters:

Name	Type	Description	Default
data	NDArray[floating]	Array of shape (n_samples, 1, n_channels) in sigproc layout, as returned by blipss.core.simulate_data.reshape_for_sigproc.	required
header	dict[str, Any]	Sigproc header dictionary, as returned by build_sigproc_header.	required
output_dir	Path	Directory in which the output file is created; created automatically if it does not exist.	required
basename	str	Filename stem; the .fil extension is appended automatically.	required

Source code in blipss/io/write_filterbank.py

def write_filterbank(
    data: npt.NDArray[np.floating],
    header: dict[str, Any],
    output_dir: Path,
    basename: str,
) -> None:
    """
    Write a numpy data array and a sigproc header to a ``.fil`` filterbank file.

    Used by the *simulate_data* pipeline, where the full data array is generated
    in memory. The array is serialised as contiguous float32 samples immediately
    after the binary sigproc header.

    Args:
        data: Array of shape ``(n_samples, 1, n_channels)`` in sigproc layout,
            as returned by ``blipss.core.simulate_data.reshape_for_sigproc``.
        header: Sigproc header dictionary, as returned by ``build_sigproc_header``.
        output_dir: Directory in which the output file is created; created
            automatically if it does not exist.
        basename: Filename stem; the ``.fil`` extension is appended automatically.
    """
    ensure_path_exists(output_dir)
    carrier = _HeaderCarrier()
    carrier.header = header
    output_path = output_dir / f"{basename}.fil"
    with open(output_path, "wb") as file_handle:
        file_handle.write(generate_sigproc_header(carrier))
        data.ravel().astype(np.float32, copy=False).tofile(file_handle)

write_waterfall(wat, output_path)

Write a blimpy Waterfall object to disk, dispatching on file extension.

Used by the inject_signal pipeline, where signal injection modifies an existing Waterfall object loaded from a real-data filterbank. The original header is preserved intact; only wat.data is expected to have been updated before calling this function.

Parameters:

Name	Type	Description	Default
wat	Waterfall	Blimpy Waterfall object to serialise.	required
output_path	Path	Full output file path including a supported extension (.fil or .h5 / .hdf5).	required

Raises:

Type	Description
ValueError	When the file extension is not in FILTERBANK_EXTENSIONS.

Source code in blipss/io/write_filterbank.py

def write_waterfall(wat: Waterfall, output_path: Path) -> None:
    """
    Write a blimpy ``Waterfall`` object to disk, dispatching on file extension.

    Used by the *inject_signal* pipeline, where signal injection modifies an
    existing ``Waterfall`` object loaded from a real-data filterbank. The
    original header is preserved intact; only ``wat.data`` is expected to have
    been updated before calling this function.

    Args:
        wat: Blimpy ``Waterfall`` object to serialise.
        output_path: Full output file path including a supported extension
            (``.fil`` or ``.h5`` / ``.hdf5``).

    Raises:
        ValueError: When the file extension is not in ``FILTERBANK_EXTENSIONS``.
    """
    suffix = output_path.suffix
    if suffix not in FILTERBANK_EXTENSIONS:
        raise ValueError(f"Unsupported output extension {suffix!r}. Expected one of {sorted(FILTERBANK_EXTENSIONS)}.")
    if suffix == ".fil":
        wat.write_to_fil(str(output_path))
    else:
        wat.write_to_hdf5(str(output_path))