Configuration API

Hierarchical configuration with pydantic v2 models.

Provides a single-source-of-truth PcleanConfig that can be built from:

  • Direct Python construction (PcleanConfig(image=ImageConfig(...))).

  • A YAML file (PcleanConfig.from_yaml('config.yaml')).

  • Layered composition (PcleanConfig.merge(base, overlay)).

  • The flat pclean() kwargs for backward compatibility (PcleanConfig.from_flat_kwargs(...)).

The resulting config can be converted to CASA-native dicts via the to_casa_*() bridge methods, or to the legacy PcleanParams via PcleanConfig.to_params() (deprecated).

class pclean.config.SelectionConfig(*, vis='', field='', spw='', timerange='', uvrange='', antenna='', scan='', observation='', intent='', datacolumn='corrected')[source]

Bases: BaseModel

Data selection parameters.

Parameters:
vis: str | list[str]
field: str | list[str]
spw: str | list[str]
timerange: str | list[str]
uvrange: str | list[str]
antenna: str | list[str]
scan: str | list[str]
observation: str
intent: str
datacolumn: str
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.ImageConfig(*, imagename='', imsize=<factory>, cell='1arcsec', phasecenter='', stokes='I', projection='SIN', startmodel='', specmode='mfs', reffreq='', nchan=-1, start='', width='', outframe='LSRK', veltype='radio', restfreq=<factory>, interpolation='linear', perchanweightdensity=True, nterms=2)[source]

Bases: BaseModel

Image definition parameters.

Parameters:
imagename: str
imsize: list[int]
cell: list[str] | str
phasecenter: str
stokes: str
projection: str
startmodel: str
specmode: str
reffreq: str
nchan: int
start: str | int
width: str | int
outframe: str
veltype: str
restfreq: list[str]
interpolation: str
perchanweightdensity: bool
nterms: int
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.GridConfig(*, gridder='standard', facets=1, wprojplanes=1, vptable='', mosweight=True, aterm=True, psterm=False, wbawp=True, conjbeams=False, cfcache='', usepointing=False, computepastep=360.0, rotatepastep=360.0, pointingoffsetsigdev=<factory>, pblimit=0.2, normtype='flatnoise', psfphasecenter='')[source]

Bases: BaseModel

Gridding parameters.

Parameters:
gridder: str
facets: int
wprojplanes: int
vptable: str
mosweight: bool
aterm: bool
psterm: bool
wbawp: bool
conjbeams: bool
cfcache: str
usepointing: bool
computepastep: float
rotatepastep: float
pointingoffsetsigdev: list[float]
pblimit: float
normtype: str
psfphasecenter: str
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.WeightConfig(*, weighting='natural', robust=0.5, noise='1.0Jy', npixels=0, uvtaper=<factory>, fracbw=None)[source]

Bases: BaseModel

Weighting parameters.

Parameters:
weighting: str
robust: float
noise: str
npixels: int
uvtaper: list[str]
fracbw: float | None
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.DeconvolutionConfig(*, deconvolver='hogbom', scales=<factory>, nterms=2, smallscalebias=0.0, fusedthreshold=0.0, largestscale=-1, restoration=True, restoringbeam=<factory>, pbcor=False, usemask='user', mask='', pbmask=0.0, sidelobethreshold=3.0, noisethreshold=5.0, lownoisethreshold=1.5, negativethreshold=0.0, smoothfactor=1.0, minbeamfrac=0.3, cutthreshold=0.01, growiterations=100, dogrowprune=True, minpercentchange=0.0, verbose=False, fastnoise=True, python_automask=True)[source]

Bases: BaseModel

Deconvolution and masking parameters.

Parameters:
deconvolver: str
scales: list[int]
nterms: int
smallscalebias: float
fusedthreshold: float
largestscale: int
restoration: bool
restoringbeam: list[str] | str
pbcor: bool
usemask: str
mask: str
pbmask: float
sidelobethreshold: float
noisethreshold: float
lownoisethreshold: float
negativethreshold: float
smoothfactor: float
minbeamfrac: float
cutthreshold: float
growiterations: int
dogrowprune: bool
minpercentchange: float
verbose: bool
fastnoise: bool
python_automask: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.IterationConfig(*, niter=0, gain=0.1, threshold='0.0mJy', nsigma=0.0, cycleniter=-1, cyclefactor=1.0, minpsffraction=0.05, maxpsffraction=0.8, interactive=False, nmajor=-1, fullsummary=False)[source]

Bases: BaseModel

Iteration control parameters.

Parameters:
niter: int
gain: float
threshold: str
nsigma: float
cycleniter: int
cyclefactor: float
minpsffraction: float
maxpsffraction: float
interactive: bool
nmajor: int
fullsummary: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.MiscConfig(*, restart=True, savemodel='none', calcres=True, calcpsf=True, psfcutoff=0.35)[source]

Bases: BaseModel

Miscellaneous parameters.

Parameters:
restart: bool
savemodel: str
calcres: bool
calcpsf: bool
psfcutoff: float
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.NormConfig(*, pblimit=0.2, normtype='flatnoise', psfcutoff=0.35)[source]

Bases: BaseModel

Normalization parameters.

Parameters:
pblimit: float
normtype: str
psfcutoff: float
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.SlurmConfig(*, queue=None, account=None, walltime='24:00:00', job_mem='20GB', cores_per_job=1, job_name=None, job_extra_directives=<factory>, python=None, local_directory=None, log_directory='logs', job_script_prologue=<factory>)[source]

Bases: BaseModel

SLURM batch-job parameters (used when cluster.type == 'slurm').

Parameters:

  • queue (str | None)

  • account (str | None)

  • walltime (str)

  • job_mem (str)

  • cores_per_job (int)

  • job_name (str | None)

  • job_extra_directives (list[str])

  • python (str | None)

  • local_directory (str | None)

  • log_directory (str)

  • job_script_prologue (list[str])

queue: str | None
account: str | None
walltime: str
job_mem: str
cores_per_job: int
job_name: str | None
job_extra_directives: list[str]
python: str | None
local_directory: str | None
log_directory: str
job_script_prologue: list[str]
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.SubmitConfig(*, workdir=None, pixi_project_dir=None, pixi_env='forge', coordinator_mem='8G', coordinator_cpus=2, coordinator_walltime='24:00:00', coordinator_job_name='pclean-coordinator', extra_sbatch=<factory>, log_dir=None, psrecord=True)[source]

Bases: BaseModel

SLURM coordinator (submit) job parameters.

These control the coordinator SLURM job that runs python -m pclean --pconfig <yaml> and lets dask-jobqueue spawn per-channel worker jobs.

Parameters:

  • workdir (str | None)

  • pixi_project_dir (str | None)

  • pixi_env (str)

  • coordinator_mem (str)

  • coordinator_cpus (int)

  • coordinator_walltime (str)

  • coordinator_job_name (str)

  • extra_sbatch (list[str])

  • log_dir (str | None)

  • psrecord (bool)

workdir: str | None
pixi_project_dir: str | None
pixi_env: str
coordinator_mem: str
coordinator_cpus: int
coordinator_walltime: str
coordinator_job_name: str
extra_sbatch: list[str]
log_dir: str | None
psrecord: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.ClusterConfig(*, type='local', nworkers=None, scheduler_address=None, threads_per_worker=1, memory_limit='0', local_directory=None, parallel=False, cube_chunksize=-1, keep_subcubes=False, keep_partimages=False, concat_mode='auto', slurm=<factory>, submit=<factory>)[source]

Bases: BaseModel

Dask cluster configuration.

Parameters:

  • type (Literal['local', 'slurm', 'address'])

  • nworkers (int | None)

  • scheduler_address (str | None)

  • threads_per_worker (int)

  • memory_limit (str)

  • local_directory (str | None)

  • parallel (bool)

  • cube_chunksize (int)

  • keep_subcubes (bool)

  • keep_partimages (bool)

  • concat_mode (Literal['auto', 'paged', 'virtual', 'movevirtual'])

  • slurm (SlurmConfig)

  • submit (SubmitConfig)

type: Literal['local', 'slurm', 'address']
nworkers: int | None
scheduler_address: str | None
threads_per_worker: int
memory_limit: str
local_directory: str | None
parallel: bool
cube_chunksize: int
keep_subcubes: bool
keep_partimages: bool
concat_mode: Literal['auto', 'paged', 'virtual', 'movevirtual']
slurm: SlurmConfig
submit: SubmitConfig
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pclean.config.PcleanConfig(*, selection=<factory>, image=<factory>, grid=<factory>, weight=<factory>, deconvolution=<factory>, iteration=<factory>, normalization=<factory>, misc=<factory>, cluster=<factory>)[source]

Bases: BaseModel

Top-level hierarchical configuration for pclean.

All parameters are grouped into logical sub-configs. This is the single source of truth for the application.

Parameters:
selection: SelectionConfig
image: ImageConfig
grid: GridConfig
weight: WeightConfig
deconvolution: DeconvolutionConfig
iteration: IterationConfig
normalization: NormConfig
misc: MiscConfig
cluster: ClusterConfig
classmethod from_yaml(path)[source]

Load a config from a YAML file.

Parameters:

path (str | Path) – Path to the YAML file.

Return type:

PcleanConfig

to_yaml(path)[source]

Dump the config to a YAML file.

Parameters:

path (str | Path) – Destination file path.

Return type:

None

classmethod merge(*configs)[source]

Deep-merge multiple configs, later values override earlier ones.

Parameters:

configs (PcleanConfig) – Two or more configs to merge left-to-right.

Return type:

PcleanConfig

classmethod from_flat_kwargs(vis='', **kwargs)[source]

Build a PcleanConfig from the flat pclean() keyword arguments.

This is the backward-compatibility shim that maps the 80+ flat keyword arguments into the hierarchical structure.

Parameters:

  • vis (str | list[str]) – Measurement set path(s).

  • **kwargs (Any) – Flat keyword arguments matching the pclean() signature.

Return type:

PcleanConfig

to_flat_kwargs()[source]

Convert back to the flat keyword dict accepted by pclean().

Returns:

Flat dictionary with all parameters.

Return type:

dict[str, Any]

to_casa_selpars()[source]

Build CASA selectdata-compatible multi-MS selection dicts.

Returns:

Dict keyed 'ms0', 'ms1', etc. with CASA-internal field names.

Return type:

dict[str, dict]

to_casa_impars()[source]

Build CASA defineimage-compatible image parameter dicts.

Returns:

Dict keyed '0' (single field) with image parameters.

Return type:

dict[str, dict]

to_casa_gridpars()[source]

Build CASA defineimage grid parameter dicts.

Return type:

dict[str, dict]

to_casa_weightpars()[source]

Build CASA setweighting-compatible weight parameter dict.

Return type:

dict

to_casa_decpars()[source]

Build CASA setupdeconvolution-compatible deconvolution dicts.

Return type:

dict[str, dict]

to_casa_normpars()[source]

Build CASA setupnormalizer-compatible normalization dicts.

Return type:

dict[str, dict]

to_casa_iterpars()[source]

Build CASA setupiteration-compatible iteration parameter dict.

Return type:

dict

to_casa_miscpars()[source]

Build miscellaneous parameter dict.

Return type:

dict

to_casa_bundle()[source]

Produce a serializable dict of all CASA-native parameter dicts.

This is the worker-facing payload for continuum-parallel actors. Cube-parallel workers receive a serialized PcleanConfig instead.

Returns:

Dict with keys matching the former PcleanParams.to_dict() format.

Return type:

dict

property specmode: str

Spectral mode ('mfs', 'cube', etc.).

property imagename: str

Output image name prefix.

property parallel: bool

Whether Dask parallelism is enabled.

property niter: int

Maximum number of CLEAN iterations.

property is_cube: bool

True if specmode indicates cube imaging.

property is_mfs: bool

True if specmode is multi-frequency synthesis.

property nfields: int

Number of image fields (currently always 1).

property nms: int

Number of measurement sets.

make_subcube_config(start, nchan, image_suffix, width=None)[source]

Return a copy tuned for a channel sub-range (cube parallelism).

Parameters:

  • start (int | str) – Start channel (int) or frequency/velocity string.

  • nchan (int) – Number of channels in this subcube.

  • image_suffix (str) – Suffix appended to the base imagename.

  • width (int | str | None) – Channel width override. When start is converted to a frequency string the caller must also supply a matching frequency width so that CASA does not reject mixed unit types.

Return type:

PcleanConfig

to_params()[source]

Convert to the legacy PcleanParams used by engines.

Deprecated since version Use: the to_casa_*() methods or pass PcleanConfig directly to engines instead.

Returns:

A fully constructed PcleanParams instance.

Return type:

Any

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pclean.config.load_defaults()[source]

Load the bundled defaults.yaml reference snapshot.

This is equivalent to PcleanConfig() (all pydantic defaults) but read from the packaged YAML file for verification purposes.

Returns:

A PcleanConfig with the reference default values.

Return type:

PcleanConfig

pclean.config.load_preset(name)[source]

Load a named preset from the bundled configs/presets/ directory.

Searches first inside the installed package, then falls back to CWD-relative paths.

Parameters:

name (str) – Preset name (without .yaml extension).

Return type:

PcleanConfig

pclean.config.get_adios2_config_path()[source]

Return the filesystem path of the bundled ADIOS2 XML config.

The file enables lossless blosc2/zstd compression and sets sensible buffer sizes for CASA Measurement Sets stored in ADIOS2 format. Point ADIOS2 at it via the environment variable before opening any MS:

import os
from pclean.config import get_adios2_config_path

path = get_adios2_config_path()
if path is not None:
    os.environ.setdefault('ADIOS2_INIT_FILE', str(path))
Returns:

Resolved filesystem Path, or None if not found.

Return type:

Path | None