plans.tools.analysis_spatial_normals#

analysis_spatial_normals – standalone PLANS tool.

Self-contained script: copy this single file anywhere and run it with Python 3. It has no dependency on the plans.tools.core module – all parameters are passed through a single JSON config file.

From a monthly raster time series (e.g., NDVI), this tool builds:

  1. Annual aggregations (one or more statistics per year).

  2. Climatological normals – annual and monthly, over a configurable horizon.

  3. Anomalies – monthly and annual, absolute and relative/percent.

Assumes all input rasters share the same grid (shape, transform, CRS), and that monthly raster filenames end with a fixed "_<YYYY>_<MM>" suffix before the extension, month zero-padded, e.g. NDVI_2020_01.tif. The only third-party dependencies are numpy and rasterio.

Usage#

python analysis_spatial_normals.py --config config.json

Config file#

The --config/-c flag points to a JSON file. Template, ready to copy and paste:

{
  "input": "data/ndvi_monthly",
  "output": "output/run1",
  "label": "ndvi_normals_v1",

  "pattern": "*.tif",
  "verbose": true,
  "annual": {
    "stats": ["mean", "sum", "max", "p90"],
    "min_months": 12
  },
  "normals": {
    "stats": ["mean", "std"],
    "annual_base_stat": "mean",
    "annual_horizon": [2000, 2020],
    "monthly_horizon": null
  },
  "anomalies": {
    "monthly": true,
    "annual": true,
    "relative": true
  },
  "styles": {
    "annual": "styles/annual.qml",
    "annual_normal": "styles/annual_normal.qml",
    "monthly_normal": "styles/monthly_normal.qml",
    "anomaly_absolute": "styles/anomaly_absolute.qml",
    "anomaly_relative": "styles/anomaly_relative.qml"
  }
}

Required fields#

  • input (str) – folder containing the monthly raster files.

  • output (str) – output folder for run artifacts.

  • label (str) – run label, used as the logger name.

Optional fields#

  • pattern (str, default "*.tif") – glob pattern for input files.

  • verbose (bool, default false) – echo logs to console.

  • annual (dict, default {}) – see aggregate_annual(); keys stats (list[str], default ["mean"]) and min_months (int, default 12).

  • normals (dict, default {}) – see compute_annual_normals() / compute_monthly_normals(); keys stats (list[str], default ["mean"]), annual_base_stat (str, default "mean"), annual_horizon and monthly_horizon ([start_year, end_year] or null, default null).

  • anomalies (dict, default {}) – see compute_monthly_anomalies() / compute_annual_anomalies(); keys monthly and annual (bool, default true), relative (bool, default true).

  • styles (dict, default {}) – see apply_style(). Each key is optional; when present, its value is a path to a QGIS .qml style template that gets copied as a same-named sidecar next to every raster in that output category, so QGIS auto-applies it on load. Keys: annual (the per-year aggregates), annual_normal, monthly_normal, anomaly_absolute (used for both monthly and annual absolute anomalies), and anomaly_relative (ditto, relative anomalies).

The REQUIRED/OPTIONAL dictionaries just below are the actual validation source of truth for the top-level fields – keep them in sync with this docstring whenever you add, rename or remove a field. Nested sub-dict defaults (annual, normals, anomalies) are filled in by process_data().

Functions

aggregate_annual(monthly_rasters, ...[, ...])

Aggregate monthly rasters into one raster per year, per requested stat.

apply_style(paths_obj, template_path, logger)

Copy a QGIS .qml style template as a same-named sidecar for each raster.

compute_annual_anomalies(...[, base_stat, ...])

Compute per-year anomalies against the mean annual normal.

compute_annual_normals(annual_paths_by_stat, ...)

Compute annual normals from a per-year annual series.

compute_monthly_anomalies(monthly_rasters, ...)

Compute per-month anomalies against the calendar-month normal.

compute_monthly_normals(monthly_rasters, ...)

Compute monthly (calendar-month) climatological normals.

export_data(processed, cfg, logger)

Write a JSON manifest listing every raster produced by the run.

get_args()

Parse the single --config/-c CLI argument.

get_logger(name, log_file[, talk])

Create a simple console + file logger.

load_config(file_config)

Load, validate and fill defaults for the JSON config file.

load_data(cfg, logger)

Scan and validate the monthly raster time series described by cfg.

main()

Entry point: load the config and run load -> process -> export.

parse_monthly_rasters(folder, logger[, pattern])

Scan a folder for monthly rasters following the fixed naming convention.

process_data(loaded, cfg, logger)

Run annual aggregation, climatological normals, and anomalies.

validate_grid_consistency(paths, logger)

Check that a set of rasters share the same grid.

Classes

MonthlyRaster(year, month, path)

Reference to a single monthly raster file.
plans.tools.analysis_spatial_normals.load_config(file_config)[source]#

Load, validate and fill defaults for the JSON config file.

Checks that every key in REQUIRED is present and of the right type, then fills in defaults for any missing key in OPTIONAL and type-checks those that were provided.

Parameters:

file_config (str or pathlib.Path) – path to the JSON config file.

Returns:

validated configuration dictionary, defaults included.

Return type:

dict

Raises:

ValueError – if a required field is missing, or any field has the wrong type.

plans.tools.analysis_spatial_normals.get_logger(name, log_file, talk=True)[source]#

Create a simple console + file logger.

Parameters:

  • name (str) – logger name, shown in every log line.

  • log_file (str or pathlib.Path) – path to the log file to write to.

  • talk (bool) – whether to also emit log records to the console.

Returns:

configured logger instance.

Return type:

logging.Logger

class plans.tools.analysis_spatial_normals.MonthlyRaster(year: int, month: int, path: str)[source]#

Bases: object

Reference to a single monthly raster file.

Parameters:

  • year (int) – Calendar year, e.g. 2020.

  • month (int) – Calendar month, 1-12.

  • path (str) – Path to the raster file on disk.

year: int#
month: int#
path: str#
__dataclass_fields__ = {'month': Field(name='month',type=<class 'int'>,default=<dataclasses._MISSING_TYPE object>,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'path': Field(name='path',type=<class 'str'>,default=<dataclasses._MISSING_TYPE object>,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'year': Field(name='year',type=<class 'int'>,default=<dataclasses._MISSING_TYPE object>,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}#
__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)#
__eq__(other)#

Return self==value.

__hash__ = None#
__init__(year: int, month: int, path: str) → None#
__match_args__ = ('year', 'month', 'path')#
__repr__()#

Return repr(self).

plans.tools.analysis_spatial_normals._resolve_stat_func(stat: str) → Callable[source]#

Resolve a stat name to a NaN-aware (array, axis) -> array reducer.

Parameters:

stat (str) – Statistic name. One of the keys in _BASE_STAT_FUNCS ("mean", "sum", "max", "min", "median", "std"), or "pN" for the Nth percentile, e.g. "p10", "p90".

Returns:

A reducer function usable as func(stack, axis=0).

Return type:

callable

Raises:

ValueError – If stat is not a recognized name, or a percentile is outside the 0-100 range.

plans.tools.analysis_spatial_normals.parse_monthly_rasters(folder, logger, pattern: str = '*.tif') → list[MonthlyRaster][source]#

Scan a folder for monthly rasters following the fixed naming convention.

Files must end with a "_<YYYY>_<MM>" suffix before the extension, month zero-padded, e.g. NDVI_2020_01.tif.

Parameters:

  • folder (str or pathlib.Path) – Directory to scan.

  • logger (logging.Logger) – Logger instance for warnings about unmatched filenames.

  • pattern (str) – Glob pattern used to list candidate files.

Returns:

Parsed monthly rasters, sorted chronologically.

Return type:

list[MonthlyRaster]

plans.tools.analysis_spatial_normals.validate_grid_consistency(paths: list[str], logger) → None[source]#

Check that a set of rasters share the same grid.

Parameters:

  • paths (list[str]) – Raster file paths to compare.

  • logger (logging.Logger) – Logger instance for a confirmation message.

Returns:

None

Return type:

None

Raises:

ValueError – If any raster’s width, height, transform, or CRS differs from the first raster’s.

plans.tools.analysis_spatial_normals._read_as_float(path) → tuple[numpy.ndarray, dict][source]#

Read band 1 of a raster as float64, converting nodata to NaN.

Parameters:

path (str or pathlib.Path) – Path to the raster file.

Returns:

A 2-tuple of the pixel array and the rasterio profile.

Return type:

tuple[numpy.ndarray, dict]

plans.tools.analysis_spatial_normals._write_raster(arr: numpy.ndarray, profile: dict, out_path) → None[source]#

Write a single-band float32 raster with NaN nodata.

Parameters:

  • arr (numpy.ndarray) – Pixel array to write.

  • profile (dict) – Rasterio profile to reuse (CRS, transform, etc.); its dtype, count, and nodata are overridden.

  • out_path (str or pathlib.Path) – Output file path; parent directories are created if needed.

Returns:

None

Return type:

None

plans.tools.analysis_spatial_normals._apply_stat(stack: numpy.ndarray, func: Callable) → numpy.ndarray[source]#

Apply a stat reducer across axis 0 of a stack.

Silences the expected “all-NaN slice” warning for pixels that are nodata in every layer of the stack (those pixels correctly come out as NaN in the result).

Parameters:

  • stack (numpy.ndarray) – Array stacked along axis 0 (e.g. one layer per month or year).

  • func (callable) – Reducer called as func(stack, axis=0).

Returns:

The reduced array.

Return type:

numpy.ndarray

plans.tools.analysis_spatial_normals._read_stack(records: list[MonthlyRaster]) → tuple[numpy.ndarray, dict][source]#

Read a list of rasters into a single stacked array.

Parameters:

records (list[MonthlyRaster]) – Rasters to read, in the order they should be stacked.

Returns:

A 2-tuple of the stacked array (axis 0 = record index) and the rasterio profile of the first record.

Return type:

tuple[numpy.ndarray, dict]

plans.tools.analysis_spatial_normals.aggregate_annual(monthly_rasters: list[MonthlyRaster], output_dir, logger, stats: list[str] = ('mean',), min_months: int = 12, file_prefix: str = 'annual') → dict[str, dict[int, str]][source]#

Aggregate monthly rasters into one raster per year, per requested stat.

Parameters:

  • monthly_rasters (list[MonthlyRaster]) – Parsed monthly rasters, as returned by parse_monthly_rasters().

  • output_dir (str or pathlib.Path) – Directory to write <file_prefix>_<stat>_<year>.tif files into.

  • logger (logging.Logger) – Logger instance for progress messages.

  • stats (list[str]) – Statistics to compute per year. See _resolve_stat_func() for accepted names.

  • min_months (int) – Minimum number of valid months required for a year to be aggregated; years with fewer are skipped (with a warning).

  • file_prefix (str) – Output filename prefix.

Returns:

Mapping of {stat: {year: output_path}}.

Return type:

dict[str, dict[int, str]]

plans.tools.analysis_spatial_normals.compute_annual_normals(annual_paths_by_stat: dict[str, dict[int, str]], output_dir, logger, base_stat: str = 'mean', stats: list[str] = ('mean',), horizon: tuple[int, int] | None = None, file_prefix: str = 'annual_normal') → dict[str, str][source]#

Compute annual normals from a per-year annual series.

For the annual series identified by base_stat (e.g. the per-year "mean" rasters from aggregate_annual()), compute each stat in stats across the years within horizon.

Parameters:

  • annual_paths_by_stat (dict[str, dict[int, str]]) – Per-year annual rasters, as returned by aggregate_annual().

  • output_dir (str or pathlib.Path) – Directory to write <file_prefix>_<stat>.tif files into.

  • logger (logging.Logger) – Logger instance for progress messages.

  • base_stat (str) – Which annual series in annual_paths_by_stat to use as input.

  • stats (list[str]) – Statistics to compute across years. See _resolve_stat_func() for accepted names.

  • horizon (tuple[int, int] or None) – Inclusive (start_year, end_year) window; None uses every available year.

  • file_prefix (str) – Output filename prefix.

Returns:

Mapping of {stat: output_path}.

Return type:

dict[str, str]

Raises:

ValueError – If base_stat was not computed in annual_paths_by_stat, or no years fall within horizon.

plans.tools.analysis_spatial_normals.compute_monthly_normals(monthly_rasters: list[MonthlyRaster], output_dir, logger, stats: list[str] = ('mean',), horizon: tuple[int, int] | None = None, file_prefix: str = 'normal_month') → dict[str, dict[int, str]][source]#

Compute monthly (calendar-month) climatological normals.

For each calendar month (1-12), compute each stat in stats across the years within horizon. horizon is independent of the horizon used for the annual normals.

Parameters:

  • monthly_rasters (list[MonthlyRaster]) – Parsed monthly rasters, as returned by parse_monthly_rasters().

  • output_dir (str or pathlib.Path) – Directory to write <file_prefix>_<MM>_<stat>.tif files into.

  • logger (logging.Logger) – Logger instance for progress messages.

  • stats (list[str]) – Statistics to compute per calendar month. See _resolve_stat_func() for accepted names.

  • horizon (tuple[int, int] or None) – Inclusive (start_year, end_year) window; None uses every available year.

  • file_prefix (str) – Output filename prefix.

Returns:

Mapping of {stat: {month: output_path}}.

Return type:

dict[str, dict[int, str]]

plans.tools.analysis_spatial_normals.compute_monthly_anomalies(monthly_rasters: list[MonthlyRaster], monthly_normal_paths_by_stat: dict[str, dict[int, str]], output_dir, logger, relative: bool = True, normal_stat: str = 'mean') → dict[str, dict[tuple[int, int], str]][source]#

Compute per-month anomalies against the calendar-month normal.

For every monthly raster, computes:

anomaly          = value - normal[month]
relative_anomaly = (value - normal[month]) / normal[month] * 100   # percent

where normal is the normal_stat monthly normal ("mean" by default – anomalies are conventionally defined relative to the mean). Pixels where the normal is exactly 0 get NaN for the relative anomaly (percent deviation is undefined there) but still get a valid absolute anomaly – relevant for variables like precipitation with dry-season zeros.

Parameters:

  • monthly_rasters (list[MonthlyRaster]) – Parsed monthly rasters, as returned by parse_monthly_rasters().

  • monthly_normal_paths_by_stat (dict[str, dict[int, str]]) – Monthly normals, as returned by compute_monthly_normals().

  • output_dir (str or pathlib.Path) – Base directory; absolute/ and relative/ subdirectories are created under it.

  • logger (logging.Logger) – Logger instance for progress messages.

  • relative (bool) – Whether to also compute the percent anomaly.

  • normal_stat (str) – Which monthly normal stat to use as the baseline.

Returns:

{"absolute": {(year, month): path}, "relative": {(year, month): path}}.

Return type:

dict[str, dict[tuple[int, int], str]]

Raises:

ValueError – If normal_stat was not computed in monthly_normal_paths_by_stat.

plans.tools.analysis_spatial_normals.compute_annual_anomalies(annual_paths_by_stat: dict[str, dict[int, str]], annual_normal_paths_by_stat: dict[str, str], output_dir, logger, base_stat: str = 'mean', relative: bool = True) → dict[str, dict[int, str]][source]#

Compute per-year anomalies against the mean annual normal.

For every year in the annual series identified by base_stat (e.g. the per-year "sum" rasters from aggregate_annual(), for a total like annual precipitation), computes:

anomaly          = annual_value[year] - annual_normal_mean
relative_anomaly = (annual_value[year] - annual_normal_mean) / annual_normal_mean * 100   # percent

The baseline is always the "mean" entry of annual_normal_paths_by_stat, matching the convention used for the monthly anomalies.

Parameters:

  • annual_paths_by_stat (dict[str, dict[int, str]]) – Per-year annual rasters, as returned by aggregate_annual().

  • annual_normal_paths_by_stat (dict[str, str]) – Annual normals, as returned by compute_annual_normals(); must include a "mean" entry.

  • output_dir (str or pathlib.Path) – Base directory; absolute/ and relative/ subdirectories are created under it.

  • logger (logging.Logger) – Logger instance for progress messages.

  • base_stat (str) – Which annual series in annual_paths_by_stat to compute anomalies for.

  • relative (bool) – Whether to also compute the percent anomaly.

Returns:

{"absolute": {year: path}, "relative": {year: path}}.

Return type:

dict[str, dict[int, str]]

Raises:

ValueError – If base_stat was not computed in annual_paths_by_stat, or the "mean" annual normal is missing.

plans.tools.analysis_spatial_normals._iter_paths(obj)[source]#

Recursively yield every path string found in a nested dict/list structure.

Walks arbitrarily nested dicts and lists/tuples (matching the shapes returned by aggregate_annual(), compute_annual_normals(), compute_monthly_normals(), compute_monthly_anomalies(), and compute_annual_anomalies()) and yields every string value found.

Parameters:

obj (dict or list or tuple or str) – nested paths structure.

Yields:

each path string found.

Return type:

collections.abc.Iterator[str]

plans.tools.analysis_spatial_normals.apply_style(paths_obj, template_path, logger) → list[str][source]#

Copy a QGIS .qml style template as a same-named sidecar for each raster.

For every raster path found in paths_obj (any nesting of dict/list, as returned by the pipeline step functions), copies template_path to <raster>.qml – i.e. the raster’s filename with its extension replaced by .qml – so QGIS auto-applies the style when the raster is added to the map.

Parameters:

  • paths_obj (dict) – nested output-paths structure to style.

  • template_path (str or pathlib.Path) – path to the .qml style template to copy.

  • logger (logging.Logger) – logger instance for progress messages.

Returns:

paths of the created .qml sidecar files.

Return type:

list[str]

Raises:

FileNotFoundError – if template_path does not exist.

plans.tools.analysis_spatial_normals._maybe_apply_style(paths_obj, template_path, logger) → None[source]#

Call apply_style() only if template_path is truthy.

Small convenience wrapper so callers can pass styles_cfg.get(key) (which is None for an unconfigured style) without an if at every call site.

Parameters:

  • paths_obj (dict) – nested output-paths structure to style.

  • template_path (str or pathlib.Path or None) – path to the .qml style template, or None to skip styling.

  • logger (logging.Logger) – logger instance for progress messages.

Returns:

None

Return type:

None

plans.tools.analysis_spatial_normals.load_data(cfg, logger)[source]#

Scan and validate the monthly raster time series described by cfg.

Parameters:

  • cfg (dict) – validated configuration dictionary from load_config().

  • logger (logging.Logger) – logger instance for progress messages.

Returns:

parsed monthly rasters, sorted chronologically.

Return type:

list[MonthlyRaster]

Raises:

ValueError – if no monthly rasters are found in cfg["input"].

plans.tools.analysis_spatial_normals.process_data(loaded, cfg, logger)[source]#

Run annual aggregation, climatological normals, and anomalies.

Parameters:

  • loaded (list[MonthlyRaster]) – monthly rasters, as returned by load_data().

  • cfg (dict) – validated configuration dictionary from load_config().

  • logger (logging.Logger) – logger instance for progress messages.

Returns:

dict with keys "annual", "annual_normals", "monthly_normals", "annual_anomalies", and "monthly_anomalies", holding the structures returned by aggregate_annual(), compute_annual_normals(), compute_monthly_normals(), compute_annual_anomalies(), and compute_monthly_anomalies() respectively. The two anomaly entries are None if disabled in the config. If cfg["styles"] has entries, apply_style() is also called for each configured category as a side effect (writing .qml sidecars next to the relevant rasters).

Return type:

dict

plans.tools.analysis_spatial_normals._manifest_safe(obj)[source]#

Recursively convert a processed results structure into a JSON-safe form.

(year, month) tuple keys become "YYYY-MM" strings; everything else is passed through unchanged.

Parameters:

obj (object) – value to convert (dict, list/tuple, or scalar).

Returns:

JSON-serializable equivalent.

Return type:

object

plans.tools.analysis_spatial_normals.export_data(processed, cfg, logger)[source]#

Write a JSON manifest listing every raster produced by the run.

Parameters:

  • processed (dict) – data returned by process_data().

  • cfg (dict) – validated configuration dictionary from load_config().

  • logger (logging.Logger) – logger instance for progress messages.

Returns:

None

Return type:

None

plans.tools.analysis_spatial_normals.get_args()[source]#

Parse the single --config/-c CLI argument.

Returns:

parsed arguments namespace, exposing args.config.

Return type:

argparse.Namespace

plans.tools.analysis_spatial_normals.main()[source]#

Entry point: load the config and run load -> process -> export.

Returns:

None

Return type:

None