Source code for pyarts.arts_ext.XsecRecordExtras

from datetime import datetime
import numpy as np
import xarray

import pyarts.arts as cxx


[docs] def to_xarray(self): """Convert XsecRecord to xarray dataset""" attrs = {'creation_date': str(datetime.now())} coords = { "coeffs": ("coeffs", range(4), { "unit": "1", "long_name": "Coefficients p00, p10, p01, p20" }), "bands": ("bands", range(len(self.fitcoeffs)), { "unit": "1", "long_name": "Band indices" }) } data_vars = {} data_vars["fitminpressures"] = ("bands", self.fitminpressures, { "unit": "Pa", "long_name": "Mininum pressures from fit" }) data_vars["fitmaxpressures"] = ("bands", self.fitmaxpressures, { "unit": "Pa", "long_name": "Maximum pressures from fit" }) data_vars["fitmintemperatures"] = ("bands", self.fitmintemperatures, { "unit": "K", "long_name": "Minimum temperatures from fit" }) data_vars["fitmaxtemperatures"] = ("bands", self.fitmaxtemperatures, { "unit": "K", "long_name": "Maximum temperatures from fit" }) for i, band in enumerate(self.fitcoeffs): coords[f"band{i}_fgrid"] = (f"band{i}_fgrid", band.get_grid(0), { "unit": "Hz", "long_name": f"Frequency grid for band {i}" }) data_vars[f"band{i}_coeffs"] = ([f"band{i}_fgrid", "coeffs"], band.data, { 'unit': "m", 'long_name': f'Fit coefficients for band {i}' }) attrs["species"] = str(self.species) attrs["version"] = self.version return xarray.Dataset(data_vars, coords, attrs)
[docs] def from_xarray(cls, ds): """Set XsecRecord from xarray dataset""" xr = cls() coeff_grid = cxx.ArrayOfString(["p00", "p10", "p01", "p20"]) gfs = [] xr.species = ds.attrs["species"] xr.version = ds.attrs["version"] xr.fitminpressures = ds["fitminpressures"].values xr.fitmaxpressures = ds["fitmaxpressures"].values xr.fitmintemperatures = ds["fitmintemperatures"].values xr.fitmaxtemperatures = ds["fitmaxtemperatures"].values for i in ds["bands"].values: g = cxx.GriddedField2() for j, gridname in enumerate(["frequency grid [Hz]", "fit coefficients [m]"]): g.set_grid_name(j, gridname) g.set_grid(0, ds[f"band{i}_fgrid"].values) g.set_grid(1, coeff_grid) g.data = ds[f"band{i}_coeffs"].values gfs.append(g) xr.fitcoeffs = cxx.ArrayOfGriddedField2(gfs) return xr
[docs] def to_netcdf(self, filename): """Save XsecRecord to NetCDF file""" self.to_xarray().to_netcdf(filename)
[docs] def from_netcdf(cls, filename): """Read XsecRecord from NetCDF file""" return cls.from_xarray(xarray.load_dataset(filename))
def xsecrecord__eq__(self, other): if isinstance(other, cxx.XsecRecord) and \ self.version == other.version and \ str(self.species) == str(other.species) and \ np.all(self.fitminpressures == other.fitminpressures) and \ np.all(self.fitmaxpressures == other.fitmaxpressures) and \ np.all(self.fitmintemperatures == other.fitmintemperatures) and \ np.all(self.fitmaxtemperatures == other.fitmaxtemperatures) and \ np.all([np.all(np.array(i.data) == np.array(j.data)) for i, j in zip(self.fitcoeffs, other.fitcoeffs)]): return True else: return False getattr(cxx, "_detailsXsecRecord").to_xarray = to_xarray getattr(cxx, "_detailsXsecRecord").from_xarray = from_xarray getattr(cxx, "_detailsXsecRecord").to_netcdf = to_netcdf getattr(cxx, "_detailsXsecRecord").from_netcdf = from_netcdf getattr(cxx, "_detailsXsecRecord").__eq__ = xsecrecord__eq__