flint.rmsynth

RM-synthesis and RM-CLEAN of Stokes Q/U cubes using the external rm-lite package (an optional dependency, see the rmsynth extra).

Attributes

FDFLabel

FDFThreshold

HeaderCards

MOMENT_MAPS

PEAK_MAPS

STOKES_I_MAP_SUFFIXES

Functions

_check_cubes_memmappable(→ None)

rm-lite reads each spatial block by reopening the cube with memmap=True.

_describe_rm_workload(→ str)

One line saying what the coming compute is actually going to chew through.

_elapsed(→ str)

Wall time since start, as a compact human-readable string.

_lazy_faraday_moments(...)

Build the lazy Faraday moment maps of an FDF cube.

_seceded_if_on_a_worker(→ collections.abc.Iterator[None])

Give this worker's compute slot back for the duration of a blocking wait.

_stamp_stokes_i_weights(→ None)

Record how the weights followed the Stokes I division, if they did.

_stokes_i_fit_header(→ astropy.io.fits.Header)

Build the celestial-WCS header for a Stokes I fit map, stamped with what

_warn_on_no_error_cubes(→ None)

Warn that rm-lite is about to measure the noise off the cubes themselves.

_write_map(→ pathlib.Path)

Write one float32 map, the last step of every writer here.

check_cubes_are_single_precision(→ None)

rm-lite takes the FDF's precision from the cubes it is given, so a

compute_rm_products(→ dict[str, Any])

Compute every product in one shared pass, reporting each as it lands.

fdf_threshold_from_snr(→ FDFThreshold)

An FDF amplitude cut snr times the theoretical noise, or None for no cut.

fdf_units_for(→ rm_lite.utils.synthesis.FDFUnits)

Amplitude units of an FDF product.

needs_rmclean(→ bool)

Whether any requested product requires RM-CLEAN to be run.

restoring_beam_cards(→ HeaderCards)

The cube's restoring beam for the 2D maps; none if it has none, or a blanked image's zero beam.

rm_synthesis_header_cards(→ HeaderCards)

The run's lambda^2_0 (unless per-pixel), RMSF width and band, so each map reads on its own.

run_rmclean_3d(→ rm_lite.tools_3d.rmclean.RMClean3DResults)

Run 3D RM-CLEAN on the results of run_rmsynth_3d.

run_rmsynth_3d(→ rm_lite.tools_3d.rmsynth.RMSynth3DResults)

Run 3D RM-synthesis on Stokes Q/U FITS cubes.

write_moment_maps_to_fits(→ list[pathlib.Path])

Write already-computed Faraday moment maps to FITS.

write_peak_maps_to_fits(→ list[pathlib.Path])

Write the already-computed FDF peak-statistic maps to FITS.

write_rm_product_to_fits(→ list[pathlib.Path])

Write one computed product, named and united from its compute key.

write_rm_products(→ list[pathlib.Path])

Batch-compute and write the requested RM-synthesis/RM-CLEAN output products.

write_rmclean_niter_map_to_fits(→ pathlib.Path)

Write the per-pixel RM-CLEAN iteration count.

write_stokes_i_coeff_maps_to_fits(→ list[pathlib.Path])

Write the fitted Stokes I model terms.

write_stokes_i_fit_maps_to_fits(→ list[pathlib.Path])

Write the per-pixel Stokes I fractional-polarisation fit maps: the fitted

Module Contents

flint.rmsynth._check_cubes_memmappable(*cubes: pathlib.Path | None) → None[source]

rm-lite reads each spatial block by reopening the cube with memmap=True. astropy cannot memmap a compressed (.gz or .fz) file, so every block read decompresses the whole cube into memory. Compress after RM-synthesis, not before.

flint.rmsynth._describe_rm_workload(synth_results: rm_lite.tools_3d.rmsynth.RMSynth3DResults, compute_keys: list[str]) → str[source]

One line saying what the coming compute is actually going to chew through.

Everything here is known before any of it runs, and none of it is otherwise reported: the FDF’s shape, how many spatial chunks it is cut into, and whether rm-lite decided every pixel needs its own RMSF (roughly doubling the footprint, and not something flint asked for – see RMSynthOptions.per_pixel_rmsf).

flint.rmsynth._elapsed(start: float) → str[source]

Wall time since start, as a compact human-readable string.

flint.rmsynth._lazy_faraday_moments(fdf_cube: dask.array.Array, synth_results: rm_lite.tools_3d.rmsynth.RMSynth3DResults, threshold: FDFThreshold, fdf_units: rm_lite.utils.synthesis.FDFUnits, debias: bool = False, debias_filter_size: int = 5) → rm_lite.utils.synthesis.FaradayMoments[source]

Build the lazy Faraday moment maps of an FDF cube.

calc_faraday_moments reduces along the (never-chunked) Faraday-depth axis, and debias_fdf handles dask via map_overlap, so the result is a set of lazy (ny, nx) maps that each spatial chunk contributes to independently. Computing these instead of the cube itself is what keeps the whole FDF out of the calling worker’s memory.

The theoretical noise is what turns the errors and the debiased maps from NaN into measurements, so it is passed whenever synthesis reported one – it is the same noise the amplitude cut is derived from.

flint.rmsynth._seceded_if_on_a_worker() → collections.abc.Iterator[None][source]

Give this worker’s compute slot back for the duration of a blocking wait.

task_write_rm_products is a prefect task, so under DaskTaskRunner it runs on a worker, and prefect_dask.get_dask_client hands back a plain Client rather than a worker_client – it does not secede. Waiting on futures from there holds the worker’s thread while the tasks being waited on need worker threads to run, and with one thread per worker that deadlocks the cluster outright.

dask.compute(scheduler=client) never showed this because Client.get secedes for you when it is called inside a worker. Draining the futures by hand is what makes it ours to do.

A no-op off a worker, which is the local-scheduler and standalone case.

flint.rmsynth._stamp_stokes_i_weights(header: astropy.io.fits.Header, weighting: str | None, alpha: float | None) → None[source]

Record how the weights followed the Stokes I division, if they did.

flint.rmsynth._stokes_i_fit_header(reference_header: astropy.io.fits.Header, ref_freq_hz: float | None, fit_function: str | None, coeff: tuple[int, str] | None = None) → astropy.io.fits.Header[source]

Build the celestial-WCS header for a Stokes I fit map, stamped with what is needed to read it. A spectral index or a reference flux means nothing without the frequency it is defined at and the functional form it belongs to, so those travel in the header rather than only in flint’s file naming.

Parameters:

  • reference_header (fits.Header) – Header to derive the spatial WCS from

  • ref_freq_hz (float | None) – Frequency the model terms are defined at, omitted from the header if rm-lite did not report one

  • fit_function (str | None) – The Stokes I fit function the terms belong to (‘log’ or ‘linear’), omitted from the header if not known

  • coeff (tuple[int, str] | None, optional) – The (popt index, name) of the model term this map holds, for the per-term maps. Defaults to None.

Returns:

The header to write the map with

Return type:

fits.Header

flint.rmsynth._warn_on_no_error_cubes(error_cubes: flint.options.ErrorCubesForRMSynth | None) → None[source]

Warn that rm-lite is about to measure the noise off the cubes themselves.

Its fallback is one mad_std per channel plane, so every pixel is told the whole field’s noise. On a mosaic, whose edge is far noisier than its middle, that makes an SNR-scaled CLEAN mask meaningless out there and RM-CLEAN grinds on noise. Warned about rather than refused: it is the right answer for a cube of uniform noise.

flint.rmsynth._write_map(data: numpy.ndarray, header: astropy.io.fits.Header, output_path: pathlib.Path) → pathlib.Path[source]

Write one float32 map, the last step of every writer here.

flint.rmsynth.check_cubes_are_single_precision(*cubes: pathlib.Path | None) → None[source]

rm-lite takes the FDF’s precision from the cubes it is given, so a double-precision cube doubles every Faraday-depth array it builds. Warned about rather than refused: it is a memory cost, not an error.

flint.rmsynth.compute_rm_products(compute_targets: dict[str, Any], fuse_config: dict[str, Any], scheduler: dask.distributed.Client | str, workload: str, on_result: collections.abc.Callable[[str, Any], Any] | None = None) → dict[str, Any][source]

Compute every product in one shared pass, reporting each as it lands.

The whole RM-synthesis stage is one compute over a lazy graph, so without this it is a single silent wait – hours on a mosaic, with no way to tell a slow run from a stuck one. Submitting the targets together (rather than one compute per product) is what keeps the synthesis and RM-CLEAN graph shared, so it still runs once no matter how many products are asked for – getting that wrong is invisible in the output and simply multiplies the runtime, so test_rmclean_runs_once_per_chunk_on_a_distributed_client holds it for this path.

Progress needs a distributed Client to report against futures. The local schedulers take the plain dask.compute path and are logged start-to-end only – that is the unit tests and standalone use, not a pipeline run.

Parameters:

  • compute_targets (dict[str, Any]) – Lazy arrays/delayed writes, keyed by product name

  • fuse_config (dict[str, Any]) – Dask fusion settings the submission must be made under

  • scheduler (Client | str) – A distributed Client, or a local scheduler name

  • workload (str) – Description of the work, logged up front. See _describe_rm_workload

  • on_result (Callable[[str, Any], Any] | None, optional) – Called with each key and its value as that value lands, and what it returns is kept in place of the value. Pass a writer to keep one product in memory rather than all of them: the maps are ny * nx each and there are 21 per label. Defaults to None, keeping the values themselves.

Returns:

Per key, the computed value, or whatever on_result returned for it

Return type:

dict[str, Any]

flint.rmsynth.fdf_threshold_from_snr(snr: float, fdf_error_noise: FDFThreshold) → FDFThreshold[source]

An FDF amplitude cut snr times the theoretical noise, or None for no cut.

Zero short-circuits rather than multiplying through: the noise is inf where a pixel carries no weight, and 0 * inf is NaN, which would blank every comparison instead of passing everything.

flint.rmsynth.fdf_units_for(label: FDFLabel) → rm_lite.utils.synthesis.FDFUnits[source]

Amplitude units of an FDF product.

RM-synthesis output is per RMSF; the CLEAN component model is a list of fluxes, so summing it must not divide by the RMSF area as well.

flint.rmsynth.needs_rmclean(cube_products: list[FDFLabel], moment_products: list[FDFLabel], peak_products: list[FDFLabel]) → bool[source]

Whether any requested product requires RM-CLEAN to be run.

Parameters:

  • cube_products (list[FDFLabel]) – Requested FDF cubes

  • moment_products (list[FDFLabel]) – Requested Faraday moment maps

  • peak_products (list[FDFLabel]) – Requested FDF peak-statistic maps

Returns:

True if RM-CLEAN is needed

Return type:

bool

flint.rmsynth.restoring_beam_cards(cube_header: astropy.io.fits.Header) → HeaderCards[source]

The cube’s restoring beam for the 2D maps; none if it has none, or a blanked image’s zero beam.

flint.rmsynth.rm_synthesis_header_cards(synth_results: rm_lite.tools_3d.rmsynth.RMSynth3DResults, lam_sq_0_m2: float | str) → HeaderCards[source]

The run’s lambda^2_0 (unless per-pixel), RMSF width and band, so each map reads on its own.

flint.rmsynth.run_rmclean_3d(rm_synth_results: rm_lite.tools_3d.rmsynth.RMSynth3DResults, rmclean_options: flint.options.RMCleanOptions) → rm_lite.tools_3d.rmclean.RMClean3DResults[source]

Run 3D RM-CLEAN on the results of run_rmsynth_3d.

Parameters:

  • rm_synth_results (RMSynth3DResults) – Results from run_rmsynth_3d

  • rmclean_options (RMCleanOptions) – Options controlling RM-CLEAN

Returns:

Lazy clean/model/residual FDF cubes and moment maps

Return type:

RMClean3DResults

flint.rmsynth.run_rmsynth_3d(stokes_cubes: flint.options.CubesForRMSynth, rmsynth_options: flint.options.RMSynthOptions, error_cubes: flint.options.ErrorCubesForRMSynth | None = None) → rm_lite.tools_3d.rmsynth.RMSynth3DResults[source]

Run 3D RM-synthesis on Stokes Q/U FITS cubes.

Parameters:

  • stokes_cubes (CubesForRMSynth) – The Stokes Q/U cubes, and optionally I to fit a per-pixel fractional-polarisation correction against

  • rmsynth_options (RMSynthOptions) – Options controlling the synthesis

  • error_cubes (ErrorCubesForRMSynth | None, optional) – Either the linmos weight cubes (an inverse variance) or a set of noise cubes (a sigma), which the type says which of. None leaves rm-lite to estimate a per-channel noise from Q/U itself. Defaults to None.

Returns:

Lazy dirty FDF cube, the RMSF, and associated parameters. The linmos weights vary with the primary beam, so pixels rarely share one RMSF and rm-lite turns on the per-pixel RMSF cube itself; see RMSynthOptions.per_pixel_rmsf for what that costs

Return type:

RMSynth3DResults

flint.rmsynth.write_moment_maps_to_fits(moments: rm_lite.utils.synthesis.FaradayMoments, reference_header: astropy.io.fits.Header, output_prefix: pathlib.Path, label: FDFLabel, debiased_moments: rm_lite.utils.synthesis.FaradayMoments | None = None) → list[pathlib.Path][source]

Write already-computed Faraday moment maps to FITS.

The moments are built lazily and computed by write_rm_products, which keeps the (n_phi, ny, nx) FDF cube they reduce out of this process; only the (ny, nx) maps arrive here. See _lazy_faraday_moments.

Parameters:

  • moments (FaradayMoments) – Computed moment maps, each (ny, nx)

  • reference_header (fits.Header) – Header to derive the spatial WCS from (e.g. the Stokes Q cube header)

  • output_prefix (Path) – Common prefix for the output files

  • label (FDFLabel) – Which FDF the moments came from (‘dirty’, ‘clean’, or ‘model’), used to name the outputs

  • debiased_moments (FaradayMoments | None, optional) – Debiased moment set, written alongside with a .debiased suffix. Defaults to None.

Returns:

One path per MOMENT_MAPS entry, plus the debiased set less mom0_debias if debiased_moments is given

Return type:

list[Path]

flint.rmsynth.write_peak_maps_to_fits(peaks: rm_lite.utils.synthesis.FaradayPeaks, reference_header: astropy.io.fits.Header, output_prefix: pathlib.Path, label: FDFLabel) → list[pathlib.Path][source]

Write the already-computed FDF peak-statistic maps to FITS.

Parameters:

  • peaks (FaradayPeaks) – Computed peak maps, each (ny, nx)

  • reference_header (fits.Header) – Header to derive the spatial WCS from

  • output_prefix (Path) – Common prefix for the output files

  • label (FDFLabel) – Which FDF the peaks came from, used to name the outputs

Returns:

The written map paths, nine per FDF

Return type:

list[Path]

flint.rmsynth.write_rm_product_to_fits(key: str, data: numpy.ndarray, reference_header: astropy.io.fits.Header, output_prefix: pathlib.Path, ref_freq_hz: float | None = None, fit_function: str | None = None, coeff_names: tuple[str, ...] | None = None, stokes_i_weighting: str | None = None, stokes_i_weight_alpha: float | None = None, fdf_tag: str = 'fdf', header_cards: HeaderCards | None = None) → list[pathlib.Path][source]

Write one computed product, named and united from its compute key.

write_rm_products computes its products as a batch and writes each one as it lands, so this takes a single map rather than a set: holding the whole set costs 21 * ny * nx * 4 bytes a label, which no chunk setting bounds. The grouped writers above take the same maps when a caller already has them all in hand.

Parameters:

  • key (str) – Compute key, e.g. moment.clean.mom0 or stokes_i_alpha

  • data (np.ndarray) – The computed map, (ny, nx), or (n_coeff, ny, nx) for the model terms

  • reference_header (fits.Header) – Header to derive the spatial WCS from

  • output_prefix (Path) – Common prefix for the output files

  • ref_freq_hz (float | None, optional) – Frequency the Stokes I fit is referenced to. Defaults to None.

  • fit_function (str | None, optional) – The Stokes I fit function. Defaults to None.

  • coeff_names (tuple[str, ...] | None, optional) – Name of each plane of the model-term cube. Defaults to None.

  • stokes_i_weighting (str | None, optional) – How the weights followed the Stokes I division, recorded as SIWEIGHT. Defaults to None.

  • stokes_i_weight_alpha (float | None, optional) – Spectral index of the weight template, recorded as SIWALPHA. Defaults to None.

  • fdf_tag (str, optional) – What the FDF products are named after, “fdf” or “fdf_no_i” for the run without Stokes I. Defaults to “fdf”.

  • header_cards (HeaderCards | None, optional) – Run metadata stamped on every map, from rm_synthesis_header_cards. Defaults to None.

Returns:

The paths written, empty for a key with no FITS output

Return type:

list[Path]

flint.rmsynth.write_rm_products(synth_results: rm_lite.tools_3d.rmsynth.RMSynth3DResults, clean_results: rm_lite.tools_3d.rmclean.RMClean3DResults | None, stokes_q_cube: pathlib.Path, rmsynth_options: flint.options.RMSynthOptions, rmclean_options: flint.options.RMCleanOptions, cube_products: list[FDFLabel], moment_products: list[FDFLabel], peak_products: list[FDFLabel], output_prefix: pathlib.Path, moment_threshold_snr: float = 5.0, dask_client: dask.distributed.Client | None = None, fdf_tag: str = 'fdf') → list[pathlib.Path][source]

Batch-compute and write the requested RM-synthesis/RM-CLEAN output products.

Parameters:

  • synth_results (RMSynth3DResults) – Results from run_rmsynth_3d

  • clean_results (RMClean3DResults | None) – Results from run_rmclean_3d, or None if ‘clean’/’model’ were not requested

  • stokes_q_cube (Path) – Path to the Stokes Q FITS cube (its header is reused for output WCS)

  • rmsynth_options (RMSynthOptions) – Options controlling RM-synthesis

  • rmclean_options (RMCleanOptions) – Options controlling RM-CLEAN

  • cube_products (list[FDFLabel]) – Which FDF cube(s) to write (‘dirty’, ‘clean’, ‘model’)

  • moment_products (list[FDFLabel]) – Which FDF(s) to compute Faraday moment maps from

  • peak_products (list[FDFLabel]) – Which FDF(s) to write peak-statistic maps from, nine per entry

  • moment_threshold_snr (float, optional) – SNR cut applied before the moment maps. Defaults to 5.0.

  • output_prefix (Path) – Common prefix for the output files

  • dask_client (Client | None, optional) – A distributed Client (e.g. the one backing a Prefect DaskTaskRunner) to compute across, rather than just the local worker. Defaults to None.

  • fdf_tag (str, optional) – What the FDF products are named after, “fdf” or “fdf_no_i” for the run without Stokes I. Defaults to “fdf”.

Returns:

Every FITS path written

Return type:

list[Path]

flint.rmsynth.write_rmclean_niter_map_to_fits(niter_map: numpy.ndarray, reference_header: astropy.io.fits.Header, output_prefix: pathlib.Path, fdf_tag: str = 'fdf', header_cards: HeaderCards | None = None) → pathlib.Path[source]

Write the per-pixel RM-CLEAN iteration count.

One small integer map, written whenever RM-CLEAN runs rather than on request: it is the map that answers “why does this field look wrong” – where CLEAN hit max_iter instead of converging on the threshold – and that question gets asked after the run, when producing it on demand would mean repeating the whole stage.

Parameters:

  • niter_map (np.ndarray) – Computed (ny, nx) iteration count

  • reference_header (fits.Header) – Header to derive the spatial WCS from

  • output_prefix (Path) – Common prefix for the output file

  • fdf_tag (str, optional) – What the FDF products are named after. Defaults to “fdf”.

  • header_cards (HeaderCards | None, optional) – Run metadata to stamp, from rm_synthesis_header_cards. Defaults to None.

Returns:

The written map path

Return type:

Path

flint.rmsynth.write_stokes_i_coeff_maps_to_fits(coeff_cube: numpy.ndarray, coeff_names: tuple[str, ...], reference_header: astropy.io.fits.Header, output_prefix: pathlib.Path, ref_freq_hz: float | None = None, fit_function: str | None = None, coeff_error_cube: numpy.ndarray | None = None) → list[pathlib.Path][source]

Write the fitted Stokes I model terms.

The terms plus the reference frequency and the fit function are the whole Stokes I model, so anything downstream can evaluate Stokes I at any frequency from these maps alone without carrying the model cube – but only if it knows which plane holds which term, which is why each gets its own named file and its REFFREQ/FITFUNC header.

Two values are meaningful rather than missing, and are recorded in the header as well: a zero is the actual value of a term the AIC dropped (it contributes nothing to the model, and stokesi.model_order says how many terms were really fitted), while a NaN is a pixel that was never fitted at all.

Parameters:

  • coeff_cube (np.ndarray) – Already-computed model terms, shape (n_coeff, ny, nx) in popt order

  • coeff_names (tuple[str, ...]) – Name of each plane of coeff_cube, e.g. (‘flux’, ‘alpha’, ‘beta’)

  • reference_header (fits.Header) – Header to derive the spatial WCS from

  • output_prefix (Path) – Common prefix for the output files

  • ref_freq_hz (float | None, optional) – Frequency the terms are defined at, recorded in each header. Defaults to None.

  • fit_function (str | None, optional) – The Stokes I fit function the terms belong to, recorded in each header. Defaults to None.

  • coeff_error_cube (np.ndarray | None, optional) – 1-sigma marginal error on each term, shaped like coeff_cube, written alongside with an _error suffix. Defaults to None.

Returns:

The written map paths, one per term, plus one more per term if coeff_error_cube is given

Return type:

list[Path]

flint.rmsynth.write_stokes_i_fit_maps_to_fits(stokes_i_maps: dict[str, numpy.ndarray], reference_header: astropy.io.fits.Header, output_prefix: pathlib.Path, ref_freq_hz: float | None = None, fit_function: str | None = None) → list[pathlib.Path][source]

Write the per-pixel Stokes I fractional-polarisation fit maps: the fitted reference flux, spectral index (alpha) and its error, and the fitted polynomial order. Cheap 2D maps, always written when a Stokes I cube is used and rm-lite actually returned that particular map (e.g. alpha_error is only produced if the model error was computed).

Parameters:

  • stokes_i_maps (dict[str, np.ndarray]) – Already-computed maps, keyed by one of stokes_i_ref_flux, stokes_i_alpha, stokes_i_alpha_error or stokes_i_model_order.

  • reference_header (fits.Header) – Header to derive the spatial WCS from

  • output_prefix (Path) – Common prefix for the output files

  • ref_freq_hz (float | None, optional) – Frequency the fit is referenced to, recorded in each header. Defaults to None.

  • fit_function (str | None, optional) – The Stokes I fit function, recorded in each header. Defaults to None.

Returns:

The written map paths, one per entry in stokes_i_maps

Return type:

list[Path]

type flint.rmsynth.FDFLabel = Literal['dirty', 'clean', 'model'][source]
type flint.rmsynth.FDFThreshold = float | np.ndarray | da.Array | None[source]
type flint.rmsynth.HeaderCards = dict[str, tuple[Any, str]][source]
flint.rmsynth.MOMENT_MAPS[source]
flint.rmsynth.PEAK_MAPS[source]
flint.rmsynth.STOKES_I_MAP_SUFFIXES[source]