low_comm_tools.plotting.baselines

Attributes

norms

Classes

SubtableData

Functions

_compute_plot_arrays(...)
_make_plot_sync(→ matplotlib.figure.Figure)
_process_baseline_sync(→ bytes | None)	Full per-baseline pipeline: read → plot → render. Runs in a worker process.
_read_subtable(→ SubtableData | None)
get_parser(→ argparse.ArgumentParser)
main(→ None)
plot_baselines(→ list[pathlib.Path | None])	Plot visibilities for every unique baseline concurrently.
process_baseline(→ pathlib.Path | None)

Module Contents

class low_comm_tools.plotting.baselines.SubtableData

Bases: NamedTuple

ant_1: int

ant_2: int

freq_chan: astropy.units.Quantity

masked_data: numpy.ma.MaskedArray

station_names: list[str]

times: astropy.time.Time

low_comm_tools.plotting.baselines._compute_plot_arrays(subtable_data: SubtableData, plot_type: Literal['spectrum', 'delay', 'delay-rate']) → tuple[numpy.typing.NDArray[numpy.floating[Any]], numpy.typing.NDArray[numpy.floating[Any]], numpy.typing.NDArray[numpy.complexfloating[Any, Any]], str, str]

low_comm_tools.plotting.baselines._make_plot_sync(x_array: numpy.typing.NDArray[numpy.floating[Any]], y_array: numpy.typing.NDArray[numpy.floating[Any]], z_array: numpy.typing.NDArray[numpy.complexfloating[Any, Any]], ant_1: int, ant_2: int, station_names: list[str], baseline_length: astropy.units.Quantity[astropy.units.m], data_type: str, xlabel: str | None = None, ylabel: str | None = None, fast_plot: bool = True, norm: matplotlib.colors.Normalize | None = None) → matplotlib.figure.Figure

low_comm_tools.plotting.baselines._process_baseline_sync(ms_path: pathlib.Path, ant_1: int, ant_2: int, data_column: str, station_names: list[str], freq_chan: astropy.units.Quantity, plot_type: Literal['spectrum', 'delay', 'delay-rate'], data_type: Literal['amp', 'phase', 'real', 'imag'], fast_plot: bool, norm: matplotlib.colors.Normalize | None) → bytes | None

Full per-baseline pipeline: read → plot → render. Runs in a worker process.

low_comm_tools.plotting.baselines._read_subtable(ms_path: pathlib.Path, ant_1: int, ant_2: int, data_column: str, station_names: list[str], freq_chan: astropy.units.Quantity) → SubtableData | None

low_comm_tools.plotting.baselines.get_parser() → argparse.ArgumentParser

low_comm_tools.plotting.baselines.main() → None

async low_comm_tools.plotting.baselines.plot_baselines(ms_path: pathlib.Path | str, fast_plot: bool = True, norm: matplotlib.colors.Normalize | None = None, plot_type: Literal['spectrum', 'delay', 'delay-rate'] = 'spectrum', data_column: str = 'DATA', data_type: Literal['amp', 'phase', 'real', 'imag'] = 'amp', out_dir: pathlib.Path | None = None, max_concurrent: int = 4, max_baselines: int | None = None) → list[pathlib.Path | None]

Plot visibilities for every unique baseline concurrently.

Parameters:

• ms_path – Path to the measurement set.

• fast_plot – Use pcolorfast instead of pcolormesh. Defaults to True.

• norm – Colour-scale normalisation. Defaults to None (linear).

• plot_type – One of "spectrum", "delay", or "delay-rate".

• data_column – MS data column to plot. Defaults to "DATA".

• data_type – "amp" or "phase". Defaults to "amp".

• out_dir – Output directory. Defaults to the MS parent directory.

• max_concurrent – Number of worker processes and max concurrent baselines. Defaults to 4.

async low_comm_tools.plotting.baselines.process_baseline(ms_path: pathlib.Path, ant_1: int, ant_2: int, data_column: str, station_names: list[str], freq_chan: astropy.units.Quantity, plot_type: Literal['spectrum', 'delay', 'delay-rate'], data_type: Literal['amp', 'phase', 'real', 'imag'], fast_plot: bool, norm: matplotlib.colors.Normalize | None, out_dir: pathlib.Path, prefix: str | None, executor: concurrent.futures.ProcessPoolExecutor, sem: asyncio.Semaphore) → pathlib.Path | None

low_comm_tools.plotting.baselines.norms: dict[str, matplotlib.colors.Normalize]