CLI

blipss.cli.compare_cands

Compare candidate periods across N candidate CSV files and output one N-digit binary code per candidate.

In the binary code, "1" denotes detection and "0" denotes non-detection. Candidate detection in file i is denoted by "1" in the i-th position of the code (read left to right).

Example usage

python -m blipss.cli.compare_cands --config config/compare_cands.yaml

main(config)

Compare candidate periods across N files and output one N-digit binary code per candidate.

Source code in blipss/cli/compare_cands.py

@app.command()
def main(config: Annotated[Path, typer.Option("--config", help="Path to YAML config file")]) -> None:
    """Compare candidate periods across N files and output one N-digit binary code per candidate."""
    t_start = time.time()

    logger.info("Loading config from YAML file: %s", config)
    raw_config = load_yaml_config(config)
    logger.info("Raw config loaded. Now validating config entries...")

    validated_config = CompareCandsConfig(**raw_config)
    logger.info("Config validation completed.")

    logger.info("Comparing candidates across files.")
    run_compare_cands(validated_config)
    logger.info("Candidate comparison completed.")

    elapsed_minutes = (time.time() - t_start) / 60.0
    logger.info("Code run time = %.3f minutes", elapsed_minutes)

run_compare_cands(cfg)

Compare candidates across all configured files and write the merged output CSV.

Source code in blipss/cli/compare_cands.py

def run_compare_cands(cfg: CompareCandsConfig) -> None:
    """Compare candidates across all configured files and write the merged output CSV."""
    files_cfg = cfg.candidate_files
    labels_cfg = cfg.on_off_classification
    snr_cutoffs = {"ON": labels_cfg.on_cutoff, "OFF": labels_cfg.off_cutoff}
    n_files = len(files_cfg.csv_list)
    logger.info("Total no. of input .csv files = %d", n_files)

    file_index: list[npt.NDArray[np.intp]] = []
    channels: list[npt.NDArray[np.intp]] = []
    radiofreqs: list[npt.NDArray[np.floating]] = []
    phase_bins: list[npt.NDArray[np.uint]] = []
    boxcar_widths: list[npt.NDArray[np.uint]] = []
    periods: list[npt.NDArray[np.floating]] = []
    snrs: list[npt.NDArray[np.floating]] = []

    for i, (csv_name, label) in enumerate(zip(files_cfg.csv_list, labels_cfg.labels, strict=True)):
        file_channels, file_radiofreqs, file_phase_bins, file_boxcar_widths, file_periods, file_snrs = (
            _read_and_filter_file(files_cfg.csv_dir / csv_name, label, snr_cutoffs[label])
        )
        file_index.append(np.full(len(file_channels), i, dtype=np.intp))
        channels.append(file_channels)
        radiofreqs.append(file_radiofreqs)
        phase_bins.append(file_phase_bins)
        boxcar_widths.append(file_boxcar_widths)
        periods.append(file_periods)
        snrs.append(file_snrs)

    logger.info("Final no. of candidates across all input files = %d", sum(len(p) for p in periods))
    logger.info(
        "Channel-wise grouping of candidate periods into clusters of radius %.2f ms using %d worker process(es)",
        cfg.candidate_grouping.cluster_radius * 1.0e3,
        cfg.candidate_grouping.n_jobs,
    )
    out_channels, out_radiofreqs, out_phase_bins, out_boxcar_widths, out_periods, out_snrs, out_codes = (
        group_candidates_by_channel(
            np.concatenate(file_index),
            np.concatenate(channels),
            np.concatenate(radiofreqs),
            np.concatenate(phase_bins),
            np.concatenate(boxcar_widths),
            np.concatenate(periods),
            np.concatenate(snrs),
            n_files,
            cfg.candidate_grouping.cluster_radius,
            n_jobs=cfg.candidate_grouping.n_jobs,
        )
    )

    output_dir = cfg.output.output_dir
    output_dir.mkdir(parents=True, exist_ok=True)
    output_csv = output_dir / f"{cfg.output.basename}_comparecands.csv"
    logger.info("Writing CSV: %s", output_csv)
    write_compared_candidates_csv(
        output_csv, out_channels, out_radiofreqs, out_phase_bins, out_boxcar_widths, out_periods, out_snrs, out_codes
    )

blipss.cli.compute_phase_resolved_ds

Produce a phase-resolved dynamic spectrum plot for a given folding period.

Reads folding parameters from a YAML config file, loads the target filterbank, folds each spectral channel time series using the riptide FFA library, and saves a grayscale phase-resolved spectrum plot.

Example Usage

python -m blipss.cli.compute_phase_resolved_ds --config config/compute_phase_resolved_ds.yaml

main(config)

Produce a phase-resolved dynamic spectrum plot for a given folding period.

Source code in blipss/cli/compute_phase_resolved_ds.py

@app.command()
def main(
    config: Annotated[Path, typer.Option("--config", help="Path to YAML config file")],
) -> None:
    """Produce a phase-resolved dynamic spectrum plot for a given folding period."""
    t_start = time.time()

    logger.info("Loading config from YAML file: %s", config)
    raw_config = load_yaml_config(config)
    logger.info("Raw config loaded. Now, validating config entries...")

    validated_config = PhaseResolvedDsConfig(**raw_config)
    logger.info("Config validation completed.")

    run_compute_phase_resolved_ds(validated_config)

    elapsed_minutes = (time.time() - t_start) / 60.0
    logger.info("Code run time = %.3f minutes", elapsed_minutes)

run_compute_phase_resolved_ds(cfg)

Orchestrate data loading, band alignment, channel clipping, phase folding, and plot writing.

Parameters:

Name	Type	Description	Default
cfg	PhaseResolvedDsConfig	Validated phase-resolved dynamic spectrum configuration.	required

Source code in blipss/cli/compute_phase_resolved_ds.py

def run_compute_phase_resolved_ds(cfg: PhaseResolvedDsConfig) -> None:
    """
    Orchestrate data loading, band alignment, channel clipping, phase folding, and plot writing.

    Args:
        cfg: Validated phase-resolved dynamic spectrum configuration.
    """
    input_cfg = cfg.input_data
    output_cfg = cfg.output
    ch_cfg = cfg.channel_selection
    fold_cfg = cfg.phase_folding_parameters
    limits = cfg.resource_limits

    file_path = input_cfg.data_dir / input_cfg.datafile
    logger.info("Reading in: %s", file_path)
    wat = read_waterfall_file(file_path, limits.mem_load)
    data, _, tsamp = extract_data_array(wat)
    logger.info("Waterfall data successfully loaded into memory.")

    freqs_MHz, start_mjd, _ = extract_waterfall_metadata(wat)
    data, freqs_MHz = align_band_orientation(data, freqs_MHz, wat.header["foff"])
    data, freqs_MHz = clip_channels(data, freqs_MHz, ch_cfg.start_ch, ch_cfg.stop_ch)

    logger.info("Computing phase-resolved spectrum over %d channels.", len(data))
    phaseresolved_ds = fold_all_channels(
        data=data,
        tsamp=tsamp,
        period=fold_cfg.period,
        bins=fold_cfg.bins,
        do_deredden=fold_cfg.do_deredden,
        rmed_width=fold_cfg.rmed_width,
        n_workers=limits.n_workers,
    )
    logger.info("Done")

    if output_cfg.plot_dir is None:
        raise ValueError("Output plot directory was not resolved by model_validator.")

    ensure_path_exists(output_cfg.plot_dir)
    plot_name = str(output_cfg.plot_dir / f"{output_cfg.basename}_period{fold_cfg.period:.5f}")
    logger.info("Saving plot to disk.")
    plot_phase_resolved_dynamic_spectrum(
        phaseresolved_ds,
        freqs_MHz,
        fold_cfg.period,
        start_mjd,
        plot_name,
        output_cfg.plot_formats,
        output_cfg.use_latex,
    )
    logger.info(f"Phase-resolved dynamic spectrum saved with basepath (excluding file extension): {plot_name}")

blipss.cli.inject_signal

Inject a periodic signal of known properties into a real-world filterbank data file.

Reads injection parameters from a YAML config file, loads the target filterbank, injects one or more channel-wide boxcar pulse trains calibrated to the local bandpass statistics, and writes the result as a .fil or .h5 file.

Example Usage

python -m blipss.cli.inject_signal --config config/inject_signal.yaml

main(config)

Inject a fake periodic signal into a real-world filterbank data file.

Source code in blipss/cli/inject_signal.py

@app.command()
def main(
    config: Annotated[Path, typer.Option("--config", help="Path to YAML config file")],
) -> None:
    """Inject a fake periodic signal into a real-world filterbank data file."""
    t_start = time.time()

    logger.info("Loading config from YAML file: %s", config)
    raw_config = load_yaml_config(config)
    logger.info("Raw config loaded. Now, validating config entries...")

    validated_config = InjectSignalConfig(**raw_config)
    logger.info("Config validation completed.")

    run_inject_signal(validated_config)

    elapsed_minutes = (time.time() - t_start) / 60.0
    logger.info("Code run time = %.3f minutes", elapsed_minutes)

run_inject_signal(cfg)

Orchestrate data loading, bandpass estimation, signal injection, and output writing.

Parameters:

Name	Type	Description	Default
cfg	InjectSignalConfig	Validated injection configuration.	required

Source code in blipss/cli/inject_signal.py

def run_inject_signal(cfg: InjectSignalConfig) -> None:
    """
    Orchestrate data loading, bandpass estimation, signal injection, and output writing.

    Args:
        cfg: Validated injection configuration.
    """
    input_cfg = cfg.input_data
    output_cfg = cfg.output
    injection_cfg = cfg.periodic_signal_injection
    limits = cfg.resource_limits

    if output_cfg.output_dir is None:
        raise ValueError("Output directory was not resolved by model_validator.")

    file_path = input_cfg.data_dir / input_cfg.datafile
    logger.info(f"Reading in: {file_path}")
    wat = read_waterfall_file(file_path, limits.mem_load)
    data, n_samples, tsamp = extract_data_array(wat)
    sample_times = np.arange(n_samples, dtype=np.float64) * tsamp
    logger.info("Waterfall data successfully loaded into memory.")

    logger.info("Computing median bandpass and per-channel standard deviation.")
    median_bp = compute_median_bandpass(data)
    std_perchan = compute_per_channel_std(data)
    logger.info("Per-channel statistics computed.")

    logger.info("Beginning periodic signal injection...")
    for i, channel in enumerate(injection_cfg.inject_channels):
        # Set the pulse peak pulse_snr sigma above the local bandpass level.
        amplitude = median_bp[channel] + injection_cfg.pulse_snr[i] * std_perchan[channel]
        # inject_periodic_signal adds its pulse_snr argument directly to on-pulse samples,
        # which equals the desired SNR only when the background has unit variance. For real
        # data, the noise variance is channel-dependent, so the noise-calibrated amplitude is
        # passed instead.
        inject_periodic_signal(
            data=data,
            sample_times=sample_times,
            channel=channel,
            period=injection_cfg.periods[i],
            duty_cycle=injection_cfg.duty_cycles[i],
            pulse_snr=amplitude,
            initial_phase=injection_cfg.initial_phase[i],
        )
        logger.info("Injected P = %.2f s signal into channel %d.", injection_cfg.periods[i], channel)
    logger.info("Signal injections completed.")

    wat = pack_data_into_waterfall(data, wat, n_samples)
    ensure_path_exists(output_cfg.output_dir)
    output_path = output_cfg.output_dir / f"{output_cfg.basename}{output_cfg.output_ext}"
    write_waterfall(wat, output_path)
    logger.info("%s written to disk.", output_path)

blipss.cli.plot_cands

Produce candidate verification plots combining periodograms, pulse profiles, and phase-time diagrams.

Example usage

python -m blipss.cli.plot_cands --config config/plot_cands.yaml

main(config)

Produce candidate verification plots combining periodograms, pulse profiles, and phase-time diagrams.

Source code in blipss/cli/plot_cands.py

@app.command()
def main(config: Annotated[Path, typer.Option("--config", help="Path to YAML config file")]) -> None:
    """Produce candidate verification plots combining periodograms, pulse profiles, and phase-time diagrams."""
    t_start = time.time()

    logger.info("Loading config from YAML file: %s", config)
    raw_config = load_yaml_config(config)
    logger.info("Raw config loaded. Now validating config entries...")

    validated_config = PlotCandsConfig(**raw_config)
    logger.info("Config validation completed.")

    run_plot_cands(validated_config)

    elapsed_minutes = (time.time() - t_start) / 60.0
    logger.info("Code run time = %.3f minutes", elapsed_minutes)

run_plot_cands(cfg)

Select candidates by code, fold each across all data files, and save verification plots.

Parameters:

Name	Type	Description	Default
cfg	PlotCandsConfig	Validated plot_cands configuration.	required

Source code in blipss/cli/plot_cands.py

def run_plot_cands(cfg: PlotCandsConfig) -> None:
    """
    Select candidates by code, fold each across all data files, and save verification plots.

    Args:
        cfg: Validated plot_cands configuration.
    """
    input_cfg = cfg.input_data
    plotting_cfg = cfg.plotting_parameters
    fold_cfg = cfg.folding_search_parameters

    logger.info("Reading file: %s", cfg.candidate_file.csvfile)
    channels, _, phase_bins, _, periods, _, codes = read_compared_candidates_csv(cfg.candidate_file.csvfile)
    sel_channels, sel_periods, sel_bins, sel_codes = select_candidates_by_code(
        channels, periods, phase_bins, codes, plotting_cfg.codes_plot
    )
    n_cands = len(sel_channels)
    logger.info("No. of candidates selected for plotting = %d", n_cands)

    all_data, start_mjds, tsamps = _load_datafiles(
        input_cfg.data_dir, input_cfg.datafile_list, cfg.resource_limits.mem_load
    )

    if plotting_cfg.plot_dir is None:
        raise ValueError("Output plot directory was not resolved by model_validator.")
    ensure_path_exists(plotting_cfg.plot_dir)

    for i in range(n_cands):
        chan = int(sel_channels[i])
        period = float(sel_periods[i])
        bins = int(sel_bins[i])
        code = str(sel_codes[i])
        logger.info(
            "Working with candidate %d/%d: channel=%d, period=%.5f s, bins=%d, code=%s",
            i + 1,
            n_cands,
            chan,
            period,
            bins,
            code,
        )
        _produce_candidate_plot(
            chan,
            period,
            bins,
            code,
            all_data,
            tsamps,
            start_mjds,
            input_cfg.beam_labels,
            plotting_cfg.plot_dir,
            plotting_cfg.basename,
            fold_cfg,
            plotting_cfg,
        )

blipss.cli.run_ffa_search

Run channel-wise FFA on a set of input filterbank files, flag harmonics, and write one CSV of candidates per file.

Example usage

python -m blipss.cli.run_ffa_search --config config/run_ffa_search.yaml

main(config)

Run channel-wise FFA period search on a set of filterbank files.

Source code in blipss/cli/run_ffa_search.py

@app.command()
def main(config: Annotated[Path, typer.Option("--config", help="Path to YAML config file")]) -> None:
    """Run channel-wise FFA period search on a set of filterbank files."""
    t_start = time.time()

    logger.info("Loading config from YAML file: %s", config)
    raw_config = load_yaml_config(config)
    logger.info("Raw config loaded. Now validating config entries...")

    validated_config = BlipssConfig(**raw_config)
    logger.info("Config validation completed.")

    logger.info("Starting FFA period search.")
    run_blipss(validated_config)
    logger.info("FFA period search completed.")

    elapsed_minutes = (time.time() - t_start) / 60.0
    logger.info("Code run time = %.3f minutes", elapsed_minutes)

run_blipss(cfg)

Orchestrate per-file FFA search and output writing for all files in the config.

Source code in blipss/cli/run_ffa_search.py

def run_blipss(cfg: BlipssConfig) -> None:
    """Orchestrate per-file FFA search and output writing for all files in the config."""
    output_dir = cfg.output.output_dir
    if output_dir is None:
        raise ValueError("output.output_dir must be resolved before use")
    output_dir.mkdir(parents=True, exist_ok=True)

    logger.info("Total input files: %d", len(cfg.input.input_file_list))
    for datafile in cfg.input.input_file_list:
        _process_single_file(datafile, cfg)

blipss.cli.simulate_data

Generate a synthetic filterbank file containing periodic signals on a Gaussian white noise background.

Reads simulation parameters from a YAML config file, injects one or more channel-wide boxcar pulse trains at specified periods and duty cycles, and writes the result as a 32-bit sigproc filterbank (.fil) file.

Example Usage

python -m blipss.cli.simulate_data --config config/simulate_data.yaml

main(config)

Generate an artificial filterbank data set.

Source code in blipss/cli/simulate_data.py

@app.command()
def main(config: Annotated[Path, typer.Option("--config", help="Path to YAML config file")]) -> None:
    """Generate an artificial filterbank data set."""
    t_start = time.time()

    logger.info("Loading config from YAML file: %s", config)
    raw_config = load_yaml_config(config)
    logger.info("Raw config loaded. Now, validating config entries...")

    validated_config = SimulateDataConfig(**raw_config)
    logger.info("Config validation completed.")

    logger.info("Initiating simulation of artificial dataset")
    run_simulate_data(validated_config)
    logger.info("Dataset simulation completed.")

    elapsed_minutes = (time.time() - t_start) / 60.0
    logger.info("Code run time = %.3f minutes", elapsed_minutes)

run_simulate_data(cfg)

Orchestrate noise generation, signal injection, and filterbank file writing.

Source code in blipss/cli/simulate_data.py

def run_simulate_data(cfg: SimulateDataConfig) -> None:
    """Orchestrate noise generation, signal injection, and filterbank file writing."""
    sim = cfg.simulation_properties
    signal_injection = cfg.periodic_signal_injection
    header_params = cfg.optional_header_parameters
    output_cfg = cfg.output

    sample_times = np.arange(sim.n_samples, dtype=np.float64) * sim.t_samp
    rng = np.random.default_rng(sim.seed)
    data = generate_white_noise_background(sim.n_channels, sim.n_samples, rng=rng)
    logger.info("Background Gaussian white noise data generated.")

    for i, channel in enumerate(signal_injection.inject_channels):
        inject_periodic_signal(
            data=data,
            sample_times=sample_times,
            channel=channel,
            period=signal_injection.periods[i],
            duty_cycle=signal_injection.duty_cycles[i],
            pulse_snr=signal_injection.pulse_snr[i],
            initial_phase=signal_injection.initial_phase[i],
        )
        logger.info("Injected P = %.2f s signal into channel %d.", signal_injection.periods[i], channel)

    data = reshape_for_sigproc(data)
    header = build_sigproc_header(sim, header_params)
    logger.info("Writing filterbank file to %s/%s.fil", output_cfg.output_dir, output_cfg.basename)
    write_filterbank(data, header, output_cfg.output_dir, output_cfg.basename)
    logger.info("Write operation successfully completed.")