karabo.simulation package

Subpackages

Submodules

karabo.simulation.beam module

class BeamPattern(cst_file_path: str, telescope: Telescope | None = None, freq_hz: float = 0.0, pol: Literal['x', 'Y', 'XY'] = 'XY', element_type_index: int = 0, average_fractional_error_factor_increase: float = 1.1, ignore_data_at_pole: bool = True, avg_frac_error: float = 0.8, beam_method: Literal['Gaussian Beam', 'EIDOS_AH', 'EIDOS_EM', 'KatBeam'] = 'Gaussian Beam', interpol: Literal['inter2d', 'RectBivariateSpline'] = 'RectBivariateSpline')

Bases: object

:param

static cart2pol(x: ndarray[Any, dtype[float64]], y: ndarray[Any, dtype[float64]]) → Tuple[ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]]]
static eidos_lineplot(B_ah: ndarray[Any, dtype[complex128]], B_em: ndarray[Any, dtype[complex128]], npix: int, path: str | None = None) → None
fit_elements(telescope: Telescope | None = None, freq_hz: float | None = None, pol: Literal['x', 'Y', 'XY'] | None = None, element_type_index: int | None = None, average_fractional_error_factor_increase: float | None = None, ignore_data_at_pole: bool | None = None, avg_frac_error: float | None = None) → None
static get_eidos_holographic_beam(npix: int, ch: int, dia: int, thres: int, mode: Literal['AH', 'EM'] = 'AH') → ndarray[Any, dtype[complex128]]

Returns beam

static get_meerkat_uhfbeam(f: int | float, pol: Literal['H', 'V', 'I'], beamextentx: int | float, beamextenty: int | float, sampling_step: int = 80) → Tuple[ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]]]
Parameters:

  • pol

  • beamextent

Returns:

static get_scaled_theta_phi(theta: ndarray[Any, dtype[float64]], phi: ndarray[Any, dtype[float64]], theta_em: ndarray[Any, dtype[float64]], phi_em: ndarray[Any, dtype[float64]], beam0: ndarray[Any, dtype[complex128]]) → ndarray[Any, dtype[float64]]
integrate(theta: Quantity | ndarray[Any, dtype[int64 | float64]], phi: Quantity | ndarray[Any, dtype[int64 | float64]], integrand: ndarray[Any, dtype[float64]]) → float64
make_cst_from_arr(arr: _SupportsArray[dtype[Any]] | _NestedSequence[_SupportsArray[dtype[Any]]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes], output_file_path: str) → None

Takes array of dimensions (*,8), and returns a cst files :param arr: :return: cst file with given output filename

plot_beam(theta: ndarray[Any, dtype[float64]], phi: ndarray[Any, dtype[float64]], absdir: ndarray[Any, dtype[float64]], path: str | None = None) → None
Parameters:

  • theta – in radians

  • phi – in radian

  • absdir – in DBs

Returns:

polar plot

plot_eidos_beam(path: str | None) → None
static pol2cart(rho: ndarray[Any, dtype[float64]], phi: ndarray[Any, dtype[float64]]) → Tuple[ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]]]
quad_crosspol(theta: Quantity | ndarray[Any, dtype[int64 | float64]], phi: Quantity | ndarray[Any, dtype[int64 | float64]], vcopol: ndarray[Any, dtype[float64]], voltage: bool = False, rel_power_dB: int = -40) → ndarray[Any, dtype[float64]]
save_cst_file(cstdata: ndarray[Any, dtype[float64]], telescope: Telescope | None = None) → bool
save_meerkat_cst_file(cstdata: ndarray[Any, dtype[float64]]) → None

Save CST file for MeerKat telescope for the custom beams

static show_eidos_beam(B_ah: ndarray[Any, dtype[complex128]], path: str | None = None) → None
static show_kat_beam(beampixels: ndarray[Any, dtype[float64]], beamextent: int, freq: int, pol: str, path: str | None = None) → None
Parameters:

  • beamextent

  • freq

  • pol

Returns:

sim_beam(beam_method: Literal['Gaussian Beam', 'EIDOS_AH', 'EIDOS_EM', 'KatBeam'] | None = None, f: float | None = None, fov: int | float = 30.0, interpol: Literal['inter2d', 'RectBivariateSpline'] | None = None) → Tuple[List[Quantity], ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]]]

Simulates the primary beam

Parameters:

  • beam_method – you can choose as beams: “Gaussian Beam”, “Eidos_AH”, “Eidos_EM”, “KatBeam”

  • f – the frequency for which the beam is simulated (MHz)

  • fov – when using “KatBeam” the beam will be simulated for this size of fov (radius)

sym_gaussian(theta: Quantity | ndarray[Any, dtype[int64 | float64]], phi: Quantity | ndarray[Any, dtype[int64 | float64]], freq: Quantity | ndarray[Any, dtype[int64 | float64]], diameter: Quantity | ndarray[Any, dtype[int64 | float64]], fwhm_fac: int = 1, voltage: bool = False, power_norm: int = 1) → ndarray[Any, dtype[float64]]

karabo.simulation.coordinate_helper module

east_north_to_long_lat(east_relative: float, north_relative: float, long: float, lat: float) → Tuple[float, float]

Calculate the longitude and latitude of an east-north coordinate based on some reference location.

Parameters:

  • east_relative – east coordinate in meters

  • north_relative – north coordinate in meters

  • long – reference location longitude

  • lat – reference location latitude

Returns:

Tuple of calculated longitude and latitude of passed east-north coordinate

wgs84_to_cartesian(lon: float | ndarray[Any, dtype[float64]], lat: float | ndarray[Any, dtype[float64]], alt: float | ndarray[Any, dtype[float64]], radius: int = 6371000) → ndarray[Any, dtype[float64]]

Transforms WGS84 to cartesian in meters.

Args:

lon: Longitude [deg]. lat: Latitude [deg]. alt: Altitude [m]. radius: Radius of earth in m.

Returns:

Cartesian x,y,z coordinates (nx3) in meters.

karabo.simulation.east_north_coordinate module

class EastNorthCoordinate(x: float, y: float, z: float = 0, x_error: float = 0, y_error: float = 0, z_error: float = 0)

Bases: object

Class defining an east-north-(up) coordinate.

Variables:

  • x – east coordinate in meters

  • y – north coordinate in meters

  • z – up coordinate in meters

  • x_error – east coordinate error in meters

  • y_error – north coordinate error in meters

  • z_error – up coordinate error in meters

karabo.simulation.interferometer module

class CorrelationType(value)

Bases: Enum

Enum for selecting between the different Correlation Types for the Simulator.

Auto_Correlations = 'Auto-correlations'
Both = 'Both'
Cross_Correlations = 'Cross-correlations'
class FilterUnits(value)

Bases: Enum

Enum for selecting between the different Filter Units for the Simulator.

Metres = 'Metres'
WaveLengths = 'Wavelengths'
class InterferometerSimulation(channel_bandwidth_hz: int | float = 0, time_average_sec: int | float = 0, max_time_per_samples: int = 8, correlation_type: CorrelationType = CorrelationType.Cross_Correlations, uv_filter_min: int | float = 0, uv_filter_max: int | float = inf, uv_filter_units: FilterUnits = FilterUnits.WaveLengths, force_polarised_ms: bool = False, ignore_w_components: bool = False, noise_enable: bool = False, noise_seed: str | int = 'time', noise_start_freq: int | float = 1000000000.0, noise_inc_freq: int | float = 100000000.0, noise_number_freq: int = 24, noise_rms_start: int | float = 0, noise_rms_end: int | float = 0, noise_rms: Literal['Range', 'Data file', 'Telescope model'] = 'Range', noise_freq: Literal['Range', 'Data file', 'Observation settings', 'Telescope model'] = 'Range', enable_array_beam: bool = False, enable_numerical_beam: bool = False, beam_polX: BeamPattern | None = None, beam_polY: BeamPattern | None = None, use_gpus: bool | None = None, use_dask: bool | None = None, split_observation_by_channels: bool = False, n_split_channels: int | str = 'each', client: Client | None = None, precision: Literal['single', 'double'] = 'single', station_type: Literal['Aperture array', 'Isotropic beam', 'Gaussian beam', 'VLA (PBCOR)'] = 'Isotropic beam', enable_power_pattern: bool = False, gauss_beam_fwhm_deg: int | float = 0.0, gauss_ref_freq_hz: int | float = 0.0, ionosphere_fits_path: str | None = None, ionosphere_screen_type: str | None = None, ionosphere_screen_height_km: float | None = 300, ionosphere_screen_pixel_size_m: float | None = 0, ionosphere_isoplanatic_screen: bool | None = False)

Bases: object

Class containing all configuration for the Interferometer Simulation. :ivar channel_bandwidth_hz: The channel width, in Hz, used to simulate bandwidth

smearing. (Note that this can be different to the frequency increment if channels do not cover a contiguous frequency range.)

Variables:

  • time_average_sec – The correlator time-average duration, in seconds, used to simulate time averaging smearing.

  • max_time_per_samples – The maximum number of time samples held in memory before being written to disk.

  • correlation_type – The type of correlations to produce. Any value of Enum CorrelationType

  • uv_filter_min – The minimum value of the baseline UV length allowed by the filter. Values outside this range are not evaluated

  • uv_filter_max – The maximum value of the baseline UV length allowed by the filter. Values outside this range are not evaluated.

  • uv_filter_units – The units of the baseline UV length filter values. Any value of Enum FilterUnits

  • force_polarised_ms – If True, always write the Measurement Set in polarised format even if the simulation was run in the single polarisation ‘Scalar’ (or Stokes-I) mode. If False, the size of the polarisation dimension in the Measurement Set will be determined by the simulation mode.

  • ignore_w_components – If enabled, baseline W-coordinate component values will be set to 0. This will disable W-smearing. Use only if you know what you’re doing!

  • noise_enable – If true, noise is added.

  • noise_seed – Random number generator seed.

  • noise_start_freq – The start frequency in Hz for which noise is included, if noise is set to true.

  • noise_inc_freq – The frequency increment in Hz, if noise is set to true.

  • noise_number_freq – The number of frequency taken into account, if noise is set to true.

  • noise_rms_start – Station RMS (noise) flux density range start value, in Jy. The range is expanded linearly over the number of frequencies for which noise is defined.

  • noise_rms_end – Station RMS (noise) flux density range end value, in Jy. The range is expanded linearly over the number of frequencies for which noise is defined.

  • noise_rms –

    The specifications for the RMS noise value:
    Telescope model: values are loaded from files in the telescope

    model directory.

    Data file: values are loaded from the specified file. Range: values are evaluated according to the specified range

    parameters (Default).

    The noise values are specified in Jy and represent the RMS noise of an unpolarised source in terms of flux measured in a single polarisation of the detector.

  • noise_freq –

    The list of frequencies for which noise values are defined: Telescope model: frequencies are loaded from a data file in

    the telescope model directory.

    Observation settings: frequencies are defined by the observation

    settings.

    Data file: frequencies are loaded from the specified data file. Range: frequencies are specified by the range parameters

    (Default).

  • enable_array_beam – If true, then the contribution to the station beam from the array pattern (given by beam-forming the antennas in the station) is evaluated.

  • enable_numerical_beam – If true, make use of any available numerical element pattern files. If numerical pattern data are missing, the functional type will be used instead.

  • beam_polX – currently only considered for ObservationLong

  • beam_polX – currently only considered for ObservationLong

  • use_gpus – Set to true if you want to use gpus for the simulation

  • client – The dask client to use for the simulation

  • split_idxs_per_group – The indices of the sky model to split for each group of workers. If None, the sky model will not be split. Useful if the sky model is too large to fit into the memory of a single worker. Group index should be strictly monotonic increasing.

  • precision – For the arithmetic use you can choose between “single” or “double” precision

  • station_type – Here you can choose the type of each station in the interferometer. You can either disable all station beam effects by choosing “Isotropic beam”. Or select one of the following beam types: “Gaussian beam”, “Aperture array” or “VLA (PBCOR)”

  • enable_power_pattern – If true, gauss_beam_fwhm_deg will be taken in as power pattern.

  • gauss_beam_fwhm_deg – If you choose “Gaussian beam” as station type you need specify the full-width half maximum value at the reference frequency of the Gaussian beam here. Units = degrees. If enable_power_pattern is True, gauss_beam_fwhm_deg is in power pattern, otherwise it is in field pattern.

  • gauss_ref_freq_hz – The reference frequency of the specified FWHM, in Hz.

  • ionosphere_fits_path – The path to a fits file containing an ionospheric screen generated with ARatmospy. The file parameters (times/frequencies) should coincide with the planned observation.

run_simulation(telescope: Telescope, sky: SkyModel, observation: Observation | ObservationLong, backend: Literal[SimulatorBackend.OSKAR] = SimulatorBackend.OSKAR, primary_beam: None = None, visibility_format: Literal['MS', 'OSKAR_VIS'] = 'MS', visibility_path: Path | str | None = None) → Visibility
run_simulation(telescope: Telescope, sky: SkyModel, observation: ObservationParallelized, backend: Literal[SimulatorBackend.OSKAR] = SimulatorBackend.OSKAR, primary_beam: None = None, visibility_format: Literal['MS', 'OSKAR_VIS'] = 'MS', visibility_path: Path | str | None = None) → List[Visibility]
run_simulation(telescope: Telescope, sky: SkyModel, observation: Observation, backend: Literal[SimulatorBackend.RASCIL], primary_beam: Image | None, visibility_format: Literal['MS'] = 'MS', visibility_path: Path | str | None = None) → Visibility

Run an interferometer simulation, generating simulated visibility data.

Args:

telescope: Telescope model defining the configuration sky: sky model defining the sky sources observation: observation settings backend: Simulation backend to be used primary_beam: Primary beam to be included into visibilities.

Currently only relevant for RASCIL. For OSKAR, use the InterferometerSimulation constructor parameters instead.

visibility_format: Visibility format in which to write generated data to

disk

visibility_path: Path for the visibility output file (OSKAR_VIS)

or directory (MS). If an observation of type ObservationParallelized is passed, this path will be interpreted as the root directory where the visibility files / dirs will be written to. If None, visibilities will be written to short term cache directory.

Returns:
Visibility object of the generated data or list of Visibility objects

for ObservationParallelized observations.

set_ionosphere(file_path: str) → None

Set the path to an ionosphere screen file generated with ARatmospy. The file parameters (times/frequencies) should coincide with the planned observation. see https://github.com/timcornwell/ARatmospy

Parameters:

file_path – file path to fits file.

simulate_foreground_vis(telescope: Telescope, foreground: SkyModel, foreground_observation: Observation, foreground_vis_file: str) → Tuple[Visibility, List[ndarray[Any, dtype[complex128]]], VisHeader, Binary, VisBlock, ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]]]

Simulates foreground sources

yes_double_precision() → bool

karabo.simulation.line_emission module

class CircleSkyRegion(center, radius)

Bases: tuple

center

Alias for field number 0

radius

Alias for field number 1

gaussian_beam_fwhm_for_frequency(desired_frequency: float, reference_fwhm_degrees: float = 1.8, reference_frequency_Hz: float = 800000000.0) → float
generate_gaussian_beam_data(fwhm_pixels: float, x_size: int, y_size: int) → ndarray[Any, dtype[float64]]

Given a FWHM in pixel units, and a size in x and y coordinates, return a 2D array of shape (x_size, y_size) containing normalized Gaussian values (such that the central value of the 2D array is 1.0).

line_emission_pipeline(output_base_directory: Path | str, pointings: List[CircleSkyRegion], sky_model: SkyModel, observation_details: Observation, telescope: Telescope, interferometer: InterferometerSimulation, simulator_backend: SimulatorBackend, dirty_imager_config: DirtyImagerConfig, primary_beams: List[ndarray[Any, dtype[float64]]], should_perform_primary_beam_correction: Literal[True] = False) → Tuple[List[List[Visibility]], List[List[Image]]]
line_emission_pipeline(output_base_directory: Path | str, pointings: List[CircleSkyRegion], sky_model: SkyModel, observation_details: Observation, telescope: Telescope, interferometer: InterferometerSimulation, simulator_backend: SimulatorBackend, dirty_imager_config: DirtyImagerConfig, primary_beams: List[ndarray[Any, dtype[float64]]] | None = None, should_perform_primary_beam_correction: Literal[False] = False) → Tuple[List[List[Visibility]], List[List[Image]]]
line_emission_pipeline(output_base_directory: Path | str, pointings: List[CircleSkyRegion], sky_model: SkyModel, observation_details: Observation, telescope: Telescope, interferometer: InterferometerSimulation, simulator_backend: SimulatorBackend, dirty_imager_config: DirtyImagerConfig, primary_beams: List[ndarray[Any, dtype[float64]]] | None = None, should_perform_primary_beam_correction: bool = False) → Tuple[List[List[Visibility]], List[List[Image]]]

Perform a line emission simulation, to compute visibilities and dirty images. A line emission simulation involves assuming every source in the input SkyModel only emits within one frequency channel.

If requested, include primary beam effects into the visibilities and dirty images. And again, if desired, perform a primary beam correction on the final dirty images.

For OSKAR, the provided primary beams will only be used for correction.

For the actual primary beam effect in OSKAR, set the relevant parameter in the InterferometerSimulation constructor.

For RASCIL, the provided primary beams are used

for both the primary beam effect and its correction.

karabo.simulation.line_emission_helpers module

convert_frequency_to_z(freq: _T) → _T

Turn given frequency (Hz) into corresponding redshift for 21cm emission.

Parameters:

freq – Frequency values to be converted into redshifts.

Returns:

Redshifts corresponding to input frequencies.

convert_z_to_frequency(z: _T) → _T

Turn given redshift into corresponding frequency (Hz) for 21cm emission.

Parameters:

z – Redshift values to be converted into frequencies.

Returns:

Frequencies corresponding to input redshifts.

freq_channels(z_obs: _T, channel_num: int = 10, equally_spaced_freq: bool = True) → Tuple[ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]], ndarray[Any, dtype[float64]], float64]

Calculates the frequency channels from the redshifts. :param z_obs: Observed redshifts from the HI sources. :param channel_num: Number of channels. :param equally_spaced_freq: If True (default), create channels

equally spaced in frequency. If False, create channels equally spaced in redshift.

Returns:

Redshift channels array, frequency channels array (in Hz), array of bin widths of frequency channel (in Hz), for convenience, and middle frequency (in Hz)

karabo.simulation.observation module

class Observation(*, mode: str = 'Tracking', start_frequency_hz: int | float = 0, start_date_and_time: datetime | str, length: timedelta = datetime.timedelta(seconds=14400), number_of_channels: int = 1, frequency_increment_hz: int | float = 0, phase_centre_ra_deg: int | float = 0, phase_centre_dec_deg: int | float = 0, number_of_time_steps: int = 1)

Bases: ObservationAbstract

class ObservationAbstract(*, mode: str = 'Tracking', start_frequency_hz: int | float = 0, start_date_and_time: datetime | str, length: timedelta = datetime.timedelta(seconds=14400), number_of_channels: int = 1, frequency_increment_hz: int | float = 0, phase_centre_ra_deg: int | float = 0, phase_centre_dec_deg: int | float = 0, number_of_time_steps: int = 1)

Bases: ABC

Abstract base class for observations

Holds all important information about an observation.

compute_hour_angles_of_observation() → ndarray[Any, dtype[float64]]

Given a total observation length and an integration time interval, determine the corresponding hour angles of observation. This utility function is used during simulations using the RASCIL backend. Approach based on https://gitlab.com/ska-sdp-china/rascil/-/blob/9002d853b64465238177b37e941c7445fed50d35/examples/performance/mid_write_ms.py#L32-40 # noqa: E501

static create_observations_oskar_from_lists(settings_tree: Dict[str, Dict[str, Any]], central_frequencies_hz: int | float | List[int | float], channel_bandwidths_hz: int | float | List[int | float], n_channels: int | List[int]) → List[Dict[str, Dict[str, Any]]]

Create observations for OSKAR settings from input lists. If there is a mix of different lengths of lists or single values, the missing information is repeated to match the longest list.

Parameters

settings_treeOskarSettingsTreeType

The OSKAR settings tree, with ‘observation’ key among others.

central_frequencies_hzint or list of int

List of central frequencies in MHz for each observation.

channel_bandwidths_hzint or list of int

List of channel bandwidths in MHz for each observation.

n_channelsint or list of int

List of numbers of channels for each observation.

Returns

list of dict

List of OSKAR observations, each as a dictionary with ‘observation’ key among others.

Raises

ValueError

If the input lists are not of the same length.

Notes

The ‘observation’ key in each dictionary in the returned list has a value which is itself a dictionary, with keys ‘start_frequency_hz’, ‘num_channels’, and ‘frequency_inc_hz’.

get_OSKAR_settings_tree() → Dict[str, Dict[str, Any]]

Get the settings of this observation as an oskar setting tree. This function returns a python dictionary formatted according to the OSKAR documentation. <https://fdulwich.github.io/oskarpy-doc/settings_tree.html>.

Returns:

Dictionary containing the full configuration

in the OSKAR Settings Tree format.

get_phase_centre() → List[float]
set_length_of_observation(hours: int | float, minutes: int | float, seconds: int | float, milliseconds: int | float) → None

Set a new length for the observation. Overriding the observation length set in the constructor.

Parameters:

  • hours – hours

  • minutes – minutes

  • seconds – seconds

  • milliseconds – milliseconds

class ObservationLong(*, mode: str = 'Tracking', start_frequency_hz: int | float = 0, start_date_and_time: datetime | str, length: timedelta = datetime.timedelta(seconds=14400), number_of_channels: int = 1, frequency_increment_hz: int | float = 0, phase_centre_ra_deg: int | float = 0, phase_centre_dec_deg: int | float = 0, number_of_time_steps: int = 1, number_of_days: int = 2)

Bases: ObservationAbstract

This class allows the use of several observations on different days over a certain period of time within one day. If only ONE observation is desired, even if it takes a little longer, this is already possible using Observation. This class extends Observation so its parameters (except length) are not discussed here. length is little different, which describes the duration of ONE observation, whose maximum duration for ObservationLong is 24h.

Variables:

number_of_days – Number of successive days to observe

class ObservationParallelized(*, mode: str = 'Tracking', center_frequencies_hz: int | float | List[int | float] = 100000000.0, start_date_and_time: datetime | str, length: timedelta = datetime.timedelta(seconds=14400), n_channels: int | List[int] = [0, 1, 2, 3, 4, 5], channel_bandwidths_hz: int | float | List[int | float] = [1], phase_centre_ra_deg: int | float = 0, phase_centre_dec_deg: int | float = 0, number_of_time_steps: int = 1)

Bases: ObservationAbstract

This class allows the use of several observations on different days over a certain period of time within one day. If only ONE observation is desired, even if it takes a little longer, this is already possible using Observation. This class extends Observation so its parameters (except length) are not discussed here. length is little different, which describes the duration of ONE observation, whose maximum duration for ObservationLong is 24h.

Variables:

number_of_days – Number of successive days to observe

karabo.simulation.sample_simulation module

run_sample_simulation(*, simulator_backend: SimulatorBackend = SimulatorBackend.OSKAR, visibility_format: Literal['MS', 'OSKAR_VIS'] = 'MS', verbose: bool = False) → Tuple[Visibility, List[int] | List[float], SkyModel, Telescope, Observation, InterferometerSimulation]

Creates example visibilities for use in tests, experiments and examples.

Args:

simulator_backend: Backend to use for simulation visibility_format: Visibility format in which to write generated data to disk verbose: Enable / disable progress prints

Returns:
A tuple (visibility, phase_center, sky, telescope, observation,

interferometer_sim) with the generated visibility data and phase center, sky model, telescope, observation and interferometer configuration used to generate it.

karabo.simulation.sky_model module

class Polarisation(value)

Bases: Enum

An enumeration.

STOKES_I = (0,)
STOKES_Q = (1,)
STOKES_U = (2,)
STOKES_V = 3
class SkyModel(sources: ~numpy.ndarray[~typing.Any, ~numpy.dtype[~numpy.float64]] | ~numpy.ndarray[~typing.Any, ~numpy.dtype[~numpy.object_]] | ~xarray.core.dataarray.DataArray | None = None, wcs: ~astropy.wcs.wcs.WCS | None = None, precision: ~typing.Type[~numpy.float64] = <class 'numpy.float64'>, h5_file_connection: ~h5py._hl.files.File | None = None)

Bases: object

Class containing all information of the to be observed Sky.

SkyModel.sources is a xarray.DataArray

( https://docs.xarray.dev/en/stable/generated/xarray.DataArray.html ).

np.ndarray are also supported as input type for SkyModel.sources,

however, the values in SkyModel.sources are converted to xarray.DataArray.

SkyModel.compute method is used to load the data into memory as a numpy array. It should be called after all the filtering and other operations are completed and to avoid doing the same calculation multiple them when e.g. on a cluster.

Variables:

  • sources –

    List of all point sources in the sky as xarray.DataArray. The source_ids reside in SkyModel.source_ids if provided through xarray.sources.coords with an arbitrary string key as index or np.ndarray as idx SOURCES_COLS. A single point source is described using the following col-order:

    • [0] right ascension (deg)

    • [1] declination (deg)

    • [2] stokes I Flux (Jy)

    • [3] stokes Q Flux (Jy): defaults to 0

    • [4] stokes U Flux (Jy): defaults to 0

    • [5] stokes V Flux (Jy): defaults to 0

    • [6] reference_frequency (Hz): defaults to 0

    • [7] spectral index (N/A): defaults to 0

    • [8] rotation measure (rad / m^2): defaults to 0

    • [9] major axis FWHM (arcsec): defaults to 0

    • [10] minor axis FWHM (arcsec): defaults to 0

    • [11] position angle (deg): defaults to 0

    • [12] true redshift: defaults to 0

    • [13] observed redshift: defaults to 0

    • [14] object-id: just for np.ndarray

      it is removed in the xr.DataArray and exists then in xr.DataArray.coords as index.

  • wcs – World Coordinate System (WCS) object representing the coordinate transformation between pixel coordinates and celestial coordinates (e.g., right ascension and declination).

  • precision – The precision of numerical values used in the SkyModel. Has to be of type np.float_.

  • h5_file_connection – An open connection to an HDF5 (h5) file that can be used to store or retrieve data related to the SkyModel.

COL_IDX: Dict[Literal['ra', 'dec', 'stokes_i', 'stokes_q', 'stokes_u', 'stokes_v', 'ref_freq', 'spectral_index', 'rm', 'major', 'minor', 'pa', 'true_redshift', 'observed_redshift', 'id'], int] = {'dec': 1, 'id': 14, 'major': 9, 'minor': 10, 'observed_redshift': 13, 'pa': 11, 'ra': 0, 'ref_freq': 6, 'rm': 8, 'spectral_index': 7, 'stokes_i': 2, 'stokes_q': 3, 'stokes_u': 4, 'stokes_v': 5, 'true_redshift': 12}
COL_NAME: Dict[int, Literal['ra', 'dec', 'stokes_i', 'stokes_q', 'stokes_u', 'stokes_v', 'ref_freq', 'spectral_index', 'rm', 'major', 'minor', 'pa', 'true_redshift', 'observed_redshift', 'id']] = {0: 'ra', 1: 'dec', 2: 'stokes_i', 3: 'stokes_q', 4: 'stokes_u', 5: 'stokes_v', 6: 'ref_freq', 7: 'spectral_index', 8: 'rm', 9: 'major', 10: 'minor', 11: 'pa', 12: 'true_redshift', 13: 'observed_redshift', 14: 'id'}
SOURCES_COLS = 14
add_point_sources(sources: ndarray[Any, dtype[float64]] | ndarray[Any, dtype[object_]] | DataArray) → None

Add new point sources to the sky model.

Parameters:

sources – np.ndarray with shape (number of sources, 1 + SOURCES_COLS),

where you can place the “source_id” at index SOURCES_COLS. OR an xarray.DataArray with shape (number of sources, SOURCES_COLS), where you can place the “source_id” at xarray.DataArray.coord or use SkyModel.source_ids later.

The column indices correspond to:

  • [0] right ascension (deg)

  • [1] declination (deg)

  • [2] stokes I Flux (Jy)

  • [3] stokes Q Flux (Jy): defaults to 0

  • [4] stokes U Flux (Jy): defaults to 0

  • [5] stokes V Flux (Jy): defaults to 0

  • [6] reference_frequency (Hz): defaults to 0

  • [7] spectral index (N/A): defaults to 0

  • [8] rotation measure (rad / m^2): defaults to 0

  • [9] major axis FWHM (arcsec): defaults to 0

  • [10] minor axis FWHM (arcsec): defaults to 0

  • [11] position angle (deg): defaults to 0

  • [12] true redshift: defaults to 0

  • [13] observed redshift: defaults to 0

  • [14] source id (object): is in SkyModel.source_ids if provided

close() → None

Closes the connection to the HDF5 file.

This method closes the connection to the HDF5 file if it is open and sets the h5_file_connection attribute to None.

compute() → None

Loads the lazy data into a numpy array, wrapped in a xarray.DataArray.

This method loads the lazy data from the sources into a numpy array, which is then wrapped in a xarray.DataArray object. It performs the computation necessary to obtain the actual data and stores it in the _sources attribute. After the computation is complete, it calls the close method to close the connection to the HDF5 file.

Returns: None

convert_to_backend(backend: Literal[SimulatorBackend.OSKAR] = SimulatorBackend.OSKAR, desired_frequencies_hz: Literal[None] = None, channel_bandwidth_hz: float | None = None, verbose: bool = False) → SkyModel
convert_to_backend(backend: Literal[SimulatorBackend.RASCIL], desired_frequencies_hz: ndarray[Any, dtype[float64]], channel_bandwidth_hz: float | None = None, verbose: bool = False) → List[SkyComponent]

Convert an existing SkyModel instance into a format acceptable by a desired backend. backend: Determines how to return the SkyModel source catalog.

OSKAR: return the current SkyModel instance, since methods in Karabo support OSKAR-formatted source np.array values. RASCIL: convert the current source array into a list of RASCIL SkyComponent instances.

desired_frequencies_hz: List of frequencies corresponding to start of desired frequency channels. This field is required to convert sources into RASCIL SkyComponents.

The array contains starting frequencies for the desired channels. E.g. [100e6, 110e6] corresponds to 2 frequency channels, which start at 100 MHz and 110 MHz, both with a bandwidth of 10 MHz.

channel_bandwidth_hz: Used if desired_frequencies_hz has only one element.

Otherwise, bandwidth is determined as the delta between the first two entries in desired_frequencies_hz.

verbose: Determines whether to display additional print statements.

static copy_sky(sky: _TSkyModel) → _TSkyModel
explore_sky(phase_center: ~typing.List[int] | ~typing.List[float], stokes: ~typing.Literal['Stokes I', 'Stokes Q', 'Stokes U', 'Stokes V'] = 'Stokes I', idx_to_plot: ~numpy.ndarray[~typing.Any, ~numpy.dtype[~numpy.int64]] | None = None, xlim: ~typing.Tuple[int | float, int | float] | None = None, ylim: ~typing.Tuple[int | float, int | float] | None = None, figsize: ~typing.Tuple[int | float, int | float] | None = None, title: str | None = None, xlabel: str | None = None, ylabel: str | None = None, cfun: ~typing.Callable[[...], bool | int | ~numpy.integer | float | ~numpy.floating | ~numpy.ndarray[~typing.Any, ~numpy.dtype[~numpy.int64 | ~numpy.float64]]] | None = <ufunc 'log10'>, cmap: str | None = 'plasma', cbar_label: str | None = None, with_labels: bool = False, wcs: ~astropy.wcs.wcs.WCS | None = None, wcs_enabled: bool = True, filename: str | None = None, block: bool = False, **kwargs: ~typing.Any) → None

A scatter plot of the SkyModel (self) where the sources are projected according to the phase_center

Parameters:

  • phase_center – [RA,DEC]

  • stokes – SkyModel stoke flux

  • idx_to_plot – If you want to plot only a subset of the sources, set the indices here.

  • xlim – RA-limit of plot

  • ylim – DEC-limit of plot

  • figsize – figsize as tuple

  • title – plot title

  • xlabel – xlabel

  • ylabel – ylabel

  • cfun – flux scale transformation function for scatter-coloring

  • cmap – matplotlib color map

  • cbar_label – color bar label

  • with_labels – Plots object ID’s if set?

  • wcs – If you want to use a custom astropy.wcs, ignores phase_center if set

  • wcs_enabled – Use wcs transformation?

  • filename – Set to path/fname to save figure (set extension to fname to overwrite .png default)

  • block – Whether or not plotting should block the rest of the program

  • kwargs – matplotlib kwargs for scatter & Collections, e.g. customize s, vmin or vmax

filter_by_column(col_idx: int, min_val: int | float, max_val: int | float) → SkyModel

Filters the sky based on a specific column index

Parameters:

  • col_idx – Column index to filter by

  • min_val – Minimum value for the column

  • max_val – Maximum value for the column

Return sky:

Filtered copy of the sky

filter_by_flux(min_flux_jy: int | float, max_flux_jy: int | float) → SkyModel
filter_by_frequency(min_freq: int | float, max_freq: int | float) → SkyModel
filter_by_radius(inner_radius_deg: int | float, outer_radius_deg: int | float, ra0_deg: int | float, dec0_deg: int | float, indices: Literal[False] = False) → SkyModel
filter_by_radius(inner_radius_deg: int | float, outer_radius_deg: int | float, ra0_deg: int | float, dec0_deg: int | float, indices: Literal[True]) → Tuple[SkyModel, ndarray[Any, dtype[int64]]]

Filters the sky according to an inner and outer circle from the phase center

Parameters:

  • inner_radius_deg – Inner radius in degrees

  • outer_radius_deg – Outer radius in degrees

  • ra0_deg – Phase center right ascension

  • dec0_deg – Phase center declination

  • indices – Optional parameter, if set to True,

we also return the indices of the filtered sky copy :return sky: Filtered copy of the sky

filter_by_radius_euclidean_flat_approximation(inner_radius_deg: int | float, outer_radius_deg: int | float, ra0_deg: int | float, dec0_deg: int | float, indices: Literal[False] = False) → SkyModel
filter_by_radius_euclidean_flat_approximation(inner_radius_deg: int | float, outer_radius_deg: int | float, ra0_deg: int | float, dec0_deg: int | float, indices: Literal[True]) → Tuple[SkyModel, ndarray[Any, dtype[int64]]]

Filters sources within an annular region using a flat Euclidean distance approximation suitable for large datasets managed by Xarray.

This function is designed for scenarios where the dataset size precludes in-memory spherical geometry calculations. By leveraging a flat Euclidean approximation, it permits the use of Xarray’s out-of-core computation capabilities, thus bypassing the limitations imposed by the incompatibility of astropy.visualization.wcsaxes.SphericalCircle with Xarray’s data structures. Although this method trades off geometric accuracy against computational efficiency, it remains a practical choice for large angular fields where the curvature of the celestial sphere can be reasonably neglected.

Parameters

inner_radius_degIntFloat

The inner radius of the annular search region, in degrees.

outer_radius_degIntFloat

The outer radius of the annular search region, in degrees.

ra0_degIntFloat

The right ascension of the search region’s center, in degrees.

dec0_degIntFloat

The declination of the search region’s center, in degrees.

indicesbool, optional

If True, returns the indices of the filtered sources in addition to the SkyModel object. Defaults to False.

Returns

SkyModel or tuple of (SkyModel, NDArray[np.int_])

The filtered SkyModel object, and optionally the indices of the filtered sources if indices is set to True.

Raises

KaraboSkyModelError

If the sources attribute is not populated in the SkyModel instance prior to invoking this function.

Notes

Use this function for large sky models where a full spherical geometry calculation is not feasible due to memory constraints. It is particularly beneficial when working with Xarray and Dask, facilitating scalable data analysis on datasets that are too large to fit into memory.

classmethod get_GLEAM_Sky(min_freq: float | None = None, max_freq: float | None = None) → _TSkyModel

Creates a SkyModel containing GLEAM sources from https://vizier.cfa.harvard.edu/ service with more than 300,000 sources on different frequency-bands.

The survey was created using MWA-telescope and covers the entire sky south of dec +30.

GLEAM’s frequency-range: 72-231 MHz

The required .fits file will get downloaded and cached on disk.

Args:

min_freq: Set min-frequency in Hz for pre-filtering. max_freq: Set max-frequency in Hz for pre-filtering.

Returns:

GLEAM sky as SkyModel.

classmethod get_MALS_DR1V3_Sky(min_freq: float | None = None, max_freq: float | None = None) → _TSkyModel

Creates a SkyModel containing “MALS V3” catalogue, of 715,760 sources, where the catalogue and it’s information are from ‘https://mals.iucaa.in/’.

The MeerKAT Absorption Line Survey (MALS) consists of 1655 hrs of MeerKAT time (anticipated raw data ~ 1.7 PB) to carry out the most sensitive search of HI and OH absorption lines at 0<z<2, the redshift range over which most of the cosmic evolution in the star formation rate density takes place. The MALS survey is described in ‘Gupta et al. (2016)’.

MALS sky-coverage consists of 392 sources trackings between all RA-angles and DEC[-78.80,32.35] and different radius.

MALS’s frequency-range: 902-1644 MHz.

For publications, please honor their work by citing them as follows: - If you describe MALS or associated science, please cite ‘Gupta et al. 2016’. - If you use DR1 data products, please cite ‘Deka et al. 2024’.

Args:

min_freq: Set min-frequency in Hz for pre-filtering. max_freq: Set max-frequency in Hz for pre-filtering.

Returns:

MALS sky as SkyModel.

classmethod get_MIGHTEE_Sky(min_freq: float | None = None, max_freq: float | None = None) → _TSkyModel

Creates a SkyModel containing “MIGHTEE Continuum Early Science L1 catalogue” sources from https://archive.sarao.ac.za service, consisting of 9896 sources.

MIGHTEE is an extragalactic radio-survey carried out using MeerKAT-telescope. It covers roughly the area between RA[149.40,150.84] and DEC[1.49,2.93].

MIGHTEES’s frequency-range: 1304-1375 MHz

The required .fits file will get downloaded and cached on disk.

Args:

min_freq: Set min-frequency in Hz for pre-filtering. max_freq: Set max-frequency in Hz for pre-filtering.

Returns:

MIGHTEE sky as SkyModel.

static get_OSKAR_sky(sky: ndarray[Any, dtype[float64]] | ndarray[Any, dtype[object_]] | DataArray | SkyModel, precision: Literal['single', 'double'] = 'double') → Sky

Get OSKAR sky model object from the defined Sky Model

Returns:

oskar sky model

get_cartesian_sky() → ndarray[Any, dtype[float64]]
static get_fits_catalog(path: str) → Table

Gets astropy.table.table.Table from a .fits catalog

Parameters:

path – Location of the .fits file

Returns:

fits catalog

classmethod get_random_poisson_disk_sky(min_size: Tuple[int | float, int | float], max_size: Tuple[int | float, int | float], flux_min: int | float, flux_max: int | float, r: int = 3) → _TSkyModel
classmethod get_sample_simulated_catalog() → _TSkyModel

Downloads a sample simulated HI source catalog and generates a sky model using the downloaded data. The catalog size is around 8MB.

Source: The simulated catalog data was provided by Luis Machado (https://github.com/lmachadopolettivalle) in collaboration with the ETHZ Cosmology Research Group.

Returns:

SkyModel: The corresponding sky model. The sky model contains the following information:

  • ‘Right Ascension’ (ra): The right ascension coordinates

    of the celestial objects.

  • ‘Declination’ (dec): The declination coordinates of the

    celestial objects.

  • ‘Flux’ (i): The flux measurements of the celestial objects.

  • ‘Observed Redshift’: Additional observed redshift information

    of the celestial objects.

Note: Other properties such as ‘stokes_q’, ‘stokes_u’, ‘stokes_v’,

‘ref_freq’, ‘spectral_index’, ‘rm’, ‘major’, ‘minor’, ‘pa’, and ‘id’

are not included in the sky model.

classmethod get_sky_model_from_fits(*, fits_file: Path | str, prefix_mapping: SkyPrefixMapping, unit_mapping: Dict[str, UnitBase], units_sources: SkySourcesUnits | None = None, min_freq: float | None = None, max_freq: float | None = None, encoded_freq: UnitBase | None = None, chunks: int | Literal['auto'] | Tuple[int, ...] | Mapping[Hashable, int] | None = 'auto', memmap: bool = False) → _TSkyModel

Creates a sky-model from fits_file. The following formats are supported: - Each data-array of the .fits file maps to a single SkyModel.sources column. - Frequency of some columns is encoded in col-names of the .fits file.

Args:

fits_file: The .fits file to create the sky-model from. prefix_mapping: Formattable col-names of .fits file. If encoded_freq is

not None, the freq-encoded field-values must have ‘{0}’ as placeholder.

unit_mapping: Mapping from col-unit to astropy.unit. units_sources: Units of SkyModel.sources. min_freq: Filter by min-freq in Hz? May increase file-reading significantly. max_freq: Filter by max-freq in Hz? May increase file-reading significantly. encoded_freq: Unit of col-name encoded frequency, if the .fits file has

it’s ref-frequency encoded in the col-names.

chunks: Coerce the array’s data into dask arrays with the given chunks.

Might be useful for reading larger-than-memory .fits files.

memmap: Whether to use memory mapping when opening the FITS file.

Allows for reading of larger-than-memory files.

Returns:

sky-model with according sources from fits_file.

classmethod get_sky_model_from_h5_to_xarray(path: str, prefix_mapping: SkyPrefixMapping | None = None, load_as: Literal['numpy_array', 'dask_array'] = 'dask_array', chunksize: int | Literal['auto'] = 'auto') → _TSkyModel

Load a sky model dataset from an HDF5 file and converts it to an xarray DataArray.

Parameters

pathstr

Path to the input HDF5 file.

prefix_mappingSkyPrefixMapping

Mapping column names to their corresponding dataset paths in the HDF5 file. If the column is not present in the HDF5 file, set its value to None.

load_asLiteral[“numpy_array”, “dask_array”], default=”dask_array”

What type of array to load the data inside the xarray Data Array as.

chunksizeUnion[int, str], default=auto

Chunk size for Dask arrays. This determines the size of chunks that the data will be divided into when read from the file. Can be an integer or ‘auto’. If ‘auto’, Dask will choose an optimal chunk size.

Returns

xr.DataArray

A 2D xarray DataArray containing the sky model data. Rows represent data points and columns represent different data fields (‘ra’, ‘dec’, …).

get_wcs() → WCS

Gets the currently active world coordinate system astropy.wcs For details see https://docs.astropy.org/en/stable/wcs/index.html

Returns:

world coordinate system

property num_sources: int
classmethod read_from_file(path: str) → _TSkyModel

Read a CSV file in to create a SkyModel. The CSV should have the following columns

  • right ascension (deg)

  • declination (deg)

  • stokes I Flux (Jy)

  • stokes Q Flux (Jy): if no information available, set to 0

  • stokes U Flux (Jy): if no information available, set to 0

  • stokes V Flux (Jy): if no information available, set to 0

  • reference_frequency (Hz): if no information available, set to 0

  • spectral index (N/A): if no information available, set to 0

  • rotation measure (rad / m^2): if no information available, set to 0

  • major axis FWHM (arcsec): if no information available, set to 0

  • minor axis FWHM (arcsec): if no information available, set to 0

  • position angle (deg): if no information available, set to 0

  • true redshift: defaults to 0

  • observed redshift: defaults to 0

  • source id (object): is in SkyModel.source_ids if provided

Parameters:

path – file to read in

Returns:

SkyModel

static read_healpix_file_to_sky_model_array(file: str, channel: int, polarisation: Polarisation) → Tuple[DataArray, int]

Read a healpix file in hdf5 format. The file should have the map keywords:

Parameters:

  • file – hdf5 file path (healpix format)

  • channel – Channels of observation (between 0 and maximum numbers of channels of observation)

  • polarisation – 0 = Stokes I, 1 = Stokes Q, 2 = Stokes U, 3 = Stokes V

Returns:

rechunk_array_based_on_self(array: DataArray) → DataArray
save_sky_model_as_csv(path: str) → None

Save source array into a csv. :param path: path to save the csv file in.

set_wcs(wcs: WCS) → None

Sets a new world coordinate system astropy.wcs For details see https://docs.astropy.org/en/stable/wcs/index.html

Parameters:

wcs – world coordinate system

setup_default_wcs(phase_center: List[int] | List[float] = [0, 0]) → WCS

Defines a default world coordinate system astropy.wcs For more details see https://docs.astropy.org/en/stable/wcs/index.html

Parameters:

phase_center – ra-dec location

Returns:

wcs

property shape: Tuple[int, ...]
classmethod sky_test() → _TSkyModel

Construction of a sky model which can be used for testing and visualizing the simulation with equal distributed point sources around the phase center ra=20, deg=-30.

Returns:

The test sky model.

property source_ids: DataArrayCoordinates[DataArray] | None
property sources: DataArray | None
to_np_array(with_obj_ids: bool = False) → ndarray[Any, dtype[float64]] | ndarray[Any, dtype[object_]]

Gets the sources as np.ndarray

Parameters:

with_obj_ids – Option whether object ids should be included or not

Returns:

the sources of the SkyModel as np.ndarray

to_sky_xarray(sources: ndarray[Any, dtype[float64]] | ndarray[Any, dtype[object_]] | DataArray) → DataArray
Converts a np.ndarray or xr.DataArray to SkyModel.sources

compatible xr.DataArray.

Args:

sources: Array to convert. Col-order see SkyModel description.

Returns:

SkyModel.sources compatible xr.DataArray.

write_to_file(path: str) → None
class SkyPrefixMapping(ra: str | List[str], dec: str | List[str], stokes_i: str | List[str], stokes_q: str | List[str] | None = None, stokes_u: str | List[str] | None = None, stokes_v: str | List[str] | None = None, ref_freq: str | None = None, spectral_index: str | List[str] | None = None, rm: str | List[str] | None = None, major: str | List[str] | None = None, minor: str | List[str] | None = None, pa: str | List[str] | None = None, true_redshift: str | List[str] | None = None, observed_redshift: str | List[str] | None = None, id: str | List[str] | None = None)

Bases: object

Defines the relation between col-names of a .fits file and SkyModel.sources.

Field-names of sources-data must be according to SkySourcesColName. Each _SkyPrefixMappingValueType can be a set of col-names, a single col-name. Just leave the values not provided in the .fits file None.

dec: str | List[str]
id: str | List[str] | None = None
major: str | List[str] | None = None
minor: str | List[str] | None = None
observed_redshift: str | List[str] | None = None
pa: str | List[str] | None = None
ra: str | List[str]
ref_freq: str | None = None
rm: str | List[str] | None = None
spectral_index: str | List[str] | None = None
stokes_i: str | List[str]
stokes_q: str | List[str] | None = None
stokes_u: str | List[str] | None = None
stokes_v: str | List[str] | None = None
true_redshift: str | List[str] | None = None
class SkySourcesUnits(ra: UnitBase = Unit('deg'), dec: UnitBase = Unit('deg'), stokes_i: UnitBase = Unit('Jy'), stokes_q: UnitBase = Unit('Jy'), stokes_u: UnitBase = Unit('Jy'), stokes_v: UnitBase = Unit('Jy'), ref_freq: UnitBase = Unit('Hz'), rm: UnitBase = Unit('rad / m2'), major: UnitBase = Unit('arcsec'), minor: UnitBase = Unit('arcsec'), pa: UnitBase = Unit('deg'))

Bases: object

Represents the units of SkyModel.sources.

This class is useful for unit-conversion from different Prefixes https://docs.astropy.org/en/stable/units/standard_units.html and different units which can be converted to another (like deg to arcmin).

UnitBase covers - Unit: e.g. u.Jy, u.Hz - CompositeUnit: e.g. u.Jy/u.beam, u.rad/u.m**2 - PrefixUnit: e.g. u.MHz, u.GHz

Just assign another unit to the constructor in case the default doesn’t fit for a specific field. It’s useful if e.g. stokes_i is u.Jy/u.beam in the .fits file instead.

dec: UnitBase = Unit("deg")
classmethod extract_names_and_freqs(string: str, cols: ColDefs, unit: Unit | PrefixUnit) → Dict[str, float]

Extracts all col-names and it’s encoded frequency (in Hz).

string has to be formattable, meaning it must contain a {0} for being the frequency placeholder. Otherwise it will fail.

Important: This function doesn’t consider the unit of SkyModel.sources “ref_freq” unit. It converts it to Hz regardless.

Args:

string: Column-name with {0} frequency-placeholder. cols: Columns of the .fits file to search through. unit: Frequency-unit encoded in col-names (e.g. u.MHz).

Returns:

Mapping col-name -> decoded-frequency in Hz.

classmethod format_sky_prefix_freq_mapping(cols: ColDefs, prefix_mapping: SkyPrefixMapping, encoded_freq: UnitBase | None) → Tuple[SkyPrefixMapping, int, Dict[str, float]]

Formats SkyPrefixMapping fields from str to list[str] if formattable.

The encoded frequencies and col-names of the formattable field-values get extracted from cols, converted into Hz. The extracted col-names and frequencies are then available in a col-name -> freq mapping. In addition, prefix_mapping gets enhanced with a list of according col-names of the .fits file. This function doesn’t do any formatting for each field, if encoded_freq is None or is not positional formattable.

Args:

cols: Columns of .fits file prefix_mapping: Should contain formattable field-values. encoded_freq: astropy.unit frequency encoded (e.g. u.MHz)

Raises:
RuntimeError: In case the number of formatting columns of cols

are not of equal for each frequency-channel.

Returns:
Formatted prefix_mapping, number of formatted col-names per frequency,

and a mapping col-name -> frequency (Hz).

classmethod get_pos_ids_to_ra_dec(pos_ids: Sequence[str] | str) → Dict[str, Tuple[float, float]]

Converts position-id(s) from str to ra-dec in degrees.

The supported format(s) of pos_ids is ‘JHHMMSS(.s+)±DDMMSS(.s+)’ (J2000). The format is not tested in it’s entirety to not have to do a sanity-check for each pos-id.

A valid example of pos_ids (here just str instead of an iterable) is `pos_ids`=”J130508.50-285042.0”, which results in RA≈196.285, DEC≈-28.845

Args:

pos_ids: Position-id(s)

Returns:

Dict from pos-id to ra-dec-tuple.

get_unit_scales(cols: ColDefs, unit_mapping: Dict[str, UnitBase], prefix_mapping: SkyPrefixMapping) → Dict[str, float]

Converts all units specified in prefix_mapping into a factor, which can be used for multiplication of the according data-array, to convert the SkyModel.sources into standard-units defined in the fields of this dataclass.

If a unit in the .fits file doesn’t match the default units of this dataclass, just change it during instantiation to match the .fits file unit.

Args:

cols: Columns from hdul[1].columns .fits file. unit_mapping: Mapping from col-unit (from .fits file) to `astropy.units.

Be aware to use the very-same unit. E.g. if it’s prefixed in the .fits, also prefix in the astropy.units (“MHZ”:u.MHz, NOT “MHZ”:u.Hz).

prefix_mapping: Mapping from SkyModel.sources to col-name(s).

Returns:

Mapping col-name -> scaling-factor to SkyModel.sources standard units.

classmethod is_pos_formattable(string: str) → bool

Checks if string is positional formattable.

It is positional-formattable if string contains a ‘{0}’.

Args:

string: str to check.

Returns:

Result if it’s formattable.

major: UnitBase = Unit("arcsec")
minor: UnitBase = Unit("arcsec")
pa: UnitBase = Unit("deg")
ra: UnitBase = Unit("deg")
ref_freq: UnitBase = Unit("Hz")
rm: UnitBase = Unit("rad / m2")
stokes_i: UnitBase = Unit("Jy")
stokes_q: UnitBase = Unit("Jy")
stokes_u: UnitBase = Unit("Jy")
stokes_v: UnitBase = Unit("Jy")

karabo.simulation.station module

class Station(position: EastNorthCoordinate, parent_longitude: float = 0.0, parent_latitude: float = 0.0, parent_altitude: float = 0.0)

Bases: object

add_station_antenna(antenna: EastNorthCoordinate) → None

karabo.simulation.telescope module

class Telescope(longitude: float, latitude: float, altitude: float = 0.0)

Bases: object

WGS84 longitude and latitude and altitude in metres centre of the telescope.png centre. A telescope is described as follows:

Each row represents one station, with the elements being the horizontal x (east), horizontal y (north), and horizontal z (up) coordinates, followed by the errors in horizontal y (east), horizontal y (north), and horizontal z (up). Example: [[x, y, z, error_x, error_y, error_z], […]]

centre_longitudefloat

WGS84 longitude at the center of the telescope.

centre_latitudefloat

WGS84 latitude at the center of the telescope.

centre_altitudefloat

Altitude (in meters) at the center of the telescope.

add_antenna_to_station(station_index: int, horizontal_x: float, horizontal_y: float, horizontal_z: float = 0, horizontal_x_coordinate_error: float = 0, horizontal_y_coordinate_error: float = 0, horizontal_z_coordinate_error: float = 0) → None

Add a new antenna to an existing station

Parameters:

  • station_index – Index of station to add antenna to

  • horizontal_x – east coordinate relative to the station center in metres

  • horizontal_y – north coordinate relative to the station center in metres

  • horizontal_z – altitude of antenna

  • horizontal_x_coordinate_error – east coordinate error

relative to the station center in metres :param horizontal_y_coordinate_error: north coordinate error relative to the station center in metres :param horizontal_z_coordinate_error: altitude of antenna error :return:

add_station(horizontal_x: float, horizontal_y: float, horizontal_z: float = 0.0, horizontal_x_coordinate_error: float = 0.0, horizontal_y_coordinate_error: float = 0.0, horizontal_z_coordinate_error: float = 0.0) → None

Specify the stations as relative to the centre position :param horizontal_x: east coordinate relative to centre :param horizontal_y: north coordinate relative to centre :param horizontal_z: up coordinate :param horizontal_x_coordinate_error: east coordinate error :param horizontal_y_coordinate_error: north coordinate error :param horizontal_z_coordinate_error: up coordinate error

classmethod ang_res(freq: float, b: float) → float

Calculates angular resolution in arcsec.

Angular resolution: θ=λ/B Wavelength: λ=c/f B: Max baseline in meters

Args:

freq: Frequency [Hz]. b: Max baseline in meters (e.g. from max_baseline).

Returns:

Angular resolution in arcsec.

classmethod constructor(name: Literal['ACA', 'ALMA', 'ATCA', 'CARMA', 'NGVLA', 'PDBI', 'SMA', 'VLA'], version: Enum, backend: Literal[SimulatorBackend.OSKAR] = SimulatorBackend.OSKAR) → Telescope
classmethod constructor(name: Literal['EXAMPLE', 'MeerKAT', 'ASKAP', 'LOFAR', 'MKATPlus', 'SKA1LOW', 'SKA1MID', 'VLBA', 'WSRT'], version: Literal[None] = None, backend: Literal[SimulatorBackend.OSKAR] = SimulatorBackend.OSKAR) → Telescope
classmethod constructor(name: Literal['LOWBD2', 'LOWBD2-CORE', 'LOW', 'LOWR3', 'LOWR4', 'LOW-AA0.5', 'MID', 'MIDR5', 'MID-AA0.5', 'MEERKAT+', 'ASKAP', 'LOFAR', 'VLAA', 'VLAA_north'], version: Literal[None] = None, backend: Literal[SimulatorBackend.RASCIL] = SimulatorBackend.RASCIL) → Telescope

Main constructor to obtain a pre-configured telescope instance. :param name: Name of the desired telescope configuration.

This name, together with the backend, is used as the key to look up the correct telescope specification file.

Parameters:

  • version – Version details required for some telescope configurations. Defaults to None.

  • backend – Underlying package to be used for the telescope configuration, since each package stores the arrays in a different format. Defaults to OSKAR.

Raises:

ValueError if the combination of input parameters is invalid. Specifically, if the requested telescope requires a version, but an invalid version (or no version) is provided, or if the requested telescope name is not supported by the requested backend.

Returns:

Telescope instance.

classmethod create_baseline_cut_telescope(lcut: bool | int | integer | float | floating, hcut: bool | int | integer | float | floating, tel: Telescope, tm_path: Path | str | None = None) → Tuple[Path | str, Dict[str, str]]

Cut telescope tel for baseline-lengths.

Args:

lcut: Lower cut hcut: Higher cut tel: Telescope to cut off tm_path: .tm file-path to save the cut-telescope.

tm_path will get overwritten if it already exists.

Returns:

.tm file-path & station-name conversion (e.g. station055 -> station009)

get_OSKAR_telescope() → Telescope

Retrieve the OSKAR Telescope object from the karabo.Telescope object.

Returns:

OSKAR Telescope object

get_backend_specific_information() → Path | str | Configuration
classmethod get_baselines_dists(baselines_wgs84: ndarray[Any, dtype[float64]]) → ndarray[Any, dtype[float64]]

Gets the interferometer baselines distances in meters.

It’s euclidean distance, not geodesic.

Args:

baselines_wgs84: nx3 wgs84 baselines.

Returns:

Interferometer baselines dists in meters.

get_baselines_wgs84() → ndarray[Any, dtype[float64]]

Gets the interferometer baselines in WGS84.

This function assumes that self.stations provides WGS84 coordinates.

Returns:

Baselines lon[deg]/lat[deg]/alt[m] (nx3).

get_cartesian_position() → ndarray[Any, dtype[float64]]
max_baseline() → float64

Gets the longest baseline in meters.

Returns:

Length of longest baseline.

property name: str | None

Gets the telescope name (if available).

It’s just the file-name of the referred telescope-file without the ending.

Returns:

Telescope name or None.

plot_telescope(file: str | None = None) → None

Plot the telescope according to which backend is being used, and save the resulting image into a file, if any is provided.

plot_telescope_OSKAR(file: str | None = None, block: bool = False) → None

Plot the telescope and all its stations and antennas with longitude altitude

classmethod read_OSKAR_tm_file(path: Path | str) → Telescope
write_to_disk(dir_name: Path | str, *, overwrite: bool = False) → None

Write dir_path to disk (must have .tm ending).

Parameters:

  • dir – directory in which the configuration will be saved in.

  • overwrite – If True an existing directory is overwritten if exists. Be careful to put the correct dir as input because the old one can get removed!

karabo.simulation.telescope_versions module

class ACAVersions(value)

Bases: Enum

An enumeration.

ALL = 'all'
CYCLE_1 = 'cycle1'
CYCLE_2_i = 'cycle2.i'
CYCLE_2_ns = 'cycle2.ns'
CYCLE_3 = 'cycle3'
CYCLE_4 = 'cycle4'
CYCLE_5 = 'cycle5'
CYCLE_6 = 'cycle6'
CYCLE_7 = 'cycle7'
CYCLE_7_named = 'cycle7.named'
CYCLE_8 = 'cycle8'
CYCLE_8_named = 'cycle8.named'
i = 'i'
ns = 'ns'
tp = 'tp'
class ALMAVersions(value)

Bases: Enum

ALMA Cycle 0 configuration : {link}

ALL = 'all'
CYCLE_0_COMPACT = 'cycle0.compact'
CYCLE_0_EXTENDED = 'cycle0.extended'
CYCLE_1_1 = 'cycle1.1'
CYCLE_1_2 = 'cycle1.2'
CYCLE_1_3 = 'cycle1.3'
CYCLE_1_4 = 'cycle1.4'
CYCLE_1_5 = 'cycle1.5'
CYCLE_1_6 = 'cycle1.6'
CYCLE_2_1 = 'cycle2.1'
CYCLE_2_2 = 'cycle2.2'
CYCLE_2_3 = 'cycle2.3'
CYCLE_2_4 = 'cycle2.4'
CYCLE_2_5 = 'cycle2.5'
CYCLE_2_6 = 'cycle2.6'
CYCLE_2_7 = 'cycle2.7'
CYCLE_3_1 = 'cycle3.1'
CYCLE_3_2 = 'cycle3.2'
CYCLE_3_3 = 'cycle3.3'
CYCLE_3_4 = 'cycle3.4'
CYCLE_3_5 = 'cycle3.5'
CYCLE_3_6 = 'cycle3.6'
CYCLE_3_7 = 'cycle3.7'
CYCLE_3_8 = 'cycle3.8'
CYCLE_4_1 = 'cycle4.1'
CYCLE_4_2 = 'cycle4.2'
CYCLE_4_3 = 'cycle4.3'
CYCLE_4_4 = 'cycle4.4'
CYCLE_4_5 = 'cycle4.5'
CYCLE_4_6 = 'cycle4.6'
CYCLE_4_7 = 'cycle4.7'
CYCLE_4_8 = 'cycle4.8'
CYCLE_4_9 = 'cycle4.9'
CYCLE_5_1 = 'cycle5.1'
CYCLE_5_10 = 'cycle6.10'
CYCLE_5_2 = 'cycle5.2'
CYCLE_5_3 = 'cycle5.3'
CYCLE_5_4 = 'cycle5.4'
CYCLE_5_5 = 'cycle5.5'
CYCLE_5_6 = 'cycle5.6'
CYCLE_5_7 = 'cycle5.7'
CYCLE_5_8 = 'cycle5.8'
CYCLE_5_9 = 'cycle5.9'
CYCLE_6_1 = 'cycle6.1'
CYCLE_6_10 = 'cycle6.10'
CYCLE_6_2 = 'cycle6.2'
CYCLE_6_3 = 'cycle6.3'
CYCLE_6_4 = 'cycle6.4'
CYCLE_6_5 = 'cycle6.5'
CYCLE_6_6 = 'cycle6.6'
CYCLE_6_7 = 'cycle6.7'
CYCLE_6_8 = 'cycle6.8'
CYCLE_6_9 = 'cycle6.9'
CYCLE_7_1 = 'cycle7.1'
CYCLE_7_10 = 'cycle7.10'
CYCLE_7_2 = 'cycle7.2'
CYCLE_7_3 = 'cycle7.3'
CYCLE_7_4 = 'cycle7.4'
CYCLE_7_5 = 'cycle7.5'
CYCLE_7_6 = 'cycle7.6'
CYCLE_7_7 = 'cycle7.7'
CYCLE_7_8 = 'cycle7.8'
CYCLE_7_9 = 'cycle7.9'
CYCLE_8_1 = 'cycle8.1'
CYCLE_8_10 = 'cycle8.10'
CYCLE_8_2 = 'cycle8.2'
CYCLE_8_3 = 'cycle8.3'
CYCLE_8_4 = 'cycle8.4'
CYCLE_8_5 = 'cycle8.5'
CYCLE_8_6 = 'cycle8.6'
CYCLE_8_7 = 'cycle8.7'
CYCLE_8_8 = 'cycle8.8'
CYCLE_8_9 = 'cycle8.9'
OUT_01 = 'out01'
OUT_02 = 'out02'
OUT_03 = 'out03'
OUT_04 = 'out04'
OUT_05 = 'out05'
OUT_06 = 'out06'
OUT_07 = 'out07'
OUT_08 = 'out08'
OUT_09 = 'out09'
OUT_10 = 'out10'
OUT_11 = 'out11'
OUT_12 = 'out12'
OUT_13 = 'out13'
OUT_14 = 'out14'
OUT_15 = 'out15'
OUT_16 = 'out16'
OUT_17 = 'out17'
OUT_18 = 'out18'
OUT_19 = 'out19'
OUT_20 = 'out20'
OUT_21 = 'out21'
OUT_22 = 'out22'
OUT_23 = 'out23'
OUT_24 = 'out24'
OUT_25 = 'out25'
OUT_26 = 'out26'
OUT_27 = 'out27'
OUT_28 = 'out28'
class ATCAVersions(value)

Bases: Enum

An enumeration.

ALL = 'all'
class CARMAVersions(value)

Bases: Enum

Combined Array for Research in Millimeter-wave Astronomy telescope configuration names

A = 'a'
B = 'b'
C = 'c'
D = 'd'
E = 'e'
class MWAVersion(value)

Bases: Enum

An enumeration.

ONE = '1'
TWO_COMPACT = '2compact'
TWO_EXTENDED = '2ext'
class NGVLAVersions(value)

Bases: Enum

next-generation Very Large Array telescope configuration names

CORE_rev_B = 'core-revB'
CORE_rev_C = 'core-revC'
gb_vlba_rev_B = 'gb-vlba-revB'
lba_rev_C = 'lba-revC'
main_rev_C = 'main-revC'
mid_subarray_rev_C = 'mid-subarray-revC'
plains_rev_B = 'plains-revB'
plains_rev_C = 'plains-revC'
rev_B = 'revB'
rev_C = 'revC'
sba_rev_B = 'sba-revB'
sba_rev_C = 'sba-revC'
class PDBIVersions(value)

Bases: Enum

An enumeration.

A = 'a'
B = 'b'
C = 'c'
D = 'd'
class SMAVersions(value)

Bases: Enum

An enumeration.

COMPACT = 'compact'
COMPACT_N = 'compact.n'
EXTENDED = 'extended'
SUBCOMPACT = 'subcompact'
VEXTENDED = 'vextended'
class VLAVersions(value)

Bases: Enum

An enumeration.

A = 'a'
B = 'b'
B_n_a = 'bna'
C = 'c'
C_n_b = 'cnb'
D = 'd'
D_n_c = 'dnc'

karabo.simulation.visibility module

class Visibility(path: Path | str)

Bases: object

Class representing visibility data on the filesystem

Args:
path: Path to visibility data directory (for MS format) or file.

Visibility format will be inferred from the path.

classmethod combine_vis(visibilities: List[Visibility], combined_ms_filepath: Path | str | None = None, group_by: str = 'day') → Path | str
class VisibilityFormatUtil

Bases: object

classmethod is_valid_path_for_format(path: Path | str, format: Literal['MS', 'OSKAR_VIS']) → bool

Tests if a path is valid for a specific format

Args:

path: path to visibility format: visibility format

Returns:

True if valid, False otherwise

classmethod parse_visibility_format_from_path(path: Path | str) → Literal['MS', 'OSKAR_VIS'] | None

Module contents