API Reference

Public Python API for skasim configuration.

class skasim.ImgConfig(**data)[source]

Bases: BaseModel

imaging parameters passed to OSKAR / WSClean.

WSClean-specific flags (mgain, multiscale, auto-threshold, etc.) are ignored when imager is oskar-dirty; they only affect argv building for wsclean.

Geometry triplet: fov_deg, pixels, cell_size_arcsec are linked. Any two determine the third. A fully specified triplet must satisfy fov_deg == pixels * cell_size_arcsec / 3600 within a relative tolerance of 1e-6. Omitting all three selects the legacy defaults and emits a deprecation warning.

auto_mask: Optional[float]

auto_threshold: Optional[float]

cell_size_arcsec: Optional[float]

channels_out: Optional[int]

clean_iterations: int

fov_deg: Optional[float]

imager: Literal['oskar-dirty', 'wsclean']

join_channels: Optional[bool]

local_rms: Optional[bool]

mgain: Optional[float]

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

multiscale: Optional[bool]

multiscale_scales: Optional[list[int]]

padding: Optional[float]

pixels: int

resolve_geometry(diffraction_fov_deg, theoretical_beam_arcsec, reference_frequency_hz)[source]

Resolve this block’s geometry against the observation-level beam.

Return type:: ImageGeometry

robust: float

tag: str

threads: Optional[int]

wsclean_command: str

wsclean_predict_command: Optional[str]

class skasim.ObsConfig(**data)[source]

Bases: BaseModel

observation parameters for Karabo.

bandwidth_mhz: Optional[float]

channel_width_mhz: Optional[float]

frequency_mhz: float

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n_channels: Optional[int]

observation_time_s: int

phase_center_dec_deg: Optional[float]

phase_center_ra_deg: Optional[float]

start_time: Optional[datetime]

class skasim.SimConfig(**data)[source]

Bases: BaseModel

simulation settings.

catalog: Optional[CatalogName]

center: Optional[str]

column_mapping: Optional[str]

description: Optional[str]

fits_image: Optional[str]

flux_scale: float

imaging: list[ImgConfig]

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

models: list[ModelEntry]

noise_rms_end: Optional[float]

noise_rms_start: Optional[float]

observation: ObsConfig

output_dir: Optional[str]

overwrite: bool

rms: bool

rms_sigma: float

rms_value: float

shadems_command: str

sky_file: Optional[str]

sky_format: Literal['auto', 'fits', 'json', 'pickle', 'random']

source_flux_jy: Optional[list[float]]

stokes_q_jy: Optional[list[float]]

stokes_u_jy: Optional[list[float]]

stokes_v_jy: Optional[list[float]]

telescope: str

telescope_version: Optional[str]

title: Optional[str]

uv_coverage: bool

uv_coverage_canvas_size: int

skasim.config

class skasim.config.CasaTaylorTermsModelEntry(**data)[source]

Bases: BaseModel

Existing CASA Taylor-term image model set.

This entry type currently only supports the legacy CASA ft backend. A wsclean_predict backend is planned but not yet implemented; the injection_backend field is reserved for future use and defaults to casa_ft.

injection_backend: Literal['casa_ft']

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reference_frequency_hz: float

tt0: str

tt1: Optional[str]

type: Literal['casa_taylor_terms']

class skasim.config.ComponentSkyModelEntry(**data)[source]

Bases: BaseModel

Existing catalog/component sky-model entry.

catalog: Optional[CatalogName]

column_mapping: Optional[str]

flux_scale: float

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path: Optional[str]

sky_format: Literal['auto', 'fits', 'json', 'pickle', 'random']

type: Literal['component_sky_model']

class skasim.config.ContinuumIAlphaModelEntry(**data)[source]

Bases: BaseModel

Continuum image model with a spatially varying spectral index.

alpha: str

injection_backend: Literal['wsclean_predict', 'casa_ft']

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reference_frequency_hz: float

stokes_i: str

type: Literal['continuum_i_alpha']

class skasim.config.ImgConfig(**data)[source]

Bases: BaseModel

imaging parameters passed to OSKAR / WSClean.

WSClean-specific flags (mgain, multiscale, auto-threshold, etc.) are ignored when imager is oskar-dirty; they only affect argv building for wsclean.

Geometry triplet: fov_deg, pixels, cell_size_arcsec are linked. Any two determine the third. A fully specified triplet must satisfy fov_deg == pixels * cell_size_arcsec / 3600 within a relative tolerance of 1e-6. Omitting all three selects the legacy defaults and emits a deprecation warning.

auto_mask: Optional[float]

auto_threshold: Optional[float]

cell_size_arcsec: Optional[float]

channels_out: Optional[int]

clean_iterations: int

fov_deg: Optional[float]

imager: Literal['oskar-dirty', 'wsclean']

join_channels: Optional[bool]

local_rms: Optional[bool]

mgain: Optional[float]

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

multiscale: Optional[bool]

multiscale_scales: Optional[list[int]]

padding: Optional[float]

pixels: int

resolve_geometry(diffraction_fov_deg, theoretical_beam_arcsec, reference_frequency_hz)[source]

Resolve this block’s geometry against the observation-level beam.

Return type:: ImageGeometry

robust: float

tag: str

threads: Optional[int]

wsclean_command: str

wsclean_predict_command: Optional[str]

class skasim.config.ObsConfig(**data)[source]

Bases: BaseModel

observation parameters for Karabo.

bandwidth_mhz: Optional[float]

channel_width_mhz: Optional[float]

frequency_mhz: float

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n_channels: Optional[int]

observation_time_s: int

phase_center_dec_deg: Optional[float]

phase_center_ra_deg: Optional[float]

start_time: Optional[datetime]

class skasim.config.SimConfig(**data)[source]

Bases: BaseModel

simulation settings.

catalog: Optional[CatalogName]

center: Optional[str]

column_mapping: Optional[str]

description: Optional[str]

fits_image: Optional[str]

flux_scale: float

imaging: list[ImgConfig]

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

models: list[ModelEntry]

noise_rms_end: Optional[float]

noise_rms_start: Optional[float]

observation: ObsConfig

output_dir: Optional[str]

overwrite: bool

rms: bool

rms_sigma: float

rms_value: float

shadems_command: str

sky_file: Optional[str]

sky_format: Literal['auto', 'fits', 'json', 'pickle', 'random']

source_flux_jy: Optional[list[float]]

stokes_q_jy: Optional[list[float]]

stokes_u_jy: Optional[list[float]]

stokes_v_jy: Optional[list[float]]

telescope: str

telescope_version: Optional[str]

title: Optional[str]

uv_coverage: bool

uv_coverage_canvas_size: int

class skasim.config.SpectralCubeModelEntry(**data)[source]

Bases: BaseModel

3D spectral-line cube (RA, Dec, FREQ) in Stokes I.

The cube describes the sky and may have a much finer spectral grid than the observation. Validation only checks that the observed channel centres lie inside the cube’s frequency extent and that the spatial dimensions match the imaging configuration. The optional reference_frequency_hz, channel_width_hz and n_channels fields are accepted for backward compatibility but are ignored; the FITS header is authoritative.

channel_width_hz: Optional[float]

cube: str

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n_channels: Optional[int]

reference_frequency_hz: Optional[float]

type: Literal['spectral_cube']

class skasim.config.StaticStokesMapsModelEntry(**data)[source]

Bases: BaseModel

Static Stokes map set. Currently Stokes I only via wsclean_predict.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

stokes_i: str

stokes_q: Optional[str]

stokes_u: Optional[str]

stokes_v: Optional[str]

type: Literal['static_stokes_maps']

skasim.config.has_spectral_cube_model(config)[source]

Return True if the simulation contains a spectral_cube model entry.

Return type:: bool

skasim.config.spectral_cube_model_entries(config)[source]

Return all spectral_cube model entries in the config.

Return type:: list

skasim.cli

cli.py — command-line entry point (adapted from synthsim.py)

skasim.cli.main(argv=None)[source]

Return type:: None

skasim.pipeline

pipeline.py — end-to-end interferometric simulation orchestrator.

skasim.pipeline.build_observation(ctx, center, telescope, sky_center=None)[source]

Return (observation, frequency, bandwidth, n_channels, delta_freq, start_freq).

sky_center is the sky model’s own centre, before any SimConfig.center override is applied to center. When omitted it defaults to center, so a caller that doesn’t track the pre-override centre simply reports no drift.

Return type:: tuple

skasim.pipeline.build_sky_model(ctx, fov)[source]

Return (sky_model, center).

Return type:: tuple[SkyModel, SkyCoord]

skasim.pipeline.build_telescope(ctx)[source]: return a Karabo Telescope instance.

skasim.pipeline.build_zero_flux_sky_model(center)[source]

Build a one-source zero-flux sky model for base-MS fallback creation.

Return type:: SkyModel

skasim.pipeline.compute_fov(telescope, fov_deg, frequency)[source]

return FoV in radians. If fov_deg is set, use it; else diffraction limit.

Return type:: Quantity

skasim.pipeline.maximum_baseline_m(telescope)[source]

Return Karabo’s maximum antenna separation in metres.

Return type:: float

skasim.pipeline.parse_center(center_str, fallback)[source]

Return type:: SkyCoord

skasim.pipeline.resolve_telescope_version(telescope_module, telescope, version)[source]: Resolve a CLI/config telescope version string to Karabo’s enum member.

skasim.pipeline.run(config)[source]

Execute the full simulation pipeline from a SimConfig.

Return type:: None

skasim.pipeline.run_simulation(ctx, telescope, observation, sky_model, fov_sim)[source]

Run InterferometerSimulation and return visibility path.

Return type:: Path

skasim.pipeline.source_ref_get_best_observation_time(center, telescope)[source]: Wrapper around Source.get_best_observation_time using a dummy Source.

skasim.imaging

imaging.py — dirty (OSKAR) and cleaned (WSClean) imaging wrappers.

skasim.imaging.build_wsclean_argv(img_config, visibility_path, fov, output_prefix, n_channels=1)[source]

Build a shell-free WSClean argv list from the resolved imaging config.

Return type:: list[str]

skasim.imaging.collect_wsclean_outputs(work_dir, output_prefix)[source]

Collect WSClean FITS outputs for one configured output prefix.

Return type:: list[Path]

skasim.imaging.run_dirty_imaging(ctx, visibility_path, fov, center, img_config, sub_dir)[source]

produce dirty image via OSKAR.

Return type:: None

skasim.imaging.run_wsclean_command(argv, work_dir)[source]

Run WSClean with argv and an explicit working directory.

Streams stdout/stderr line-by-line through loguru so that WSClean progress appears in the skasim logs in real time.

skasim.imaging.run_wsclean_imaging(ctx, visibility_path, fov, img_config, sub_dir, n_channels=1)[source]

produce cleaned image via external WSClean binary.

Return type:: None

skasim.imaging.stack_channels(channel_paths, output_path)[source]

Stack WSClean per-channel FITS images into a single 3D spectral cube.

Uses the same logic as scripts/wsclean_channels_to_cube.py but inlined here so it works without putting scripts/ on PYTHONPATH.

Return type:: None

skasim.imaging.write_fits_preview(img_path, png_path, title, recenter=None, scale_factor=1000.0, bunit='mJy/beam', colorbar_label='mJy/beam')[source]

Write a publication-style PNG preview for a WSClean FITS image, optionally recentered.

Return type:: None

skasim.imaging.write_psf_profile_preview(psf_path, png_path, title='Point spread function')[source]

Write a PSF preview with 1D x/y cuts through the peak, to judge gaussianity.

The 2D panel shows the PSF core; the two line panels are slices through the peak along x and y so an asymmetric or non-Gaussian main lobe (a common sign of poor UV coverage or excessive weighting) is visible by eye.

Return type:: None

skasim.imaging.write_sky_model_previews(sky_model, center, fov, work_dir, run_id)[source]

Write full and FoV sky-model source previews.

Return type:: list[tuple[str, str]]

skasim.imaging.wsclean_output_prefix(ctx)[source]

Return the stable WSClean output prefix for this run.

Return type:: str

skasim.sky

class skasim.sky.SkyModel(*args: Any, **kwargs: Any)[source]

Bases: SkyModel

static from_fits(fits_file, total_intensity=<Quantity 1. Jy>, fov=<Quantity 1. deg>, frequency=<Quantity 1. GHz>, log_file='sky_model.log', prefix='sky_model', t0=0)[source]: Load a SkyModel from a FITS file. Parameters: - fits_file: Path to the FITS file. Returns: - SkyModel object.

static from_fits_table(fits_file, log_file='sky_model.log', prefix='sky_model')[source]: Load a SkyModel from a FITS table file. Parameters: - fits_file: Path to the FITS file. Returns: - SkyModel object.

static from_json(json_data)[source]

get_center(sources=None)[source]

Return type:: SkyCoord

phase_center = None

show(**kwargs)[source]

to_json()[source]

class skasim.sky.Source(ra, dec, I, Q=<Quantity 0. Jy>, U=<Quantity 0. Jy>, V=<Quantity 0. Jy>, ref_freq=<Quantity 0. Hz>, spec_index=0, rot_meas=<Quantity 0. rad / m2>, major_axis=<Quantity 0. arcsec>, minor_axis=<Quantity 0. arcsec>, pa=<Quantity 0. deg>, true_redshift=0, obs_redshift=0, obj_id=None, resolved=False, isl_rms=<Quantity 0. Jy>)[source]

Bases: object

coords(frame='icrs')[source]

property flux

static from_array(array, colnames=['ra', 'dec', 'I', 'Q', 'U', 'V', 'ref_freq', 'spec_index', 'rot_meas', 'major_axis', 'minor_axis', 'pa', 'true_redshift', 'obs_redshift', 'resolved', 'isl_rms'])[source]

static from_json(json_data)[source]

static from_name(name)[source]

static from_sky_model(data)[source]: Reconstruct Source from 14-element tuple (inverse of to_sky_model).

static from_table_in_fits(table)[source]

get_best_observation_time(telescope, date=None)[source]

Returns the local time at which an object with a given RA/Dec culminates (best observation time).

Parameters: - ra_hours: Right Ascension in hours (float) - dec_degrees: Declination in degrees (float) - lat_deg: Observer’s latitude in degrees (float) - lon_deg: Observer’s longitude in degrees (float, positive to the East) - elevation_m: Altitude above sea level (optional) - date: Date as a string ‘YYYY-MM-DD’ (optional, defaults to today if not provided) - timezone_offset: Time difference relative to UTC (e.g., -6 for CDMX)

Returns: - Best time.

get_flux(freq=None, alpha=None)[source]

to_fits_fmt()[source]

to_json(coords_fmt='deg')[source]

to_sky_model(reduced_form=False)[source]

skasim.manifest

manifest.py — Pydantic models for run manifest, milestone tracking, and pipeline context.

class skasim.manifest.Milestone(**data)[source]

Bases: BaseModel

single checkpoint in a simulation run.

details: dict

elapsed_s: Optional[float]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

status: Literal['started', 'completed', 'failed']

timestamp_utc: datetime

class skasim.manifest.OutputRecord(**data)[source]

Bases: BaseModel

One output produced by a run.

image_product_id: Optional[str]

imager: Optional[str]

kind: OutputKind

metadata: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path: str

role: Optional[str]

class skasim.manifest.RunContext(**data)[source]

Bases: BaseModel

passed through all pipeline functions; bundles config, paths, and manifest.

add_milestone(*args, **kwargs)[source]

convenience: add milestone to manifest and persist to disk.

Return type:: Milestone

config: SimConfig

log_path: Path

manifest: RunManifest

manifest_path: Path

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

save_manifest()[source]

write the current manifest state to disk (overwrites).

Return type:: None

sky_file_resolved: Optional[Path]

visibility_path: Path

weblog_path: Path

work_dir: Path

class skasim.manifest.RunManifest(**data)[source]

Bases: BaseModel

canonical machine-readable record of one simulation run.

add_milestone(name, status, elapsed_s=None, details=None)[source]

append a milestone and return it.

Return type:: Milestone

add_output(kind, path, image_product_id=None, imager=None, role=None, metadata=None)[source]

append a structured output record and return it.

Return type:: OutputRecord

completed_at: Optional[datetime]

config: SimConfig

errors: list[str]

invocation: Optional[list[str]]

mark_completed()[source]

mark the run as completed.

Return type:: None

mark_failed(error)[source]

mark the run as failed and record the error.

Return type:: None

milestones: list[Milestone]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_dump_json(**kwargs)[source]

serialize to pretty-printed JSON string.

Return type:: str

outputs: list[OutputRecord]

run_id: str

started_at: datetime

status: Literal['running', 'completed', 'failed']

skasim.manifest.create_run_context(config)[source]

create work_dir, init logger, build RunContext with empty manifest.

Return type:: RunContext

skasim.runtime

Runtime dependency helpers.

exception skasim.runtime.CasacoreRuntimeError[source]

Bases: RuntimeError

Raised when CASA image-table manipulation needs python-casacore but it is unavailable.

exception skasim.runtime.KaraboRuntimeError[source]

Bases: RuntimeError

Raised when full simulation execution needs Karabo but it is unavailable.

exception skasim.runtime.OskarRuntimeError[source]

Bases: RuntimeError

Raised when FITS image ingestion needs OSKAR but it is unavailable.

skasim.runtime.require_casacore()[source]: Import casacore.tables and provide a clear runtime error if unavailable.

skasim.runtime.require_karabo_module(module_name)[source]

Import a Karabo module or raise the supported runtime setup message.

Return type:: ModuleType

skasim.runtime.require_oskar_module()[source]

Import the top-level oskar module or raise a clear error.

Return type:: ModuleType

skasim.utils

helpers extracted from legacy scripts/utils.py.

class skasim.utils.NpEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: JSONEncoder

default(obj)[source]

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return super().default(o)

skasim.utils.build_shadems_uv_coverage_argv(shadems_command, visibility_path, output_dir, png_name, title, canvas_size=600)[source]

Build a shell-free shadeMS argv list for a U/V coverage plot.

Return type:: list[str]

skasim.utils.define_extra_units()[source]

Return type:: None

skasim.utils.get_diameter(telescope_name)[source]

skasim.utils.init_logger(log_file=None)[source]

Cconfigure loguru: stderr + optional log file.

Return type:: None

skasim.utils.mapping_unit(unit_str)[source]

Return type:: Optional[str]

skasim.utils.run_shadems_command(argv, work_dir)[source]: Run shadeMS with argv, an explicit cwd, and writable cache directories.

skasim.utils.shadems_uv_coverage_env(work_dir)[source]

Return an environment with writable cache directories for shadeMS imports.

Return type:: dict[str, str]

skasim.loaders

skasim.loaders — Loaders that produce SkyModel from external sources.

class skasim.loaders.CasaModelProduct(model_paths, nterms, reffreq, intermediates, cube_data=None, header=None, freq_axis=None, model_dir=None)[source]

Bases: object

CASA-ready model product generated from one model entry.

cube_data: ndarray | None = None

freq_axis: int | None = None

header: Header | None = None

intermediates: list[Path]

model_dir: Path | None = None

model_paths: list[Path]

nterms: int

reffreq: str

class skasim.loaders.FitsCatalogLoader(fpath, column_mapping, scale_I=1.0, ref_freq_hz=None, frequency=None)[source]

Bases: object

Load a SkyModel from a FITS table using a column_mapping string.

Automatically falls back to a custom loader (no Karabo) when mapped columns lack TUNIT in the FITS header. When all mapped columnsprovide TUNIT, the path via SkyPrefixMapping + SkySourcesUnits is used.

has_missing_unit()[source]

Return True if any mapped column that requires a unit lacks it.

Only positions 1-12 are checked, with 0 (id) and 7 (spectral_index) ignored as dimensionless

Return type:: bool

load()[source]

Return type:: SkyModel

class skasim.loaders.FitsCubeInfo(path, shape, spatial_shape, unit, n_channels, channel_width_hz, start_frequency_hz, reference_frequency_hz)[source]

Bases: object

Small summary of an accepted 3D spectral-cube model.

channel_width_hz: float

n_channels: int

path: Path

reference_frequency_hz: float

shape: tuple[int, int, int]

spatial_shape: tuple[int, int]

start_frequency_hz: float

unit: str

class skasim.loaders.FitsImageInfo(path, spatial_shape, unit, celestial_header, center)[source]

Bases: object

Small summary of one accepted FITS image model plane.

celestial_header: dict[str, object]

center: SkyCoord | None

path: Path

spatial_shape: tuple[int, int]

unit: str | None

class skasim.loaders.FitsImageLoader(fpath, fallback_freq_mhz=None)[source]

Bases: object

Load a SkyModel from a FITS image using oskar.Sky.

The protocol is:: oskar.Sky.from_fits_file(path) → to_array() → np.ndarray → SkyModel

This loader encapsulates the entire OSKAR dependency, allowing lazy import via require_oskar_module so that the module can be imported even when OSKAR is not installed.

load()[source]

Convert the FITS image to a Karabo SkyModel.

Return type:: SkyModel

skasim.loaders.adjust_spectral_reference(image_path, old_ref_hz, new_ref_hz, alpha_map=None)[source]

Adjust the spectral reference of a CASA image to the observation band centre.

For nterms=1 (alpha_map is None), set CRVAL4 to new_ref_hz only — the model is spectrally flat and no pixel-data correction is needed.

For nterms≥2 (alpha_map provided), correct the pixel data element-wise following CASA’s Taylor-series convention:

tt0'(x,y) = tt0(x,y) · (ν_new / ν_old) ^ α(x,y)

where α(x,y) = tt1(x,y) / tt0(x,y) for each pixel. CRVAL4 is also set.

Returns the adjusted reference frequency in Hz (always new_ref_hz).

Return type:: float

skasim.loaders.component_model_entries(config)[source]

Return type:: list[ComponentSkyModelEntry]

skasim.loaders.has_spectral_cube_model(config)[source]

Return True if the simulation contains a spectral_cube model entry.

Return type:: bool

skasim.loaders.image_model_center(entries)[source]

Return the centre of the first image model with usable celestial WCS.

Return type:: SkyCoord | None

skasim.loaders.image_model_entries(config)[source]

Return type:: list[Union[ComponentSkyModelEntry, ContinuumIAlphaModelEntry, CasaTaylorTermsModelEntry, StaticStokesMapsModelEntry, SpectralCubeModelEntry]]

skasim.loaders.inject_image_models(ctx, visibility_path, imaging_configs=None)[source]

Inject configured image models into an existing Measurement Set.

Return type:: None

skasim.loaders.inject_spectral_cube_with_wsclean_predict(ctx, entry, index, visibility_path, img_config, cube_data, header, freq_axis)[source]

Write per-channel FITS and run wsclean -predict for a spectral cube.

Return type:: dict

skasim.loaders.merge_model_data_into_data(visibility_path)[source]

Add image-model MODEL_DATA into the delivered DATA column.

Return type:: None

skasim.loaders.prepare_continuum_i_alpha_for_casa(ctx, entry, index)[source]

Create CASA image products for a continuum I+alpha model.

Adjusts the spectral reference to the observation band centre using the explicit spectral index from the model entry.

Return type:: CasaModelProduct

skasim.loaders.prepare_spectral_cube_for_casa(ctx, entry, index, report)[source]

Resample a 3D FITS spectral cube to the MS spectral grid.

Returns the resampled cube array in numpy order (freq, dec, ra) together with a 3D FITS header that describes the per-channel model images. The caller is responsible for splitting this into WSClean per-channel model images and running wsclean -predict.

Return type:: CasaModelProduct

skasim.loaders.primary_model_fits_path(entry)[source]

Return the representative FITS image for previews and phase-centre inference.

Return type:: Path | None

skasim.loaders.read_fits_cube_info(path)[source]

Read metadata from a 3D FITS spectral cube.

Return type:: FitsCubeInfo

skasim.loaders.read_fits_image_info(path)[source]

Read FITS image metadata used by validation and reporting.

Return type:: FitsImageInfo

skasim.loaders.run_casa_exportfits(work_dir, imagename, fitsimage)[source]

Run CASA exportfits in batch mode for a CASA image table.

Return type:: None

skasim.loaders.run_casa_ft(visibility_path, model_paths, nterms, reffreq, incremental)[source]

Run CASA ft into MODEL_DATA for one prepared model entry.

Return type:: None

skasim.loaders.run_casa_set_spectral_coordinate(work_dir, image_paths, frequency_hz)[source]

Set the single-channel spectral coordinate of CASA images to the run reference.

Return type:: None

skasim.loaders.run_wsclean_predict(wsclean_command, visibility_path, img_config, prefix, n_channels, work_dir, pixel_size_arcsec=None, n_pixels=None)[source]

Run WSClean in predict mode to fill MODEL_DATA of the MS.

Return type:: None

skasim.loaders.spectral_cube_model_entries(config)[source]

Return all spectral_cube model entries in the config.

Return type:: list

skasim.loaders.validate_continuum_i_alpha(entry)[source]

Validate the continuum image contract and return report metadata.

Return type:: dict

skasim.loaders.validate_spectral_cube(entry, obs, img)[source]

Validate the spectral-cube contract against observation/imaging config.

The cube’s spatial dimensions are checked against the user’s explicit pixels setting when present. If pixels was not supplied (e.g. only fov_deg was given and skasim derives the image size automatically), the cube dimensions are accepted as the authoritative model geometry and a warning is recorded so the run log remains traceable.

Return type:: dict

skasim.loaders.validate_static_stokes_maps(entry, obs, img_config)[source]

Validate a static Stokes I map and return report metadata.

Currently only stokes_i is supported. The image must be a 2D spatial FITS file with Jy/pixel-compatible BUNIT.

Return type:: dict

skasim.loaders.write_image_model_previews(ctx, fov, center=None)[source]

Write two-panel FITS model previews for the weblog sky-model section.

Each continuum/Stokes-map/CASA-Taylor-term entry produces a side-by-side PNG: LEFT shows the FITS at its natural image extent so small models are legible; RIGHT shows the simulation FoV contextual view with the primary beam as a circle and the image-model footprint as a rectangle. Spectral cubes keep the existing single-panel moment-8 preview.

Return type:: None

Parameters

ctx: Run context with the model configuration and manifest.
fov: Simulation field of view used for the right-hand contextual panel.
center: Simulation phase centre. When None, the contextual panel falls back to the FITS model’s own centre if it has a usable WCS.

skasim.loaders.write_spectral_cube_input_preview(ctx, fov)[source]

Render a peak-intensity (moment-8) PNG preview of the input spectral cube.

The preview is added to the manifest as a sky-model plot so it appears in the weblog’s Sky Model section.

Return type:: None

skasim.loaders.fits_catalogue

loaders/fits_catalogue.py — FITS catalogue loading helpers.

Loads FITS table sky models using an explicit column mapping.

class skasim.loaders.fits_catalogue.FitsCatalogLoader(fpath, column_mapping, scale_I=1.0, ref_freq_hz=None, frequency=None)[source]

Bases: object

Load a SkyModel from a FITS table using a column_mapping string.

Automatically falls back to a custom loader (no Karabo) when mapped columns lack TUNIT in the FITS header. When all mapped columnsprovide TUNIT, the path via SkyPrefixMapping + SkySourcesUnits is used.

has_missing_unit()[source]

Return True if any mapped column that requires a unit lacks it.

Only positions 1-12 are checked, with 0 (id) and 7 (spectral_index) ignored as dimensionless

Return type:: bool

load()[source]

Return type:: SkyModel

skasim.loaders.fits_image

loaders/fits_image.py — Load a SkyModel from a FITS image via OSKAR.

class skasim.loaders.fits_image.FitsImageLoader(fpath, fallback_freq_mhz=None)[source]

Bases: object

Load a SkyModel from a FITS image using oskar.Sky.

The protocol is:: oskar.Sky.from_fits_file(path) → to_array() → np.ndarray → SkyModel

This loader encapsulates the entire OSKAR dependency, allowing lazy import via require_oskar_module so that the module can be imported even when OSKAR is not installed.

load()[source]

Convert the FITS image to a Karabo SkyModel.

Return type:: SkyModel

skasim.loaders.image_models

Public façade for image-model loading, validation and injection.

class skasim.loaders.image_models.CasaModelProduct(model_paths, nterms, reffreq, intermediates, cube_data=None, header=None, freq_axis=None, model_dir=None)[source]

Bases: object

CASA-ready model product generated from one model entry.

cube_data: ndarray | None = None

freq_axis: int | None = None

header: Header | None = None

intermediates: list[Path]

model_dir: Path | None = None

model_paths: list[Path]

nterms: int

reffreq: str

class skasim.loaders.image_models.FitsCubeInfo(path, shape, spatial_shape, unit, n_channels, channel_width_hz, start_frequency_hz, reference_frequency_hz)[source]

Bases: object

Small summary of an accepted 3D spectral-cube model.

channel_width_hz: float

n_channels: int

path: Path

reference_frequency_hz: float

shape: tuple[int, int, int]

spatial_shape: tuple[int, int]

start_frequency_hz: float

unit: str

class skasim.loaders.image_models.FitsImageInfo(path, spatial_shape, unit, celestial_header, center)[source]

Bases: object

Small summary of one accepted FITS image model plane.

celestial_header: dict[str, object]

center: SkyCoord | None

path: Path

spatial_shape: tuple[int, int]

unit: str | None

skasim.loaders.image_models.adjust_spectral_reference(image_path, old_ref_hz, new_ref_hz, alpha_map=None)[source]

Adjust the spectral reference of a CASA image to the observation band centre.

For nterms=1 (alpha_map is None), set CRVAL4 to new_ref_hz only — the model is spectrally flat and no pixel-data correction is needed.

For nterms≥2 (alpha_map provided), correct the pixel data element-wise following CASA’s Taylor-series convention:

tt0'(x,y) = tt0(x,y) · (ν_new / ν_old) ^ α(x,y)

where α(x,y) = tt1(x,y) / tt0(x,y) for each pixel. CRVAL4 is also set.

Returns the adjusted reference frequency in Hz (always new_ref_hz).

Return type:: float

skasim.loaders.image_models.component_model_entries(config)[source]

Return type:: list[ComponentSkyModelEntry]

skasim.loaders.image_models.has_spectral_cube_model(config)[source]

Return True if the simulation contains a spectral_cube model entry.

Return type:: bool

skasim.loaders.image_models.image_model_center(entries)[source]

Return the centre of the first image model with usable celestial WCS.

Return type:: SkyCoord | None

skasim.loaders.image_models.image_model_entries(config)[source]

Return type:: list[Union[ComponentSkyModelEntry, ContinuumIAlphaModelEntry, CasaTaylorTermsModelEntry, StaticStokesMapsModelEntry, SpectralCubeModelEntry]]

skasim.loaders.image_models.inject_image_models(ctx, visibility_path, imaging_configs=None)[source]

Inject configured image models into an existing Measurement Set.

Return type:: None

skasim.loaders.image_models.merge_model_data_into_data(visibility_path)[source]

Add image-model MODEL_DATA into the delivered DATA column.

Return type:: None

skasim.loaders.image_models.prepare_casa_taylor_terms(ctx, entry, index)[source]

Copy CASA Taylor-term images into the run and align their spectral reference.

The reference frequency is adjusted to the observation band centre. For nterms≥2, tt0 pixel data is scaled: tt0’ = tt0 · (ν_obs / ν_old)^α where α = mean(tt1) / mean(tt0). tt1 pixel data is unchanged. For nterms=1, only CRVAL4 is updated.

Return type:: CasaModelProduct

skasim.loaders.image_models.prepare_continuum_i_alpha_for_casa(ctx, entry, index)[source]

Create CASA image products for a continuum I+alpha model.

Adjusts the spectral reference to the observation band centre using the explicit spectral index from the model entry.

Return type:: CasaModelProduct

skasim.loaders.image_models.prepare_spectral_cube_for_casa(ctx, entry, index, report)[source]

Resample a 3D FITS spectral cube to the MS spectral grid.

Returns the resampled cube array in numpy order (freq, dec, ra) together with a 3D FITS header that describes the per-channel model images. The caller is responsible for splitting this into WSClean per-channel model images and running wsclean -predict.

Return type:: CasaModelProduct

skasim.loaders.image_models.primary_model_fits_path(entry)[source]

Return the representative FITS image for previews and phase-centre inference.

Return type:: Path | None

skasim.loaders.image_models.read_fits_cube_info(path)[source]

Read metadata from a 3D FITS spectral cube.

Return type:: FitsCubeInfo

skasim.loaders.image_models.read_fits_image_info(path)[source]

Read FITS image metadata used by validation and reporting.

Return type:: FitsImageInfo

skasim.loaders.image_models.require_casa_executable()[source]

Return a CASA executable for batch-mode fallback.

Return type:: Path

skasim.loaders.image_models.run_casa_exportfits(work_dir, imagename, fitsimage)[source]

Run CASA exportfits in batch mode for a CASA image table.

Return type:: None

skasim.loaders.image_models.run_casa_ft(visibility_path, model_paths, nterms, reffreq, incremental)[source]

Run CASA ft into MODEL_DATA for one prepared model entry.

Return type:: None

skasim.loaders.image_models.run_casa_importfits(work_dir, images)[source]

Run CASA importfits in batch mode for prepared FITS images.

Return type:: None

skasim.loaders.image_models.run_casa_set_spectral_coordinate(work_dir, image_paths, frequency_hz)[source]

Set the single-channel spectral coordinate of CASA images to the run reference.

Return type:: None

skasim.loaders.image_models.run_moment8_for_spectral_cube(ctx, work_dir, output_prefix, tag)[source]

Generate a moment-8 (peak intensity) map and an average spectrum plot from the stacked WSClean clean cube.

Uses pure NumPy over the FITS cube; no CASA required.

Return type:: None

skasim.loaders.image_models.spectral_cube_model_entries(config)[source]

Return all spectral_cube model entries in the config.

Return type:: list

skasim.loaders.image_models.validate_casa_taylor_terms(entry)[source]

Validate an existing CASA Taylor-term image model entry.

Return type:: dict

skasim.loaders.image_models.validate_continuum_i_alpha(entry)[source]

Validate the continuum image contract and return report metadata.

Return type:: dict

skasim.loaders.image_models.validate_spectral_cube(entry, obs, img)[source]

Validate the spectral-cube contract against observation/imaging config.

The cube’s spatial dimensions are checked against the user’s explicit pixels setting when present. If pixels was not supplied (e.g. only fov_deg was given and skasim derives the image size automatically), the cube dimensions are accepted as the authoritative model geometry and a warning is recorded so the run log remains traceable.

Return type:: dict

skasim.loaders.image_models.validate_static_stokes_maps(entry, obs, img_config)[source]

Validate a static Stokes I map and return report metadata.

Currently only stokes_i is supported. The image must be a 2D spatial FITS file with Jy/pixel-compatible BUNIT.

Return type:: dict

skasim.loaders.image_models.write_image_model_previews(ctx, fov, center=None)[source]

Write two-panel FITS model previews for the weblog sky-model section.

Each continuum/Stokes-map/CASA-Taylor-term entry produces a side-by-side PNG: LEFT shows the FITS at its natural image extent so small models are legible; RIGHT shows the simulation FoV contextual view with the primary beam as a circle and the image-model footprint as a rectangle. Spectral cubes keep the existing single-panel moment-8 preview.

Return type:: None

Parameters

ctx: Run context with the model configuration and manifest.
fov: Simulation field of view used for the right-hand contextual panel.
center: Simulation phase centre. When None, the contextual panel falls back to the FITS model’s own centre if it has a usable WCS.

skasim.loaders.image_models.write_spectral_cube_input_preview(ctx, fov)[source]

Render a peak-intensity (moment-8) PNG preview of the input spectral cube.

The preview is added to the manifest as a sky-model plot so it appears in the weblog’s Sky Model section.

Return type:: None