flint.misc.holo¶

Helper utilities around the bespoke holography format of ASKAP

Classes¶

`BeamResponse`	One beam's response from a holography cube, normalised to 1 at the beam centre
`ConcatHoloOptions`	Options to use to concatenate holography cubes today
`FITSCubeInfo`	Slim container to represent a FITS cube
`FinalFITSCubeInfo`	Simple representation of final FITS cube
`FrequencyGrid`	Representative frequency axis information
`ReprojectWorker`

Functions¶

`_direction_cosines`(→ tuple[float, float])	Offset of position from centre as direction cosines (l to the east, m to the north).
`_exit_scale`(→ float)	Fraction of the pixel offset (dx, dy) from the beam centre that stays a pixel inside the cube.
`_get_cube_header`(→ astropy.io.fits.Header)	Return a header object from the input cube that is compatible with `reproject`
`_sample_planes`(→ numpy.typing.NDArray[numpy.floating])	Bilinearly interpolate every plane at one pixel position, giving 0 off the grid.
`celestial_wcs_from_header`(→ astropy.wcs.WCS)	Extract just the celestial (RA/Dec) 2D WCS from a 5D FITS header.
`cli`(→ None)
`concatenate_holography`(→ pathlib.Path)	Reproject a set of ASKAP IQUV primary beam cubes into a single output cube.
`construct_frequency_grid`(→ FrequencyGrid)	Construct the final frequency grid of the concatenated holography cube
`construct_spatial_output_wcs`(→ astropy.io.fits.Header)	Construct the output WCS that will be used for each channel. The
`create_fits_info`(→ FITSCubeInfo)	Load a FITS cube container the holography and extra information
`create_output_header`(→ astropy.io.fits.Header)	Create the final outcube cube header. This uses the header from an example
`create_placeholder_cube`(→ FinalFITSCubeInfo)	Write out the final fits cube, but fill it with placeholder
`get_freq_axis`(→ numpy.typing.NDArray[numpy.floating])	Return array of channel centre frequencies (Hz).
`get_output_data_shape`(→ list[int])	Construct the final output shape of the concatenated holography cube
`get_parser`(→ argparse.ArgumentParser)
`load_and_sort_cubes`(→ list[FITSCubeInfo])	Load and extra appropriate information from each input cube, and
`load_beam_response`(→ BeamResponse)	Read one beam's response from a holography cube, normalised to 1 at the beam centre.
`map_frequencies_to_channels`(...)	Create mappings resolving freq_2 to channels in freqs_1
`outside_planes`(→ numpy.typing.NDArray[numpy.bool_])	Which freqs lie more than half a plane beyond the usable planes, and so take the nearest plane's beam.
`reproject_cubes`(→ FinalFITSCubeInfo)	Reproject the input cubes onto a final output spatial grid, as defined by
`reproject_wrapper`(→ ReprojectWorker)
`sample_beam_response`(...)	Sample a beam's response at a sky position the same way linmos does.

Module Contents¶

class flint.misc.holo.BeamResponse[source]¶

Bases: NamedTuple

One beam’s response from a holography cube, normalised to 1 at the beam centre

centre_pixel: tuple[float, float][source]¶: Zero-based (x, y) pixel of the beam centre

freqs_hz: numpy.typing.NDArray[numpy.floating][source]¶: Frequency of each usable plane

pixel_scale_rad: tuple[float, float][source]¶: Signed pixel size along x and y, in radians (CDELT1, CDELT2)

planes: numpy.typing.NDArray[numpy.floating][source]¶: Response on each usable plane, shaped (freq, y, x)

class flint.misc.holo.ConcatHoloOptions(/, **data: Any)[source]¶

Bases: capn_crunch.BaseOptions

Options to use to concatenate holography cubes today

holo_cubes: tuple[pathlib.Path, ...] = ()[source]¶: The path to the holography IQUV cubes to concatenate together. No sensible default exists (it always names specific input cubes); an empty tuple is a placeholder that must be overridden before calling concatenate_holography, which raises if it is left unset

max_workers: int = 8[source]¶: The maximum number of sub-processes that may be spawned

output_path: pathlib.Path | None = None[source]¶: Output holography cube to make

class flint.misc.holo.FITSCubeInfo[source]¶

Slim container to represent a FITS cube

celestial_wcs: astropy.wcs.WCS[source]¶: The world coordinate system specification that accompanies the spatial dimension

freqs_hz: numpy.typing.NDArray[numpy.floating][source]¶: The minimum frequency in the cube (in Hertz, used for ordering)

header: astropy.io.fits.Header[source]¶: Header from the FITS cube

index: int = 0[source]¶: The HDU index read

path: pathlib.Path[source]¶: Path to the cube being considered

spatial_shape: tuple[int, int][source]¶: The (ny, nx) spatial shape of the input cubes

class flint.misc.holo.FinalFITSCubeInfo[source]¶

Simple representation of final FITS cube

data_offset: int[source]¶: The number of bytes the FITS header represents

output_shape: list[int][source]¶: The output shape of the data

path: pathlib.Path[source]¶: The output path of the fits file

class flint.misc.holo.FrequencyGrid[source]¶

Representative frequency axis information

cdelt: float[source]¶: The CDELT field

grid: numpy.typing.NDArray[numpy.floating][source]¶: The final frequency grid

max_hz: float[source]¶: The maximum frequency in Hz

min_hz: float[source]¶: The minimum frequency in Hz

class flint.misc.holo.ReprojectWorker[source]¶

beam: int[source]¶: The beam index

final_fits_cube_info: FinalFITSCubeInfo[source]¶: Information around the destination FITS cube

in_ch_idx: int[source]¶: The channel index of the input cube

in_cube_header: astropy.io.fits.Header[source]¶: The header that corresponds to plane

in_cube_path: pathlib.Path[source]¶: The cube to load

out_idx: int[source]¶: The channel index

stokes: int[source]¶: The stokes index

flint.misc.holo._direction_cosines(centre: astropy.coordinates.SkyCoord, position: astropy.coordinates.SkyCoord) → tuple[float, float][source]¶: Offset of position from centre as direction cosines (l to the east, m to the north).

flint.misc.holo._exit_scale(response: BeamResponse, dx: float, dy: float) → float[source]¶: Fraction of the pixel offset (dx, dy) from the beam centre that stays a pixel inside the cube.

flint.misc.holo._get_cube_header(fits_cube_info: FITSCubeInfo) → astropy.io.fits.Header[source]¶: Return a header object from the input cube that is compatible with reproject

flint.misc.holo._sample_planes(planes: numpy.typing.NDArray[numpy.floating], x: float, y: float) → numpy.typing.NDArray[numpy.floating][source]¶: Bilinearly interpolate every plane at one pixel position, giving 0 off the grid.

flint.misc.holo.celestial_wcs_from_header(header: astropy.io.fits.Header) → astropy.wcs.WCS[source]¶: Extract just the celestial (RA/Dec) 2D WCS from a 5D FITS header.

flint.misc.holo.cli() → None[source]¶

flint.misc.holo.concatenate_holography(concat_holo_options: ConcatHoloOptions) → pathlib.Path[source]¶

Reproject a set of ASKAP IQUV primary beam cubes into a single output cube. An optimal spatial grid is computed internally through reproject, and input cubes are placed onto a consistent channel frequency gride.

Parameters:: concat_holo_options (ConcatHoloOptions) – Options to direction the concatenation of the holography cubes.
Returns:: Path to the output cube formed
Return type:: Path

flint.misc.holo.construct_frequency_grid(fits_cube_infos: list[FITSCubeInfo]) → FrequencyGrid[source]¶

Construct the final frequency grid of the concatenated holography cube

Parameters:: fits_cube_infos (list[FITSCubeInfo]) – The sortted FITS cube meta data
Raises:: ValueError – Raised when the data do not appear in ascended order
Returns:: The final frequency grid
Return type:: FrequencyGrid

flint.misc.holo.construct_spatial_output_wcs(fits_cube_infos: list[FITSCubeInfo]) → astropy.io.fits.Header[source]¶

Construct the output WCS that will be used for each channel. The optimal WCS is find cia the reproject module.

Parameters:: fits_cube_infos (list[FITSCubeInfo]) – The set of FITS cube
Returns:: The two-dimensional FITS header
Return type:: fits.Header

flint.misc.holo.create_fits_info(cube_path: pathlib.Path, hdu_index: int = 0) → FITSCubeInfo[source]¶

Load a FITS cube container the holography and extra information from it.

Parameters:

cube_path (Path) – Cube to load
hdu_index (int, optional) – The HDU the header and data are to be drawn from

Returns:

Representative information

Return type:

FITSCubeInfo

flint.misc.holo.create_output_header(base_header: astropy.io.fits.Header, spatial_header: astropy.io.fits.Header, frequency_grid: FrequencyGrid) → astropy.io.fits.Header[source]¶

Create the final outcube cube header. This uses the header from an example FITS cube as a base, and applies appropriate keywords to indicate the new spatial and frequency dimensions

Parameters:

base_header (fits.Header) – The header to use a base to update from
spatial_header (fits.Header) – The optimal spatial header consistent across all input cubes
frequency_grid (FrequencyGrid) – The frequency axis to embed

Returns:

The final output header

Return type:

fits.Header

flint.misc.holo.create_placeholder_cube(fits_cube_infos: list[FITSCubeInfo], spatial_header: astropy.io.fits.Header, frequency_grid: FrequencyGrid, output_path: pathlib.Path) → FinalFITSCubeInfo[source]¶

Write out the final fits cube, but fill it with placeholder data.

Parameters:

fits_cube_infos (list[FITSCubeInfo]) – The set of loaded FITS cubes
spatial_header (fits.Header) – The new spatial header description
frequency_grid (FrequencyGrid) – The output frequency grid
output_path (Path) – The output path to write the cube to

Returns:

The output cube written

Return type:

FinalFITSCubeInfo

flint.misc.holo.get_freq_axis(header: astropy.io.fits.Header) → numpy.typing.NDArray[numpy.floating][source]¶: Return array of channel centre frequencies (Hz).

flint.misc.holo.get_output_data_shape(fits_cube_infos: list[FITSCubeInfo], spatial_header: astropy.io.fits.Header, frequency_grid: FrequencyGrid) → list[int][source]¶

Construct the final output shape of the concatenated holography cube

Parameters:

fits_cube_infos (list[FITSCubeInfo]) – The loaded information from the individual FITS cubes
spatial_header (fits.Header) – The new output spatial header information
frequency_grid (FrequencyGrid) – The output frequency grid

Returns:

The expected output shape of the new holography

Return type:

list[int]

flint.misc.holo.get_parser() → argparse.ArgumentParser[source]¶

flint.misc.holo.load_and_sort_cubes(cube_paths: tuple[pathlib.Path, ...]) → list[FITSCubeInfo][source]¶

Load and extra appropriate information from each input cube, and return a sorted list with increasing frequency

Parameters:: cube_paths (tuple[Path, ...]) – Holography cubes to load
Returns:: Sort list of cube information
Return type:: list[FITSCubeInfo]

flint.misc.holo.load_beam_response(holofile: pathlib.Path, beam: int, stokes: int = 0) → BeamResponse[source]¶

Read one beam’s response from a holography cube, normalised to 1 at the beam centre.

As in linmos, the spatial axes are offsets about the beam centre: only CRPIX and CDELT are used, never CRVAL. NaNs count as no response, and planes with no response at the beam centre (such as flagged channels) are dropped. The usable planes are returned in increasing frequency.

Parameters:

holofile (Path) – Holography cube with axes (beam, stokes, freq, dec, ra)
beam (int) – ASKAP beam number (0-indexed) to read
stokes (int, optional) – Stokes index to read. Defaults to 0 (Stokes I).

Raises:

ValueError – Raised if the beam centre (CRPIX) is within a pixel of the cube’s edge
UnusableBeamError – Raised if the beam has no usable planes

Returns:

The beam’s normalised response on its usable planes

Return type:

BeamResponse

flint.misc.holo.map_frequencies_to_channels(freqs_1: numpy.NDArray[numpy.floating], freqs_2: numpy.typing.NDArray[numpy.floating], tol: float) → tuple[numpy.typing.NDArray[numpy.int_], numpy.typing.NDArray[numpy.int_]][source]¶

Create mappings resolving freq_2 to channels in freqs_1

The outputs here are channel indices. The first array are the indices into the freqs_1 grid. The second set are the the elements of freqs_2 that were successfully matched.

Parameters:

freqs_1 (np.NDArray[np.floating]) – The final output frequency grid that the cube will be written to
freqs_2 (NDArray[np.floating]) – The frequency axis of the subject cube
tol (float) – The appropriate tolerance to use when mapping freqs_2 onto freqs_1 for a match to be found.

Returns:

The output channels into freqs_1, and channel indices of freqs_2 elements into freqs_1

Return type:

tuple[NDArray[np.int_], NDArray[np.int_]]

flint.misc.holo.outside_planes(response: BeamResponse, freqs: astropy.units.Quantity) → numpy.typing.NDArray[numpy.bool_][source]¶: Which freqs lie more than half a plane beyond the usable planes, and so take the nearest plane’s beam.

flint.misc.holo.reproject_cubes(fits_cube_infos: list[FITSCubeInfo], spatial_header: astropy.io.fits.Header, frequency_grid: FrequencyGrid, final_fits_cube_info: FinalFITSCubeInfo, cdelt_tol: float = 1e-06, max_workers: int = 2) → FinalFITSCubeInfo[source]¶

Reproject the input cubes onto a final output spatial grid, as defined by spatial_grid.

Parameters:

fits_cube_infos (list[FITSCubeInfo]) – The input FITS cubes to me reprojected
spatial_header (fits.Header) – The description of the final output spatial grid data will be mapped to
frequency_grid (FrequencyGrid) – The frequency axis of the output cube
cdelt_tol (float, optional) – The acceptable tolerance between matching the frequency grid of input data to the output cube. Defaults to 1e-6.
final_fits_cube_info (FinalFITSCubeInfo) – The description of the final output cube

Returns:

The final FITS cube info that was provided as input

Return type:

FinalFITSCubeInfo

flint.misc.holo.reproject_wrapper(reproject_worker: ReprojectWorker, output_projection: astropy.io.fits.Header, shape_out: tuple[int, int]) → ReprojectWorker[source]¶

flint.misc.holo.sample_beam_response(response: BeamResponse, beam_centre: astropy.coordinates.SkyCoord, position: astropy.coordinates.SkyCoord, freqs: astropy.units.Quantity, alpha: float = 0.0, taper: bool = False) → numpy.typing.NDArray[numpy.floating][source]¶

Sample a beam’s response at a sky position the same way linmos does.

The offset of position from beam_centre is rotated by alpha before it is looked up, as linmos does with ASKAP_PB.alpha. Positions off the cube give 0 unless taper is set. With taper set, a position off the cube takes the value where the line from the beam centre through it leaves the cube (a pixel inside the edge), falling off past there like a Gaussian beam.

Parameters:

response (BeamResponse) – The beam to sample, from load_beam_response
beam_centre (SkyCoord) – Direction of the beam centre, normally the MS phase centre
position (SkyCoord) – Sky position to sample at
freqs (u.Quantity) – Frequencies to interpolate the response onto
alpha (float, optional) – Rotation in radians, from flint.coadd.linmos.compute_pb_rotation_alpha. Defaults to 0.0.
taper (bool, optional) – Continue the response past the edge of the cube. Defaults to False.

Returns:

Response at each of freqs

Return type:

NDArray[np.floating]