openghg_inversions.inversion_data.serialise#

Functions for saving and loading data used for inversions.

  • _save_merged_data saves the fp_all dict created by get_data.data_processing_surface_notracer to disk (either as a pickle file, netCDF, or zarr)

  • load_merged_data restores the fp_all dict from these saved formats

  • make_combined_scenario converts the fp_all dict into a xr.Dataset

DataTree and pickle loads restore stored .units metadata unchanged. fp_all_from_dataset derives the numeric scale from the combined dataset’s common mf units.

openghg_inversions.inversion_data.serialise.clear_datatree_encoding(dt: DataTree) → DataTree#

Clean encoding attribute of variables to avoid issues when writing.

openghg_inversions.inversion_data.serialise.clear_datatree_time_attrs(dt: DataTree) → DataTree#
openghg_inversions.inversion_data.serialise.combine_scenario_attrs(attrs_list: list[dict[str, Any]], context) → dict[str, Any]#

Combine attributes when concatenating scenarios from different sites.

The ModelScenario.scenario`s in `get_combined_scenario have the key “scenario” added to their attributes as a flag so this function can process the dataset attributes and the data variable attributes differently.

TODO: add ‘time_period’, ‘high_time/spatial_resolution’, ‘short_lifetime’, ‘heights’?

Is ‘time_period’ from the footprint? Need to check model scenario…

Parameters:

  • attrs_list – list of attributes from datasets being concatenated

  • context – additional parameter supplied by concatenate (this is required/supplied by xarray)

Returns:

dict that will be used as attributes for concatenated dataset

openghg_inversions.inversion_data.serialise.dataset_to_bc_data(ds: Dataset) → BoundaryConditionsData#
openghg_inversions.inversion_data.serialise.dataset_to_flux_data(ds: Dataset) → FluxData#
openghg_inversions.inversion_data.serialise.datatree_compression_encoding(dt: DataTree, compressor: Blosc) → dict#

Creating encoding dictionary for saving DataTree to zarr.

openghg_inversions.inversion_data.serialise.datatree_to_flux_dict(dt: DataTree) → dict[str, FluxData]#

Convert an xarray DataTree to a dict of FluxData objects.

Parameters:

dt – DataTree whose child nodes are converted to datasets and then to FluxData.

Returns:

Mapping from node keys (as strings) to FluxData instances.

openghg_inversions.inversion_data.serialise.datatree_to_fp_all(dt: DataTree) → dict#
openghg_inversions.inversion_data.serialise.flux_dict_to_datatree(flux_dict: dict[str, FluxData], netcdf_safe_attrs: bool = False) → DataTree#
openghg_inversions.inversion_data.serialise.fp_all_from_dataset(ds: Dataset) → dict#

Recover “fp_all” dictionary from “combined scenario” dataset.

This is the inverse of make_combined_scenario, except that the attributes of the scenarios, fluxes, and boundary conditions may be different. New datasets retain source timestamps on flux_time with explicit source presence; older datasets without that metadata use value-based padding removal or fall back to the first observation time for compatibility.

Parameters:

ds – dataset created by make_combined_scenario

Returns:

dictionary containing model scenarios keyed by site, as well as flux and boundary conditions.

Raises:

ValueError – If serialized mf units are invalid or are not a molar mixing ratio. Missing units default to mol/mol.

openghg_inversions.inversion_data.serialise.fp_all_to_datatree(fp_all: dict, netcdf_safe_attrs: bool = False) → DataTree#
openghg_inversions.inversion_data.serialise.load_merged_data(merged_data_dir: str | Path, species: str | None = None, start_date: str | None = None, output_name: str | None = None, merged_data_name: str | None = None, output_format: Literal['pickle', 'netcdf', 'zarr', 'zarr.zip'] | None = None) → dict#

Load fp_all dictionary from a file in merged_data_dir.

The name of the pickle file can be specified using merged_data_name, or a standard name will be created given species, start_date, and output_name.

If merged_data_name is not given, then species, start_date, and output_name must be provided.

This function tries to automatically find a compatible format of merged data, if a format is not specified. First, it checks for data in “zarr” (or zipped zarr) format, then in netCDF, and finally in pickle.

Note: if data is stored in a zarr ZipStore, then the data is eagerly loaded, since the data needs to loaded before the zip file is closed.

Parameters:

  • merged_data_dir – path to directory where merged data will be saved

  • species – species of inversion

  • start_date – start date of inversion period

  • output_name – output name parameter used for inversion run

  • merged_data_name – name to use for saved data.

  • output_format – format of data to load (if not specified, this will be inferred).

Returns:

fp_all dictionary

openghg_inversions.inversion_data.serialise.make_combined_scenario(fp_all: dict) → Dataset#

Combine scenarios and merge in fluxes and boundary conditions.

Flux time coordinates are stored on a separate flux_time dimension, with explicit per-source timestamp presence, so that source periods beginning before the observations and all-NaN slices are preserved. Singleton boundary-condition time dimensions are still dropped.

Parameters:

fp_all – Inversion data keyed by site, with flux sources under ".flux" and optional boundary conditions under ".bc".

Returns:

A combined dataset containing site scenarios, source-specific flux periods, and optional boundary conditions.

openghg_inversions.inversion_data.serialise.openghg_data_to_dataset(openghg_data: _BaseData, netcdf_safe_attrs: bool = False) → Dataset#