Configuration#

Pydantic-based configuration schemas. Three layers: user-facing defaults (ParamConfig), user overrides (UserConfig), and the fully-resolved immutable runtime config (InternalConfig).

ParamConfig: Expert defaults for Adapt pipeline.

This module defines the complete, scientifically-validated default configuration. ALL pipeline parameters must have defaults here. No runtime code should define fallback values - this is the single source of truth for defaults.

Runtime code NEVER reads from ParamConfig directly - it only receives InternalConfig.

class adapt.configuration.schemas.param.AnalyzerConfig(*, radar_variables=<factory>, exclude_fields=<factory>, adjacency_min_touching_boundary_pixels=1)#

Bases: AdaptBaseModel

Cell analysis configuration.

Parameters:

  • radar_variables (list[str])

  • exclude_fields (list[str])

  • adjacency_min_touching_boundary_pixels (int)

adjacency_min_touching_boundary_pixels: int#
exclude_fields: list[str]#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

radar_variables: list[str]#
class adapt.configuration.schemas.param.CoordNamesConfig(*, time='time', z='z', y='y', x='x')#

Bases: AdaptBaseModel

Coordinate name mappings.

Parameters:

  • time (str)

  • z (str)

  • y (str)

  • x (str)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

time: str#
x: str#
y: str#
z: str#
class adapt.configuration.schemas.param.DownloaderConfig(*, radar=None, output_dir=None, latest_files=5, latest_minutes=60, poll_interval_sec=300, max_fetch_retries=3, start_time=None, end_time=None, min_file_size=1024, max_queue_size=100, queue_resume_fraction=0.1)#

Bases: AdaptBaseModel

NEXRAD data downloader configuration.

Parameters:

  • radar (str | None)

  • output_dir (str | None)

  • latest_files (int)

  • latest_minutes (int)

  • poll_interval_sec (int)

  • max_fetch_retries (int)

  • start_time (str | None)

  • end_time (str | None)

  • min_file_size (int)

  • max_queue_size (int)

  • queue_resume_fraction (float)

end_time: str | None#
latest_files: int#
latest_minutes: int#
max_fetch_retries: int#
max_queue_size: int#
min_file_size: int#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

output_dir: str | None#
poll_interval_sec: int#
queue_resume_fraction: float#
radar: str | None#
start_time: str | None#
class adapt.configuration.schemas.param.FlowParamsConfig(*, pyr_scale=0.5, levels=3, winsize=10, iterations=3, poly_n=7, poly_sigma=1.5, flags=0)#

Bases: AdaptBaseModel

OpenCV optical flow parameters.

Parameters:

  • pyr_scale (float)

  • levels (int)

  • winsize (int)

  • iterations (int)

  • poly_n (int)

  • poly_sigma (float)

  • flags (int)

flags: int#
iterations: int#
levels: int#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

poly_n: int#
poly_sigma: float#
pyr_scale: float#
winsize: int#
class adapt.configuration.schemas.param.GlobalConfig(*, z_level=2000.0, max_scan_gap_minutes=10.0, tracking_field='reflectivity', coord_names=<factory>)#

Bases: AdaptBaseModel

Global pipeline settings.

Parameters:

  • z_level (float)

  • max_scan_gap_minutes (float)

  • tracking_field (str)

  • coord_names (CoordNamesConfig)

classmethod coerce_z_level_to_float(v)#

Allow int or float for z_level.

coord_names: CoordNamesConfig#
max_scan_gap_minutes: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

tracking_field: str#
z_level: float#
class adapt.configuration.schemas.param.LoggingConfig(*, level='INFO')#

Bases: AdaptBaseModel

Logging configuration.

Parameters:

level (Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])

level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

class adapt.configuration.schemas.param.OutputConfig(*, compression='snappy')#

Bases: AdaptBaseModel

Output file configuration.

Parameters:

compression (Literal['snappy', 'gzip', 'lz4', 'none'])

compression: Literal['snappy', 'gzip', 'lz4', 'none']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

class adapt.configuration.schemas.param.ParamConfig(*, mode='realtime', source='aws_nexrad', source_dir=None, reader=<factory>, downloader=<factory>, regridder=<factory>, segmenter=<factory>, global_=<factory>, projector=<factory>, analyzer=<factory>, tracker=<factory>, visualization=<factory>, output=<factory>, logging=<factory>)#

Bases: AdaptBaseModel

Complete expert configuration with all defaults.

This is the single source of truth for all pipeline parameters. Every tunable parameter MUST have a default here.

Usage#

This config is NOT used directly by runtime code. It serves as the base layer in config resolution:

internal_cfg = resolve_config(param_cfg, user_cfg, cli_cfg)

Runtime code only sees InternalConfig.

analyzer: AnalyzerConfig#
downloader: DownloaderConfig#
global_: GlobalConfig#
logging: LoggingConfig#
mode: Literal['realtime', 'historical']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

output: OutputConfig#
projector: ProjectorConfig#
reader: ReaderConfig#
regridder: RegridderConfig#
segmenter: SegmenterConfig#
source: str#
source_dir: str | None#
tracker: TrackerConfig#
visualization: VisualizationConfig#
Parameters:
class adapt.configuration.schemas.param.ProjectorConfig(*, method='adapt_default', max_projection_steps=3, nan_fill_value=0.0, flow_params=<factory>, min_motion_threshold=1.0, max_flow_magnitude=20.0, registration_step_minutes=1, projection_horizon_minutes=15)#

Bases: AdaptBaseModel

Cell projection configuration.

Parameters:

  • method (Literal['adapt_default'])

  • max_projection_steps (int)

  • nan_fill_value (float)

  • flow_params (FlowParamsConfig)

  • min_motion_threshold (float)

  • max_flow_magnitude (float)

  • registration_step_minutes (int)

  • projection_horizon_minutes (int)

flow_params: FlowParamsConfig#
max_flow_magnitude: float#
max_projection_steps: int#
method: Literal['adapt_default']#
min_motion_threshold: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

nan_fill_value: float#
classmethod normalize_method_name(v)#

Normalize method names to lowercase.

projection_horizon_minutes: int#
registration_step_minutes: int#
class adapt.configuration.schemas.param.ReaderConfig(*, file_format='nexrad_archive', field_map=<factory>, fields=<factory>)#

Bases: AdaptBaseModel

Radar file reader configuration.

Parameters:

  • file_format (Literal['nexrad_archive'])

  • field_map (dict[str, str])

  • fields (list[str])

field_map: dict[str, str]#
fields: list[str]#
file_format: Literal['nexrad_archive']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

class adapt.configuration.schemas.param.RegridderConfig(*, grid_shape=(41, 301, 301), grid_limits=((0.0, 20000.0), (-150000.0, 150000.0), (-150000.0, 150000.0)), roi_func='dist_beam', min_radius=1750.0, weighting_function='cressman')#

Bases: AdaptBaseModel

PyART regridding configuration.

Parameters:

  • grid_shape (tuple[int, int, int])

  • grid_limits (tuple[tuple[float, float], tuple[float, float], tuple[float, float]])

  • roi_func (Literal['dist_beam', 'dist'])

  • min_radius (float)

  • weighting_function (Literal['cressman', 'barnes', 'nearest'])

grid_limits: tuple[tuple[float, float], tuple[float, float], tuple[float, float]]#
grid_shape: tuple[int, int, int]#
min_radius: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

roi_func: Literal['dist_beam', 'dist']#
weighting_function: Literal['cressman', 'barnes', 'nearest']#
class adapt.configuration.schemas.param.SegmenterConfig(*, method='threshold', min_cellsize_gridpoint=5, max_cellsize_gridpoint=None, closing_radius=2, filter_by_size=True, h_maxima=5.0, seed_carry_frames=1, carried_exempt_size_filter=False, seed_carry_min_separation=5, threshold_params=<factory>, conv_strat_raut_params=<factory>, conv_strat_yuter_params=<factory>, feature_detection_params=<factory>, steiner_conv_strat_params=<factory>)#

Bases: AdaptBaseModel

Cell segmentation configuration.

The user selects a method; the parameters for that method come from the matching nested block below (Adapt-owned defaults). The shared labeling parameters apply to every method’s convective mask.

Parameters:

  • method (Literal['conv_strat_raut', 'conv_strat_yuter', 'feature_detection', 'steiner_conv_strat', 'threshold'])

  • min_cellsize_gridpoint (int)

  • max_cellsize_gridpoint (int | None)

  • closing_radius (int)

  • filter_by_size (bool)

  • h_maxima (float)

  • seed_carry_frames (int)

  • carried_exempt_size_filter (bool)

  • seed_carry_min_separation (int)

  • threshold_params (ThresholdParams)

  • conv_strat_raut_params (ConvStratRautParams)

  • conv_strat_yuter_params (ConvStratYuterParams)

  • feature_detection_params (FeatureDetectionParams)

  • steiner_conv_strat_params (SteinerConvStratParams)

carried_exempt_size_filter: bool#
closing_radius: int#
conv_strat_raut_params: ConvStratRautParams#
conv_strat_yuter_params: ConvStratYuterParams#
feature_detection_params: FeatureDetectionParams#
filter_by_size: bool#
h_maxima: float#
max_cellsize_gridpoint: int | None#
method: Literal['conv_strat_raut', 'conv_strat_yuter', 'feature_detection', 'steiner_conv_strat', 'threshold']#
min_cellsize_gridpoint: int#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

seed_carry_frames: int#
seed_carry_min_separation: int#
steiner_conv_strat_params: SteinerConvStratParams#
threshold_params: ThresholdParams#
class adapt.configuration.schemas.param.TrackerConfig(*, cell_threshold=30.0, polarity=1, growth_max_km=2.0, growth_size_km=12.0, max_overlap_mismatch=0.6, max_link_cost=1.8, residual_weight=0.25, heading_weight=1.0, max_speed_ms=40.0, max_acceleration_ms2=0.04, split_overlap_threshold=0.65, merge_overlap_threshold=0.7, latent_scans=2, identity_intensity_weight=1.0, identity_score_margin=0.33, core_field_threshold=40.0, cell_uid=<factory>)#

Bases: AdaptBaseModel

Cell tracking configuration.

Parameters:

  • cell_threshold (float)

  • polarity (Literal[1, -1])

  • growth_max_km (float)

  • growth_size_km (float)

  • max_overlap_mismatch (float)

  • max_link_cost (float)

  • residual_weight (float)

  • heading_weight (float)

  • max_speed_ms (float)

  • max_acceleration_ms2 (float)

  • split_overlap_threshold (float)

  • merge_overlap_threshold (float)

  • latent_scans (int)

  • identity_intensity_weight (float)

  • identity_score_margin (float)

  • core_field_threshold (float)

  • cell_uid (CellUidConfig)

class CellUidConfig(*, width=10, alphabet='base36_upper')#

Bases: AdaptBaseModel

Track ID generation configuration.

v2 uids hash (scan_id, cell_label) — deterministic and collision-free by construction; only the token width/alphabet remain configurable.

Parameters:

  • width (Annotated[int, Ge(ge=1)])

  • alphabet (Literal['base36_upper'])

alphabet: Literal['base36_upper']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

width: int#
cell_threshold: float#
cell_uid: CellUidConfig#
core_field_threshold: float#
growth_max_km: float#
growth_size_km: float#
heading_weight: float#
identity_intensity_weight: float#
identity_score_margin: float#
latent_scans: int#
max_acceleration_ms2: float#
max_overlap_mismatch: float#
max_speed_ms: float#
merge_overlap_threshold: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

polarity: Literal[1, -1]#
residual_weight: float#
split_overlap_threshold: float#
class adapt.configuration.schemas.param.VisualizationConfig(*, enabled=True, dpi=200, figsize=(18.0, 8.0), output_format='png', use_basemap=True, basemap_alpha=0.6, seg_linewidth=0.8, proj_linewidth=1.0, proj_alpha=0.8, flow_scale=0.5, flow_subsample=10, min_reflectivity=10.0, refl_vmin=10.0, refl_vmax=50.0)#

Bases: AdaptBaseModel

Visualization settings.

Parameters:

  • enabled (bool)

  • dpi (int)

  • figsize (tuple[float, float])

  • output_format (Literal['png', 'pdf', 'jpeg'])

  • use_basemap (bool)

  • basemap_alpha (float)

  • seg_linewidth (float)

  • proj_linewidth (float)

  • proj_alpha (float)

  • flow_scale (float)

  • flow_subsample (int)

  • min_reflectivity (float)

  • refl_vmin (float)

  • refl_vmax (float)

basemap_alpha: float#
dpi: int#
enabled: bool#
figsize: tuple[float, float]#
flow_scale: float#
flow_subsample: int#
min_reflectivity: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

output_format: Literal['png', 'pdf', 'jpeg']#
proj_alpha: float#
proj_linewidth: float#
refl_vmax: float#
refl_vmin: float#
seg_linewidth: float#
use_basemap: bool#

UserConfig: Forgiving, minimal user-facing configuration.

This schema accepts user inputs in a variety of formats, with aliases for common naming patterns (e.g., RADAR_ID → radar, MODE → mode).

UserConfig is intentionally minimal - users only specify what they want to override from the expert defaults. Validation is lenient to accept both uppercase and lowercase keys, integers where floats are expected, etc.

class adapt.configuration.schemas.user.UserAnalyzerConfig(*, radar_variables=None, exclude_fields=None, **extra_data)#

Bases: _UserSection

User-facing analyzer config.

Parameters:

  • radar_variables (list[str] | None)

  • exclude_fields (list[str] | None)

  • extra_data (Any)

exclude_fields: list[str] | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

radar_variables: list[str] | None#
class adapt.configuration.schemas.user.UserConfig(*, MODE=None, source=None, source_dir=None, RADAR_ID=None, BASE_DIR=None, LATEST_FILES=None, LATEST_MINUTES=None, POLL_INTERVAL_SEC=None, MAX_FETCH_RETRIES=None, START_TIME=None, END_TIME=None, GRID_SHAPE=None, GRID_LIMITS=None, Z_LEVEL=None, REFLECTIVITY_VAR=None, SEGMENTATION_METHOD=None, MIN_CELLSIZE_GRIDPOINT=None, MAX_CELLSIZE_GRIDPOINT=None, PROJECTION_METHOD=None, MAX_PROJECTION_STEPS=None, radar_variables=None, exclude_fields=None, extensions=<factory>, modules=None, module_params=None, downloader=None, regridder=None, segmenter=None, global_=None, projector=None, analyzer=None, **extra_data)#

Bases: AdaptBaseModel

User-facing configuration schema.

Minimal, forgiving, and uses common aliases. Users only specify what they want to override from ParamConfig defaults.

This config is converted to internal overrides during resolution.

Usage#

user_cfg = UserConfig(

mode=”historical”, radar=”KHTX”, base_dir=”/data/adapt”, z_level=2000, threshold=35,

)

internal = resolve_config(param_cfg, user_cfg, cli_cfg)

analyzer: UserAnalyzerConfig | None#
base_dir: str | None#
classmethod coerce_numeric_fields(v)#

Accept int or float for numeric fields.

downloader: UserDownloaderConfig | None#
end_time: str | None#
exclude_fields: list[str] | None#
extensions: list[str]#
global_: UserGlobalConfig | None#
grid_limits: tuple[tuple[float, float], tuple[float, float], tuple[float, float]] | None#
grid_shape: tuple[int, int, int] | None#
infer_historical_mode_from_times()#

If times provided but mode not specified, set mode to historical.

This is a schema responsibility: if user config indicates a time range, the mode should automatically be historical.

latest_files: int | None#
latest_minutes: int | None#
max_cellsize_gridpoint: int | None#
max_fetch_retries: int | None#
max_projection_steps: int | None#
min_cellsize_gridpoint: int | None#
mode: Literal['realtime', 'historical'] | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

module_params: dict[str, dict[str, Any]] | None#
modules: list[str] | None#
classmethod normalize_method_names(v)#

Normalize method names to lowercase.

poll_interval_sec: int | None#
projection_method: str | None#
projector: UserProjectorConfig | None#
radar: str | None#
radar_variables: list[str] | None#
reflectivity_var: str | None#
regridder: UserRegridderConfig | None#
segmentation_method: str | None#
segmenter: UserSegmenterConfig | None#
source: str | None#
source_dir: str | None#
start_time: str | None#
to_internal_overrides()#

Convert flat UserConfig to nested InternalConfig structure.

Returns:

Nested dictionary matching InternalConfig structure

Return type:

dict

z_level: float | None#
Parameters:

  • MODE (Literal['realtime', 'historical'] | None)

  • source (str | None)

  • source_dir (str | None)

  • RADAR_ID (str | None)

  • BASE_DIR (str | None)

  • LATEST_FILES (int | None)

  • LATEST_MINUTES (int | None)

  • POLL_INTERVAL_SEC (int | None)

  • MAX_FETCH_RETRIES (int | None)

  • START_TIME (str | None)

  • END_TIME (str | None)

  • GRID_SHAPE (tuple[int, int, int] | None)

  • GRID_LIMITS (tuple[tuple[float, float], tuple[float, float], tuple[float, float]] | None)

  • Z_LEVEL (float | None)

  • REFLECTIVITY_VAR (str | None)

  • SEGMENTATION_METHOD (str | None)

  • MIN_CELLSIZE_GRIDPOINT (int | None)

  • MAX_CELLSIZE_GRIDPOINT (int | None)

  • PROJECTION_METHOD (str | None)

  • MAX_PROJECTION_STEPS (int | None)

  • radar_variables (list[str] | None)

  • exclude_fields (list[str] | None)

  • extensions (list[str])

  • modules (list[str] | None)

  • module_params (dict[str, dict[str, Any]] | None)

  • downloader (UserDownloaderConfig | None)

  • regridder (UserRegridderConfig | None)

  • segmenter (UserSegmenterConfig | None)

  • global_ (UserGlobalConfig | None)

  • projector (UserProjectorConfig | None)

  • analyzer (UserAnalyzerConfig | None)

  • extra_data (Any)

class adapt.configuration.schemas.user.UserDownloaderConfig(*, radar=None, output_dir=None, latest_files=None, latest_minutes=None, poll_interval_sec=None, max_fetch_retries=None, start_time=None, end_time=None, max_queue_size=None, queue_resume_fraction=None, **extra_data)#

Bases: _UserSection

User-facing downloader config.

Parameters:

  • radar (str | None)

  • output_dir (str | None)

  • latest_files (int | None)

  • latest_minutes (int | None)

  • poll_interval_sec (int | None)

  • max_fetch_retries (int | None)

  • start_time (str | None)

  • end_time (str | None)

  • max_queue_size (int | None)

  • queue_resume_fraction (float | None)

  • extra_data (Any)

end_time: str | None#
latest_files: int | None#
latest_minutes: int | None#
max_fetch_retries: int | None#
max_queue_size: int | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

output_dir: str | None#
poll_interval_sec: int | None#
queue_resume_fraction: float | None#
radar: str | None#
start_time: str | None#
class adapt.configuration.schemas.user.UserGlobalConfig(*, z_level=None, max_scan_gap_minutes=None, tracking_field=None, coord_names=None, **extra_data)#

Bases: _UserSection

User-facing global config.

Parameters:

  • z_level (float | None)

  • max_scan_gap_minutes (float | None)

  • tracking_field (str | None)

  • coord_names (dict[str, str] | None)

  • extra_data (Any)

classmethod coerce_z_level(v)#

Accept int or float for z_level.

coord_names: dict[str, str] | None#
max_scan_gap_minutes: float | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

tracking_field: str | None#
z_level: float | None#
class adapt.configuration.schemas.user.UserProjectorConfig(*, method=None, max_projection_steps=None, nan_fill_value=None, flow_params=None, min_motion_threshold=None, max_flow_magnitude=None, registration_step_minutes=None, projection_horizon_minutes=None, **extra_data)#

Bases: _UserSection

User-facing projector config.

Parameters:

  • method (str | None)

  • max_projection_steps (int | None)

  • nan_fill_value (float | None)

  • flow_params (dict[str, Any] | None)

  • min_motion_threshold (float | None)

  • max_flow_magnitude (float | None)

  • registration_step_minutes (int | None)

  • projection_horizon_minutes (int | None)

  • extra_data (Any)

flow_params: dict[str, Any] | None#
max_flow_magnitude: float | None#
max_projection_steps: int | None#
method: str | None#
min_motion_threshold: float | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

nan_fill_value: float | None#
classmethod normalize_method(v)#

Normalize method names to lowercase.

projection_horizon_minutes: int | None#
registration_step_minutes: int | None#
class adapt.configuration.schemas.user.UserRegridderConfig(*, grid_shape=None, grid_limits=None, roi_func=None, min_radius=None, weighting_function=None, **extra_data)#

Bases: _UserSection

User-facing regridder config.

Parameters:

  • grid_shape (tuple[int, int, int] | None)

  • grid_limits (tuple[tuple[float, float], tuple[float, float], tuple[float, float]] | None)

  • roi_func (str | None)

  • min_radius (float | None)

  • weighting_function (str | None)

  • extra_data (Any)

grid_limits: tuple[tuple[float, float], tuple[float, float], tuple[float, float]] | None#
grid_shape: tuple[int, int, int] | None#
min_radius: float | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

roi_func: str | None#
weighting_function: str | None#
class adapt.configuration.schemas.user.UserSegmenterConfig(*, method=None, min_cellsize_gridpoint=None, max_cellsize_gridpoint=None, closing_radius=None, filter_by_size=None, h_maxima=None, seed_carry_frames=None, seed_carry_min_separation=None, carried_exempt_size_filter=None, **extra_data)#

Bases: _UserSection

User-facing segmentation config with aliases.

Per-method scientific parameters are not exposed here yet — the user selects only method, and its parameters resolve from ParamConfig. Advanced users may still override a nested <method>_params block, which passes through via this section’s extra="allow".

Parameters:

  • method (str | None)

  • min_cellsize_gridpoint (int | None)

  • max_cellsize_gridpoint (int | None)

  • closing_radius (int | None)

  • filter_by_size (bool | None)

  • h_maxima (float | None)

  • seed_carry_frames (int | None)

  • seed_carry_min_separation (int | None)

  • carried_exempt_size_filter (bool | None)

  • extra_data (Any)

carried_exempt_size_filter: bool | None#
closing_radius: int | None#
filter_by_size: bool | None#
classmethod fold_legacy_closing_kernel(data)#

Accept the retired closing_kernel: [w, h] rectangle as a disk radius.

(1, 1) was a no-op and maps to 0; a w x h rectangle maps to max(w, h) // 2 so the smoothing scale is preserved.

classmethod fold_legacy_seed_carry(data)#

Accept the retired seed_carry / seed_carry_max_frames pair.

One integer replaced them: 0 is off, N carries for N frames. A legacy config maps as seed_carry: false -> 0 and seed_carry: true -> seed_carry_max_frames (default 2, its old default), so existing runs keep their behaviour.

h_maxima: float | None#
max_cellsize_gridpoint: int | None#
method: str | None#
min_cellsize_gridpoint: int | None#
model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

classmethod normalize_method(v)#

Normalize method names to lowercase.

seed_carry_frames: int | None#
seed_carry_min_separation: int | None#

InternalConfig: Authoritative runtime configuration.

This is the ONLY config schema that runtime code sees. It is fully validated, normalized, and contains NO optional fields that processing code depends on.

All .get() calls, fallback defaults, and validation logic are FORBIDDEN in runtime code - everything is explicit here.

class adapt.configuration.schemas.internal.InternalAnalyzerConfig(*, radar_variables, exclude_fields, adjacency_min_touching_boundary_pixels)#

Bases: AdaptBaseModel

Runtime analysis configuration.

Parameters:

  • radar_variables (list[str])

  • exclude_fields (list[str])

  • adjacency_min_touching_boundary_pixels (Annotated[int, Ge(ge=1)])

adjacency_min_touching_boundary_pixels: int#
exclude_fields: list[str]#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

radar_variables: list[str]#
class adapt.configuration.schemas.internal.InternalConfig(*, mode, base_dir, source='aws_nexrad', source_dir=None, run_id=None, output_dirs=None, reader, downloader, regridder, segmenter, global_, projector, analyzer, tracker, visualization, output, logging, processor=<factory>, extensions=<factory>, module_params=<factory>, modules=None, only_modules=None, exclude_modules=None)#

Bases: AdaptBaseModel

Authoritative runtime configuration.

This is the ONLY configuration schema that processing code sees. It is fully validated, immutable, and contains explicit values for all parameters (no None for fields that runtime depends on).

Usage#

Runtime modules receive InternalConfig and access fields directly:

def __init__(self, config: InternalConfig):

self.threshold = config.segmenter.threshold # NOT .get() self.z_level = config.global_.z_level

Rules#

  • NO .get() calls

  • NO fallback defaults

  • NO type checking

  • NO validation

All of that happens during config resolution, not in runtime code.

analyzer: InternalAnalyzerConfig#
base_dir: str#
downloader: InternalDownloaderConfig#
exclude_modules: list[str] | None#
extensions: list[str]#
global_: InternalGlobalConfig#
logging: InternalLoggingConfig#
mode: Literal['realtime', 'historical']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'populate_by_name': True, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

module_params: dict[str, dict[str, Any]]#
modules: list[str] | None#
only_modules: list[str] | None#
output: InternalOutputConfig#
output_dirs: dict[str, str] | None#
processor: InternalProcessorConfig#
projector: InternalProjectorConfig#
reader: InternalReaderConfig#
regridder: InternalRegridderConfig#
run_id: str | None#
segmenter: InternalSegmenterConfig#
source: str#
source_dir: str | None#
tracker: InternalTrackerConfig#
visualization: InternalVisualizationConfig#
Parameters:
class adapt.configuration.schemas.internal.InternalCoordNamesConfig(*, time, z, y, x)#

Bases: AdaptBaseModel

Runtime coordinate name mappings.

Parameters:

  • time (str)

  • z (str)

  • y (str)

  • x (str)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

time: str#
x: str#
y: str#
z: str#
class adapt.configuration.schemas.internal.InternalDownloaderConfig(*, mode, radar, output_dir, latest_files, latest_minutes, poll_interval_sec, max_fetch_retries, start_time, end_time, min_file_size, max_queue_size=100, queue_resume_fraction=0.1)#

Bases: AdaptBaseModel

Runtime downloader configuration.

Parameters:

  • mode (Literal['realtime', 'historical'])

  • radar (str)

  • output_dir (str)

  • latest_files (int)

  • latest_minutes (int)

  • poll_interval_sec (int)

  • max_fetch_retries (int)

  • start_time (str | None)

  • end_time (str | None)

  • min_file_size (int)

  • max_queue_size (int)

  • queue_resume_fraction (float)

end_time: str | None#
latest_files: int#
latest_minutes: int#
max_fetch_retries: int#
max_queue_size: int#
min_file_size: int#
mode: Literal['realtime', 'historical']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

output_dir: str#
poll_interval_sec: int#
queue_resume_fraction: float#
radar: str#
start_time: str | None#
class adapt.configuration.schemas.internal.InternalFlowParamsConfig(*, pyr_scale, levels, winsize, iterations, poly_n, poly_sigma, flags)#

Bases: AdaptBaseModel

Runtime optical flow parameters.

Parameters:

  • pyr_scale (float)

  • levels (int)

  • winsize (int)

  • iterations (int)

  • poly_n (int)

  • poly_sigma (float)

  • flags (int)

flags: int#
iterations: int#
levels: int#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

poly_n: int#
poly_sigma: float#
pyr_scale: float#
winsize: int#
class adapt.configuration.schemas.internal.InternalGlobalConfig(*, z_level, max_scan_gap_minutes=10.0, tracking_field, coord_names)#

Bases: AdaptBaseModel

Runtime global settings.

Parameters:

  • z_level (float)

  • max_scan_gap_minutes (Annotated[float, Gt(gt=0)])

  • tracking_field (str)

  • coord_names (InternalCoordNamesConfig)

coord_names: InternalCoordNamesConfig#
max_scan_gap_minutes: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

tracking_field: str#
z_level: float#
class adapt.configuration.schemas.internal.InternalLoggingConfig(*, level, console_level='WARNING', enabled=True, traces=True, metrics=True, json_logs=False, console_logs=True, progress_every=30.0)#

Bases: AdaptBaseModel

Runtime logging + observability configuration.

level governs the full file/JSON log; console_level keeps the console quiet independently. The remaining toggles enable/disable the observability subsystem and its pillars; the orchestrator translates these into an ObsSettings when it builds the provider.

Parameters:

  • level (Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])

  • console_level (Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])

  • enabled (bool)

  • traces (bool)

  • metrics (bool)

  • json_logs (bool)

  • console_logs (bool)

  • progress_every (Annotated[float, Gt(gt=0)])

console_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']#
console_logs: bool#
enabled: bool#
json_logs: bool#
level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']#
metrics: bool#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

progress_every: float#
traces: bool#
class adapt.configuration.schemas.internal.InternalOutputConfig(*, compression)#

Bases: AdaptBaseModel

Runtime output configuration.

Parameters:

compression (Literal['snappy', 'gzip', 'lz4', 'none'])

compression: Literal['snappy', 'gzip', 'lz4', 'none']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

class adapt.configuration.schemas.internal.InternalProcessorConfig(*, max_history=2, min_file_size=5000, db_filename_pattern='{radar}_cells_statistics.db')#

Bases: AdaptBaseModel

Runtime processor configuration.

Parameters:

  • max_history (Annotated[int, Ge(ge=2), Le(le=10)])

  • min_file_size (Annotated[int, Ge(ge=1000)])

  • db_filename_pattern (str)

db_filename_pattern: str#
max_history: int#
min_file_size: int#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

class adapt.configuration.schemas.internal.InternalProjectorConfig(*, method, max_projection_steps, nan_fill_value, flow_params, min_motion_threshold, max_flow_magnitude, registration_step_minutes, projection_horizon_minutes)#

Bases: AdaptBaseModel

Runtime projection configuration.

Parameters:

  • method (str)

  • max_projection_steps (Annotated[int, Ge(ge=1), Le(le=10)])

  • nan_fill_value (float)

  • flow_params (InternalFlowParamsConfig)

  • min_motion_threshold (float)

  • max_flow_magnitude (float)

  • registration_step_minutes (Annotated[int, Ge(ge=1)])

  • projection_horizon_minutes (Annotated[int, Ge(ge=1)])

flow_params: InternalFlowParamsConfig#
max_flow_magnitude: float#
max_projection_steps: int#
method: str#
min_motion_threshold: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

nan_fill_value: float#
projection_horizon_minutes: int#
registration_step_minutes: int#
class adapt.configuration.schemas.internal.InternalReaderConfig(*, file_format, field_map, fields)#

Bases: AdaptBaseModel

Runtime reader configuration.

Parameters:

  • file_format (Literal['nexrad_archive'])

  • field_map (dict[str, str])

  • fields (list[str])

field_map: dict[str, str]#
fields: list[str]#
file_format: Literal['nexrad_archive']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

class adapt.configuration.schemas.internal.InternalRegridderConfig(*, grid_shape, grid_limits, roi_func, min_radius, weighting_function)#

Bases: AdaptBaseModel

Runtime regridding configuration.

Parameters:

  • grid_shape (tuple[int, int, int])

  • grid_limits (tuple[tuple[float, float], tuple[float, float], tuple[float, float]])

  • roi_func (Literal['dist_beam', 'dist'])

  • min_radius (float)

  • weighting_function (Literal['cressman', 'barnes', 'nearest'])

grid_limits: tuple[tuple[float, float], tuple[float, float], tuple[float, float]]#
grid_shape: tuple[int, int, int]#
min_radius: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

roi_func: Literal['dist_beam', 'dist']#
weighting_function: Literal['cressman', 'barnes', 'nearest']#
class adapt.configuration.schemas.internal.InternalSegmenterConfig(*, method, min_cellsize_gridpoint, max_cellsize_gridpoint, closing_radius, filter_by_size, h_maxima, seed_carry_frames, seed_carry_min_separation, carried_exempt_size_filter=False, threshold_params, conv_strat_raut_params, conv_strat_yuter_params, feature_detection_params, steiner_conv_strat_params)#

Bases: AdaptBaseModel

Runtime segmentation configuration.

Parameters:

  • method (Literal['conv_strat_raut', 'conv_strat_yuter', 'feature_detection', 'steiner_conv_strat', 'threshold'])

  • min_cellsize_gridpoint (int)

  • max_cellsize_gridpoint (int | None)

  • closing_radius (int)

  • filter_by_size (bool)

  • h_maxima (float)

  • seed_carry_frames (int)

  • seed_carry_min_separation (int)

  • carried_exempt_size_filter (bool)

  • threshold_params (ThresholdParams)

  • conv_strat_raut_params (ConvStratRautParams)

  • conv_strat_yuter_params (ConvStratYuterParams)

  • feature_detection_params (FeatureDetectionParams)

  • steiner_conv_strat_params (SteinerConvStratParams)

carried_exempt_size_filter: bool#
closing_radius: int#
conv_strat_raut_params: ConvStratRautParams#
conv_strat_yuter_params: ConvStratYuterParams#
feature_detection_params: FeatureDetectionParams#
filter_by_size: bool#
h_maxima: float#
max_cellsize_gridpoint: int | None#
method: Literal['conv_strat_raut', 'conv_strat_yuter', 'feature_detection', 'steiner_conv_strat', 'threshold']#
min_cellsize_gridpoint: int#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

seed_carry_frames: int#
seed_carry_min_separation: int#
steiner_conv_strat_params: SteinerConvStratParams#
threshold_params: ThresholdParams#
class adapt.configuration.schemas.internal.InternalTrackerConfig(*, cell_threshold, polarity, growth_max_km, growth_size_km, max_overlap_mismatch, max_link_cost, residual_weight, heading_weight, max_speed_ms, max_acceleration_ms2, split_overlap_threshold, merge_overlap_threshold, latent_scans, identity_intensity_weight, identity_score_margin, core_field_threshold, cell_uid)#

Bases: AdaptBaseModel

Runtime tracking configuration.

Parameters:

  • cell_threshold (float)

  • polarity (Literal[1, -1])

  • growth_max_km (Annotated[float, Ge(ge=0)])

  • growth_size_km (Annotated[float, Gt(gt=0)])

  • max_overlap_mismatch (Annotated[float, Gt(gt=0), Le(le=1.0)])

  • max_link_cost (Annotated[float, Gt(gt=0)])

  • residual_weight (Annotated[float, Ge(ge=0)])

  • heading_weight (Annotated[float, Ge(ge=0)])

  • max_speed_ms (Annotated[float, Gt(gt=0)])

  • max_acceleration_ms2 (Annotated[float, Ge(ge=0)])

  • split_overlap_threshold (Annotated[float, Gt(gt=0), Le(le=1.0)])

  • merge_overlap_threshold (Annotated[float, Gt(gt=0), Le(le=1.0)])

  • latent_scans (Annotated[int, Ge(ge=0)])

  • identity_intensity_weight (Annotated[float, Ge(ge=0)])

  • identity_score_margin (Annotated[float, Ge(ge=0)])

  • core_field_threshold (float)

  • cell_uid (InternalCellUidConfig)

class InternalCellUidConfig(*, width, alphabet)#

Bases: AdaptBaseModel

Runtime cell UID configuration.

Parameters:

  • width (Annotated[int, Ge(ge=1)])

  • alphabet (Literal['base36_upper'])

alphabet: Literal['base36_upper']#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

width: int#
cell_threshold: float#
cell_uid: InternalCellUidConfig#
core_field_threshold: float#
growth_max_km: float#
growth_size_km: float#
heading_weight: float#
identity_intensity_weight: float#
identity_score_margin: float#
latent_scans: int#
max_acceleration_ms2: float#
max_overlap_mismatch: float#
max_speed_ms: float#
merge_overlap_threshold: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

polarity: Literal[1, -1]#
residual_weight: float#
split_overlap_threshold: float#
class adapt.configuration.schemas.internal.InternalVisualizationConfig(*, enabled, dpi, figsize, output_format, use_basemap, basemap_alpha, seg_linewidth, proj_linewidth, proj_alpha, flow_scale, flow_subsample, min_reflectivity, refl_vmin, refl_vmax)#

Bases: AdaptBaseModel

Runtime visualization settings.

Parameters:

  • enabled (bool)

  • dpi (int)

  • figsize (tuple[float, float])

  • output_format (Literal['png', 'pdf', 'jpeg'])

  • use_basemap (bool)

  • basemap_alpha (float)

  • seg_linewidth (float)

  • proj_linewidth (float)

  • proj_alpha (float)

  • flow_scale (float)

  • flow_subsample (int)

  • min_reflectivity (float)

  • refl_vmin (float)

  • refl_vmax (float)

basemap_alpha: float#
dpi: int#
enabled: bool#
figsize: tuple[float, float]#
flow_scale: float#
flow_subsample: int#
min_reflectivity: float#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True}#

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

output_format: Literal['png', 'pdf', 'jpeg']#
proj_alpha: float#
proj_linewidth: float#
refl_vmax: float#
refl_vmin: float#
seg_linewidth: float#
use_basemap: bool#

Configuration resolution and merging logic.

This module provides the SINGLE AUTHORITATIVE ENTRYPOINT for configuration resolution: resolve_config(). It merges ParamConfig, UserConfig, and CLIConfig in the correct precedence order and returns a validated InternalConfig.

Precedence (highest to lowest): 1. CLIConfig (command-line overrides) 2. UserConfig (user file overrides) 3. ParamConfig (expert defaults)

Merge Semantics: - Nested dictionaries: recursively merged (child keys from higher-priority

configs override lower-priority configs)

  • Lists: completely REPLACED (higher-priority list replaces entire lower-priority list; no concatenation)

  • Other values: replaced

Special Cases: - analyzer.exclude_fields: UNIONED (default excludes + user excludes) - projector.max_projection_steps: capped at 10 for safety - mode inference: auto-set to ‘historical’ if start_time/end_time provided

but no explicit mode

Example precedence: - ParamConfig: {“radar_variables”: [“A”, “B”], “threshold”: 30} - UserConfig: {“radar_variables”: [“C”], “threshold”: 40} - Result: {“radar_variables”: [“C”], “threshold”: 40} # List replaced

adapt.configuration.schemas.resolve.deep_merge(base, *overrides)#

Deep merge multiple dictionaries.

Later dictionaries override earlier ones. Nested dictionaries are merged recursively; other values are replaced.

Parameters:

  • base (dict) – Base dictionary (lowest priority)

  • *overrides (dict) – Override dictionaries (higher priority, left to right)

Returns:

Merged dictionary

Return type:

dict

Examples

>>> base = {"a": 1, "b": {"c": 2, "d": 3}}
>>> override = {"b": {"d": 4, "e": 5}, "f": 6}
>>> deep_merge(base, override)
{'a': 1, 'b': {'c': 2, 'd': 4, 'e': 5}, 'f': 6}
adapt.configuration.schemas.resolve.resolve_config(param_cfg, user_cfg=None, cli_cfg=None)#

Resolve final runtime configuration from param, user, and CLI configs.

This is the SINGLE ENTRYPOINT for configuration resolution. It validates and merges configs in the correct precedence order, then returns an immutable InternalConfig for runtime use.

Precedence (highest to lowest): 1. CLIConfig (command-line overrides) 2. UserConfig (user file overrides) 3. ParamConfig (expert defaults)

Parameters:

  • param_cfg (dict | ParamConfig) – Expert configuration with complete defaults. Required.

  • user_cfg (dict | UserConfig | None) – User configuration with overrides. If None or empty, uses only param defaults.

  • cli_cfg (dict | CLIConfig | None) – Command-line overrides. If None or empty, no CLI overrides applied.

Returns:

Fully validated, immutable runtime configuration

Return type:

InternalConfig

Raises:

ConfigError – If any config fails validation — with a clear message naming the offending section/field, its value, and why.

Examples

>>> from adapt.configuration.schemas import resolve_config, ParamConfig, UserConfig
>>>
>>> # Load expert defaults
>>> param = ParamConfig()
>>>
>>> # User wants higher threshold
>>> user = UserConfig(threshold=35, radar="KHTX")
>>>
>>> # Resolve to internal config
>>> config = resolve_config(param, user)
>>> config.segmenter.threshold
35.0
>>> config.downloader.radar
'KHTX'

Registry-driven module config engine.

Iterates registered modules and asks each to build its own frozen config from the resolved InternalConfig via build_config. Returns a dict keyed by the context key each module declares (<name>_config).

This replaces the hardcoded materialize_module_configs — adding a module never edits this file. The engine never knows which modules exist; it discovers them from the registry.

adapt.configuration.schemas.module_resolver.resolve_module_configs(internal_config)#

Build one frozen config per registered module.

For each registered module, calls build_config(internal_config). Modules that return None (need no config) contribute no key.

Return type:

dict[str, Any]

Parameters:

internal_config (Any)

On this page
Show Source