Reference

This reference manual details functions, modules, and objects included in joseki, describing what they are and what they do.

Accessor

Accessor module.

JosekiAccessor

Joseki accessor.

Source code in src/joseki/accessor.py

@xr.register_dataset_accessor("joseki")
class JosekiAccessor:  # pragma: no cover
    """Joseki accessor."""

    def __init__(self, xarray_obj):
        self._obj = xarray_obj

    @property
    def molecules(self) -> t.List[str]:
        """Return list of molecules."""
        return [c[2:] for c in self._obj.data_vars if c.startswith("x_")]

    @property
    def column_number_density(
        self,
    ) -> t.Dict[str, pint.Quantity]:
        r"""Compute column number density.

        Returns:
            A mapping of molecule and column number density.

        Notes:
            The column number density is given by:

            $$
            N_{\mathrm{M}} = \int n_{\mathrm{M}}(z) \, \mathrm{d} z
            $$

            with

            $$
            n_{\mathrm{M}}(z) = x_{\mathrm{M}}(z) \, n(z)
            $$

            where

            * $z$ is the altitude,
            * $x_{\mathrm{M}}(z)$ is the volume mixing ratio of molecule M
            at altitude $z$,
            * $n(z)$ is the air number density at altitude $z$,
            * $n_{\mathrm{M}}(z)$ is the number density of molecule M at
            altitude $z$.

            If the dataset has a `z_bounds` coordinate, the integral is computed
            using the centered rectangle method, where the `z` coordinate
            corresponds to the rectangle centers.

            If the dataset does not have a `z_bounds` coordinate, the 
            integration is performed using the trapezoidal rule.
        """
        ds = self._obj
        try:
            with xr.set_options(keep_attrs=True):
                dz = to_quantity(ds.z_bounds.diff(dim="zbv", n=1).squeeze())

            logger.debug(
                "Computing column number density using the centered rectangle "
                "rule."
            )

            n = to_quantity(ds.n)

            _column_number_density = {}
            for m in self.molecules:
                xm = to_quantity(ds[f"x_{m}"])
                _column_number_density[m] = (
                    (xm * n * dz).sum().to_base_units()
                )  # integrate using the centered rectangle rule

            return _column_number_density

        except AttributeError:  # z_bounds attribute does not exist

            logger.debug(
                "Computing column number density using the trapezoidal rule."
            )

            _column_number_density = {}
            for m in self.molecules:
                integral = (ds[f"x_{m}"] * ds.n).integrate(
                    coord="z"
                )  # integrate using the trapezoidal rule
                units = " ".join(
                    [ds[var].attrs["units"] for var in [f"x_{m}", "n", "z"]]
                )
                _column_number_density[m] = (
                    integral.values * ureg.Unit(units)
                ).to_base_units()

            return _column_number_density

    @property
    def column_mass_density(
        self,
    ) -> t.Dict[str, pint.Quantity]:
        r"""Compute column mass density.

        Returns:
            A mapping of molecule and column mass density.

        Notes:
            The column mass density is given by:

            $$
            \sigma_{\mathrm{M}} = N_{\mathrm{M}} \, m_{\mathrm{M}}
            $$

            where

            * $N_{\mathrm{M}}$ is the column number density of molecule M,
            * $m_{\mathrm{M}}$ is the molecular mass of molecule M.
        """
        _column_number_density = self.column_number_density
        return {
            m: (molecular_mass(m) * _column_number_density[m]).to("kg/m^2")
            for m in self.molecules
        }

    @property
    def number_density_at_sea_level(
        self,
    ) -> t.Dict[str, pint.Quantity]:
        """Compute number density at sea level.

        Returns:
            A mapping of molecule and number density at sea level.
        """
        ds = self._obj
        n = to_quantity(ds.n.isel(z=0))
        return {m: (to_quantity(ds[f"x_{m}"].isel(z=0)) * n) for m in self.molecules}

    @property
    def mass_density_at_sea_level(
        self,
    ) -> t.Dict[str, pint.Quantity]:
        """Compute mass density at sea level.

        Returns:
            A mapping of molecule and mass density at sea level.
        """
        _number_density_at_sea_level = self.number_density_at_sea_level
        return {
            m: (molecular_mass(m) * _number_density_at_sea_level[m]).to("kg/m^3")
            for m in self.molecules
        }

    @property
    def volume_fraction_at_sea_level(
        self,
    ) -> t.Dict[str, pint.Quantity]:
        """Compute volume fraction at sea level.

        Returns:
            A mapping of molecule and volume mixing fraction at sea level.
        """
        ds = self._obj
        return {m: to_quantity(ds[f"x_{m}"].isel(z=0)) for m in self.molecules}

    @property
    def volume_fraction(self) -> xr.DataArray:
        """Extract volume fraction and tabulate as a function of (m, z).

        Returns:
            Volume fraction.
        """
        ds = self._obj
        molecules = self.molecules
        concatenated = xr.concat([ds[f"x_{m}"] for m in molecules], dim="m")
        concatenated["m"] = ("m", molecules, {"long_name": "molecule"})
        concatenated.attrs.update(
            {
                "standard_name": "volume_fraction",
                "long_name": "volume fraction",
                "units": "dimensionless",
            }
        )
        concatenated.name = "x"
        return concatenated

    def scaling_factors(
        self, target: t.MutableMapping[str, pint.Quantity]
    ) -> t.MutableMapping[str, float]:
        """Compute scaling factor(s) to reach specific target amount(s).

        Args:
            target: Mapping of molecule and target amount.

        Raises:
            ValueError: If a target amount has dimensions that are not supported.

        Returns:
            Mapping of molecule and scaling factors.

        Notes:
            For each molecule in the ``target`` mapping, the target amount is
            interpreted, depending on its dimensions (indicated in square 
            brackets), as:

            * a column number density [`length^-2`],
            * a column mass density [`mass * length^-2`],
            * a number densitx at sea level [`length^-3`],
            * a mass density at sea level [`mass * length^-3`],
            * a volume mixing fraction at sea level [`dimensionless`]

            The scaling factor is then evaluated as the ratio of the target amount
            with the original amount, for each molecule.

        See Also:
            `rescale`
        """
        compute_initial_amount = {
            "[length]^-2": self.column_number_density,
            "[mass] * [length]^-2": self.column_mass_density,
            "[length]^-3": self.number_density_at_sea_level,
            "[mass] * [length]^-3": self.mass_density_at_sea_level,
            "": self.volume_fraction_at_sea_level,
        }
        factors = {}
        for m, target_amount in target.items():
            initial_amount = None
            for dim in compute_initial_amount.keys():
                if target_amount.check(dim):
                    initial_amount = compute_initial_amount[dim][m]
            if initial_amount is None:
                raise ValueError
            factors[m] = _scaling_factor(
                initial_amount=initial_amount, target_amount=target_amount
            )
        return factors

    def rescale(
        self,
        factors: t.MutableMapping[str, float],
        check_volume_fraction_sum: bool = False
    ) -> xr.Dataset:
        """Rescale molecules concentration in atmospheric profile.

        Args:
            factors: A mapping of molecule and scaling factor.
            check_volume_fraction_sum: if True, check that volume fraction sums
                are never larger than one.
        Raises:
            ValueError: if check_volume_fraction_sum is `True` and the 
                dataset is not valid.

        Returns:
            Rescaled dataset (new object).
        """
        ds = self._obj

        # update volume fraction
        x_new = {}
        for m in factors:
            with xr.set_options(keep_attrs=True):
                x_new[f"x_{m}"] = ds[f"x_{m}"] * factors[m]

        ds = ds.assign(x_new)

        # validate rescaled dataset
        try:
            ds.joseki.validate(check_volume_fraction_sum=check_volume_fraction_sum)
        except ValueError as e:
            raise ValueError("Cannot rescale") from e

        # update history attribute
        now = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
        for m in factors.keys():
            ds.attrs["history"] += (
                f"\n{now} - rescaled {m}'s volume mixing ratio using a scaling "
                f"factor of {factors[m]:.3f} - joseki, version {__version__}"
            )

        return ds

    def validate(
        self,
        check_volume_fraction_sum: bool = False,
        ret_true_if_valid: bool = False,
    ) -> bool:
        """Validate atmosphere thermophysical profile dataset schema.

        Returns:
            `True` if the dataset complies with the schema, else `False`.
        """
        return schema.validate(
            ds=self._obj,
            check_volume_fraction_sum=check_volume_fraction_sum,
            ret_true_if_valid=ret_true_if_valid,
        )

    @property
    def is_valid(self):
        return self.validate(ret_true_if_valid=True)

column_mass_density: t.Dict[str, pint.Quantity] property

Compute column mass density.

Returns:

Type	Description
t.Dict[str, pint.Quantity]	A mapping of molecule and column mass density.

Notes

The column mass density is given by:

\[ \sigma_{\mathrm{M}} = N_{\mathrm{M}} \, m_{\mathrm{M}} \]

where

• \(N_{\mathrm{M}}\) is the column number density of molecule M,
• \(m_{\mathrm{M}}\) is the molecular mass of molecule M.

column_number_density: t.Dict[str, pint.Quantity] property

Compute column number density.

Returns:

Type	Description
t.Dict[str, pint.Quantity]	A mapping of molecule and column number density.

Notes

The column number density is given by:

\[ N_{\mathrm{M}} = \int n_{\mathrm{M}}(z) \, \mathrm{d} z \]

with

\[ n_{\mathrm{M}}(z) = x_{\mathrm{M}}(z) \, n(z) \]

where

• \(z\) is the altitude,
• \(x_{\mathrm{M}}(z)\) is the volume mixing ratio of molecule M at altitude \(z\),
• \(n(z)\) is the air number density at altitude \(z\),
• \(n_{\mathrm{M}}(z)\) is the number density of molecule M at altitude \(z\).

If the dataset has a z_bounds coordinate, the integral is computed using the centered rectangle method, where the z coordinate corresponds to the rectangle centers.

If the dataset does not have a z_bounds coordinate, the integration is performed using the trapezoidal rule.

mass_density_at_sea_level: t.Dict[str, pint.Quantity] property

Compute mass density at sea level.

Returns:

Type	Description
t.Dict[str, pint.Quantity]	A mapping of molecule and mass density at sea level.

molecules: t.List[str] property

Return list of molecules.

number_density_at_sea_level: t.Dict[str, pint.Quantity] property

Compute number density at sea level.

Returns:

Type	Description
t.Dict[str, pint.Quantity]	A mapping of molecule and number density at sea level.

volume_fraction: xr.DataArray property

Extract volume fraction and tabulate as a function of (m, z).

Returns:

Type	Description
xr.DataArray	Volume fraction.

volume_fraction_at_sea_level: t.Dict[str, pint.Quantity] property

Compute volume fraction at sea level.

Returns:

Type	Description
t.Dict[str, pint.Quantity]	A mapping of molecule and volume mixing fraction at sea level.

rescale(factors, check_volume_fraction_sum=False)

Rescale molecules concentration in atmospheric profile.

Parameters:

Name	Type	Description	Default
factors	t.MutableMapping[str, float]	A mapping of molecule and scaling factor.	required
check_volume_fraction_sum	bool	if True, check that volume fraction sums are never larger than one.	False

Raises:

Type	Description
ValueError	if check_volume_fraction_sum is True and the dataset is not valid.

Returns:

Type	Description
xr.Dataset	Rescaled dataset (new object).

Source code in src/joseki/accessor.py

def rescale(
    self,
    factors: t.MutableMapping[str, float],
    check_volume_fraction_sum: bool = False
) -> xr.Dataset:
    """Rescale molecules concentration in atmospheric profile.

    Args:
        factors: A mapping of molecule and scaling factor.
        check_volume_fraction_sum: if True, check that volume fraction sums
            are never larger than one.
    Raises:
        ValueError: if check_volume_fraction_sum is `True` and the 
            dataset is not valid.

    Returns:
        Rescaled dataset (new object).
    """
    ds = self._obj

    # update volume fraction
    x_new = {}
    for m in factors:
        with xr.set_options(keep_attrs=True):
            x_new[f"x_{m}"] = ds[f"x_{m}"] * factors[m]

    ds = ds.assign(x_new)

    # validate rescaled dataset
    try:
        ds.joseki.validate(check_volume_fraction_sum=check_volume_fraction_sum)
    except ValueError as e:
        raise ValueError("Cannot rescale") from e

    # update history attribute
    now = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
    for m in factors.keys():
        ds.attrs["history"] += (
            f"\n{now} - rescaled {m}'s volume mixing ratio using a scaling "
            f"factor of {factors[m]:.3f} - joseki, version {__version__}"
        )

    return ds

scaling_factors(target)

Compute scaling factor(s) to reach specific target amount(s).

Parameters:

Name	Type	Description	Default
target	t.MutableMapping[str, pint.Quantity]	Mapping of molecule and target amount.	required

Raises:

Type	Description
ValueError	If a target amount has dimensions that are not supported.

Returns:

Type	Description
t.MutableMapping[str, float]	Mapping of molecule and scaling factors.

Notes

For each molecule in the target mapping, the target amount is interpreted, depending on its dimensions (indicated in square brackets), as:

• a column number density [length^-2],
• a column mass density [mass * length^-2],
• a number densitx at sea level [length^-3],
• a mass density at sea level [mass * length^-3],
• a volume mixing fraction at sea level [dimensionless]

The scaling factor is then evaluated as the ratio of the target amount with the original amount, for each molecule.

See Also

rescale

Source code in src/joseki/accessor.py

def scaling_factors(
    self, target: t.MutableMapping[str, pint.Quantity]
) -> t.MutableMapping[str, float]:
    """Compute scaling factor(s) to reach specific target amount(s).

    Args:
        target: Mapping of molecule and target amount.

    Raises:
        ValueError: If a target amount has dimensions that are not supported.

    Returns:
        Mapping of molecule and scaling factors.

    Notes:
        For each molecule in the ``target`` mapping, the target amount is
        interpreted, depending on its dimensions (indicated in square 
        brackets), as:

        * a column number density [`length^-2`],
        * a column mass density [`mass * length^-2`],
        * a number densitx at sea level [`length^-3`],
        * a mass density at sea level [`mass * length^-3`],
        * a volume mixing fraction at sea level [`dimensionless`]

        The scaling factor is then evaluated as the ratio of the target amount
        with the original amount, for each molecule.

    See Also:
        `rescale`
    """
    compute_initial_amount = {
        "[length]^-2": self.column_number_density,
        "[mass] * [length]^-2": self.column_mass_density,
        "[length]^-3": self.number_density_at_sea_level,
        "[mass] * [length]^-3": self.mass_density_at_sea_level,
        "": self.volume_fraction_at_sea_level,
    }
    factors = {}
    for m, target_amount in target.items():
        initial_amount = None
        for dim in compute_initial_amount.keys():
            if target_amount.check(dim):
                initial_amount = compute_initial_amount[dim][m]
        if initial_amount is None:
            raise ValueError
        factors[m] = _scaling_factor(
            initial_amount=initial_amount, target_amount=target_amount
        )
    return factors

validate(check_volume_fraction_sum=False, ret_true_if_valid=False)

Validate atmosphere thermophysical profile dataset schema.

Returns:

Type	Description
bool	True if the dataset complies with the schema, else False.

Source code in src/joseki/accessor.py

def validate(
    self,
    check_volume_fraction_sum: bool = False,
    ret_true_if_valid: bool = False,
) -> bool:
    """Validate atmosphere thermophysical profile dataset schema.

    Returns:
        `True` if the dataset complies with the schema, else `False`.
    """
    return schema.validate(
        ds=self._obj,
        check_volume_fraction_sum=check_volume_fraction_sum,
        ret_true_if_valid=ret_true_if_valid,
    )

molecular_mass(m)

Return the average molecular mass of a molecule.

Parameters:

Name	Type	Description	Default
m	str	Molecule formula.	required

Returns:

Type	Description
pint.Quantity	Average molecular mass.

Source code in src/joseki/accessor.py

def molecular_mass(m: str) -> pint.Quantity:
    """Return the average molecular mass of a molecule.

    Args:
        m: Molecule formula.

    Returns:
        Average molecular mass.
    """
    return MM[m] * ureg("dalton")

Data

Raw data files.

Profiles

Factory

Profile factory module.

ProfileFactory

Profile factory class.

Source code in src/joseki/profiles/factory.py

@define
class ProfileFactory:
    """
    Profile factory class.
    """

    """Profile registry."""
    registry: t.Dict[str, Profile] = field(factory=dict)

    def register(
        self,
        identifier: str,
    ) -> t.Callable:
        """
        Register a profile class.

        Args:
            identifier: Profile identifier.

        Returns:
            Decorator function.
        """

        def inner_wrapper(wrapped_class: Profile) -> t.Callable:
            logger.info("Registering profile %s", identifier)
            if identifier in self.registry:
                logger.warning(  # pragma: no cover
                    "Profile %s already exists. Will replace it",
                    identifier,
                )
            self.registry[identifier] = wrapped_class
            return wrapped_class

        return inner_wrapper

    def create(self, identifier: str, **kwargs) -> Profile:
        """
        Create a profile instance.

        Args:
            identifier: Profile identifier.

        Returns:
            Profile instance.
        """
        if identifier not in self.registry:
            logger.fatal("Profile %s does not exist in the registry", identifier)
            raise ValueError(f"Profile {identifier} does not exist in the registry")

        logger.debug("Creating profile %s", identifier)
        profile_cls = self.registry[identifier]
        profile = profile_cls(**kwargs)
        return profile

create(identifier, **kwargs)

Create a profile instance.

Parameters:

Name	Type	Description	Default
identifier	str	Profile identifier.	required

Returns:

Type	Description
Profile	Profile instance.

Source code in src/joseki/profiles/factory.py

def create(self, identifier: str, **kwargs) -> Profile:
    """
    Create a profile instance.

    Args:
        identifier: Profile identifier.

    Returns:
        Profile instance.
    """
    if identifier not in self.registry:
        logger.fatal("Profile %s does not exist in the registry", identifier)
        raise ValueError(f"Profile {identifier} does not exist in the registry")

    logger.debug("Creating profile %s", identifier)
    profile_cls = self.registry[identifier]
    profile = profile_cls(**kwargs)
    return profile

register(identifier)

Register a profile class.

Parameters:

Name	Type	Description	Default
identifier	str	Profile identifier.	required

Returns:

Type	Description
t.Callable	Decorator function.

Source code in src/joseki/profiles/factory.py

def register(
    self,
    identifier: str,
) -> t.Callable:
    """
    Register a profile class.

    Args:
        identifier: Profile identifier.

    Returns:
        Decorator function.
    """

    def inner_wrapper(wrapped_class: Profile) -> t.Callable:
        logger.info("Registering profile %s", identifier)
        if identifier in self.registry:
            logger.warning(  # pragma: no cover
                "Profile %s already exists. Will replace it",
                identifier,
            )
        self.registry[identifier] = wrapped_class
        return wrapped_class

    return inner_wrapper

Core

Core module for atmosphere thermophysical profiles.

The Profile abstract class defines the interface for atmosphere thermophysical profiles. The interp function is used to interpolate an atmosphere thermophysical profile on new altitude values. The represent_profile_in_cells function is used to compute the cells representation of the atmosphere thermophysical profile.

Profile

Bases: ABC

Abstract class for atmosphere thermophysical profiles.

Source code in src/joseki/profiles/core.py

@define
class Profile(ABC):
    """
    Abstract class for atmosphere thermophysical profiles.
    """

    @abstractmethod
    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        """
        Return the profile as a dataset.

        Args:
            z: Altitude grid.
                If the profile can be evaluated at arbitrary altitudes, this
                parameter is passed to the evaluating method for that profile.
                If the profile is defined on a fixed altitude grid, this parameter
                is used to interpolate the profile on the specified altitude grid.
            interp_method: Interpolation method for each variable.
                If ``None``, the default interpolation method is used.
                Interpolation may be required if the profile is defined on a fixed
                altitude grid, and the altitude grid is not the same as the one
                used to define the profile.
                Interpolation may also not be required, e.g. if the profile is
                defined by analytical function(s) of the altitude variable.
            conserve_column: If `True`, ensure that column densities are conserved
                during interpolation.
            kwargs: Parameters passed to lower-level methods.

        Returns:
            Atmospheric profile.
        """
        pass  # pragma: no cover

to_dataset(z=None, interp_method=None, conserve_column=False, **kwargs) abstractmethod

Return the profile as a dataset.

Parameters:

Name	Type	Description	Default
z	t.Optional[pint.Quantity]	Altitude grid. If the profile can be evaluated at arbitrary altitudes, this parameter is passed to the evaluating method for that profile. If the profile is defined on a fixed altitude grid, this parameter is used to interpolate the profile on the specified altitude grid.	None
interp_method	t.Optional[t.Mapping[str, str]]	Interpolation method for each variable. If None, the default interpolation method is used. Interpolation may be required if the profile is defined on a fixed altitude grid, and the altitude grid is not the same as the one used to define the profile. Interpolation may also not be required, e.g. if the profile is defined by analytical function(s) of the altitude variable.	None
conserve_column	bool	If True, ensure that column densities are conserved during interpolation.	False
kwargs	t.Any	Parameters passed to lower-level methods.	{}

Returns:

Type	Description
xr.Dataset	Atmospheric profile.

Source code in src/joseki/profiles/core.py

@abstractmethod
def to_dataset(
    self,
    z: t.Optional[pint.Quantity] = None,
    interp_method: t.Optional[t.Mapping[str, str]] = None,
    conserve_column: bool = False,
    **kwargs: t.Any,
) -> xr.Dataset:
    """
    Return the profile as a dataset.

    Args:
        z: Altitude grid.
            If the profile can be evaluated at arbitrary altitudes, this
            parameter is passed to the evaluating method for that profile.
            If the profile is defined on a fixed altitude grid, this parameter
            is used to interpolate the profile on the specified altitude grid.
        interp_method: Interpolation method for each variable.
            If ``None``, the default interpolation method is used.
            Interpolation may be required if the profile is defined on a fixed
            altitude grid, and the altitude grid is not the same as the one
            used to define the profile.
            Interpolation may also not be required, e.g. if the profile is
            defined by analytical function(s) of the altitude variable.
        conserve_column: If `True`, ensure that column densities are conserved
            during interpolation.
        kwargs: Parameters passed to lower-level methods.

    Returns:
        Atmospheric profile.
    """
    pass  # pragma: no cover

interp(ds, z_new, method=DEFAULT_METHOD, conserve_column=False)

Interpolate atmospheric profile on new altitudes.

Parameters:

Name	Type	Description	Default
ds	xr.Dataset	Atmospheric profile to interpolate.	required
z_new	pint.Quantity	Altitudes values at which to interpolate the atmospheric profile.	required
method	t.Dict[str, str]	Mapping of variable and interpolation method. If a variable is not in the mapping, the linear interpolation is used. By default, linear interpolation is used for all variables.	DEFAULT_METHOD
conserve_column	bool	If True, ensure that column densities are conserved.	False

Mapping of variable and interpolation method.

If a variable is not in the mapping, the linear interpolation is used. By default, linear interpolation is used for all variables.

Returns:

Type	Description
xr.Dataset	Interpolated atmospheric profile.

Source code in src/joseki/profiles/core.py

def interp(
    ds: xr.Dataset,
    z_new: pint.Quantity,
    method: t.Dict[str, str] = DEFAULT_METHOD,
    conserve_column: bool = False,
) -> xr.Dataset:
    """Interpolate atmospheric profile on new altitudes.

    Args:
        ds: Atmospheric profile to interpolate.
        z_new: Altitudes values at which to interpolate the atmospheric profile.
        method: Mapping of variable and interpolation method.
            If a variable is not in the mapping, the linear interpolation is used.
            By default, linear interpolation is used for all variables.
        conserve_column: If True, ensure that column densities are conserved.

    method: Mapping of variable and interpolation method.
        If a variable is not in the mapping, the linear interpolation is used.
        By default, linear interpolation is used for all variables.

    Returns:
        Interpolated atmospheric profile.
    """
    z_units = ds.z.attrs["units"]
    z_new_values = z_new.m_as(z_units)

    coords = {"z": z_new.to(z_units)}

    # Interpolate pressure, temperature and density
    data_vars = {}
    for var in ["p", "t", "n"]:
        f = interpolate.interp1d(
            x=ds.z.values,
            y=ds[var].values,
            kind=method.get(var, method["default"]),
            bounds_error=True,
        )
        data_vars[var] = ureg.Quantity(f(z_new_values), ds[var].attrs["units"])

    # Interpolate volume fraction
    for m in ds.joseki.molecules:
        var = f"x_{m}"
        f = interpolate.interp1d(
            x=ds.z.values,
            y=ds[var].values,
            kind=method.get(var, method["default"]),
            bounds_error=True,
        )
        data_vars[var] = ureg.Quantity(f(z_new_values), ds[var].attrs["units"])

    # Attributes
    attrs = ds.attrs
    author = f"joseki, version {__version__}"
    attrs.update(
        {
            "history": f"{utcnow()} - dataset interpolation by {author}.",
        }
    )

    # Convert to dataset
    logger.debug("convert interpolated data to dataset")
    interpolated = schema.convert(
        data_vars=data_vars,
        coords=coords,
        attrs=attrs,
    )

    # Compute scaling factors to conserve column densities
    if conserve_column:
        return rescale_to_column(reference=ds, ds=interpolated)

    return interpolated

represent_profile_in_cells(ds, method=DEFAULT_METHOD, conserve_column=False)

Compute the cells representation of the atmosphere thermophysical profile.

Parameters:

Name	Type	Description	Default
ds	xr.Dataset	Initial atmospheric profile.	required
method	t.Dict[str, str]	Mapping of variable and interpolation method. If a variable is not in the mapping, the linear interpolation is used. By default, linear interpolation is used for all variables.	DEFAULT_METHOD
conserve_column	bool	If True, ensure that column densities are conserved.	False

Returns:

Type	Description
xr.Dataset	Cells representation of the atmosphere thermophysical profile.

Notes

Atmosphere cells (or layers) are defined by two consecutive altitude values. The layer's center altitude is defined as the arithmetic average of these two values. The pressure, temperature, number density and volume fraction fields are interpolated at these layer' center altitude values. In the new atmospheric profile, the z coordinate is updated with layer' center altitude values and a data variable z_bounds indicating the altitude bounds of each layer, is added. A copy of the dataset is returned, the original dataset is not modified.

Source code in src/joseki/profiles/core.py

def represent_profile_in_cells(
    ds: xr.Dataset,
    method: t.Dict[str, str] = DEFAULT_METHOD,
    conserve_column: bool = False,
) -> xr.Dataset:
    """Compute the cells representation of the atmosphere thermophysical profile.

    Args:
        ds: Initial atmospheric profile.
        method: Mapping of variable and interpolation method.
            If a variable is not in the mapping, the linear interpolation is used.
            By default, linear interpolation is used for all variables.
        conserve_column: If True, ensure that column densities are conserved.

    Returns:
        Cells representation of the atmosphere thermophysical profile.

    Notes:
        Atmosphere cells (or layers) are defined by two consecutive altitude 
        values. The layer's center altitude is defined as the arithmetic 
        average of these two values. The pressure, temperature, number density 
        and volume fraction fields are interpolated at these layer' center 
        altitude values. In the new atmospheric profile, the `z` coordinate 
        is updated with layer' center altitude values and a data variable 
        `z_bounds` indicating the altitude bounds of each layer, is added.
        A copy of the dataset is returned, the original dataset is not 
        modified.
    """
    # if the profile is already represented in cells, do nothing
    if ds.z.standard_name == "layer_center_altitude":
        return ds

    z_nodes = to_quantity(ds.z)
    z_centers = (z_nodes[:-1] + z_nodes[1:]) / 2.0

    logger.debug("O3 column amount: %s", ds.joseki.column_number_density["O3"].to("dobson_unit"))
    interpolated = interp(
        ds=ds,
        z_new=z_centers,
        method=method,
        conserve_column=False,  # we rescale later
    )
    interpolated.z.attrs = dict(
        standard_name="layer_center_altitude",
        long_name="layer center altitude",
        units="km",
    )
    z_bounds = np.stack([z_nodes[:-1], z_nodes[1:]])
    interpolated = interpolated.assign(
        z_bounds=(
            ("zbv", "z"),
            z_bounds.m_as("km"),
            dict(
                standard_name="cell_bound_altitude",
                long_name="cell bound altitude",
                units="km",
            ),
        )
    )
    logger.debug("O3 column amount: %s", interpolated.joseki.column_number_density["O3"].to("dobson_unit"))
    interpolated.attrs.update(
        history=interpolated.history + f"\n{utcnow()} "
        f"- represent profile on cells - joseki, version {__version__}"
    )

    if conserve_column:
        return rescale_to_column(reference=ds, ds=interpolated)
    return interpolated

rescale_to_column(reference, ds)

Rescale volume fraction to ensure that column densities are conserved.

Parameters:

Name	Type	Description	Default
reference	xr.Dataset	Reference profile.	required
ds	xr.Dataset	Profile to rescale.	required

Returns:

Type	Description
xr.Dataset	Rescaled profile.

Source code in src/joseki/profiles/core.py

def rescale_to_column(reference: xr.Dataset, ds: xr.Dataset) -> xr.Dataset:
    """Rescale volume fraction to ensure that column densities are conserved.

    Args:
        reference: Reference profile.
        ds: Profile to rescale.

    Returns:
        Rescaled profile.
    """
    desired = reference.joseki.column_number_density
    actual = ds.joseki.column_number_density
    rescaled = ds.joseki.rescale(
        factors= {
            m: (desired[m] / actual[m]).m_as("dimensionless")
            for m in reference.joseki.molecules
        }
    )
    return rescaled

Dataset schema

Dataset schema for atmosphere thermophysical profiles.

The dataset schema defines the variables, coordinates and attributes that are expected in a dataset representing an atmosphere thermophysical profile.

Schema

Dataset schema for atmosphere thermophysical profiles.

Source code in src/joseki/profiles/schema.py

@define(frozen=True)
class Schema:
    """Dataset schema for atmosphere thermophysical profiles."""

    data_vars = {
        "p": (["z"], npt.NDArray[np.float64], "Pa", "air_pressure"),
        "t": (["z"], npt.NDArray[np.float64], "K", "air_temperature"),
        "n": (["z"], npt.NDArray[np.float64], "m^-3", "air_number_density"),
    }

    coords = {
        "z": ("z", npt.NDArray[np.float64], "km", "altitude"),
    }

    attrs = {
        "Conventions": str,
        "title": str,
        "institution": str,
        "source": str,
        "history": str,
        "references": str,
        "url": str,
        "urldate": str,
    }

    def validate(
        self,
        ds: xr.Dataset,
        check_volume_fraction_sum: bool = False,
        ret_true_if_valid: bool = False,
    ) -> t.Optional[bool]:
        """Validate dataset.

        Args:
            ds: Dataset to validate.
            check_volume_fraction_sum: if True, check that volume fraction sums
                are never larger than one.
            ret_true_if_valid: make this method return True if the dataset is
                valid.

        Raises:
            ValueError: If the dataset does not match the schema.

        Returns:
            None or bool: If `ret_true_if_valid` is True, returns True if the 
                dataset is valid, otherwise returns None.
        """
        logger.debug("Validating dataset")

        logger.debug("Checking that all data variables are present")
        for var in self.data_vars:
            if var not in ds.data_vars:
                raise ValueError(f"missing data variable: {var}")  # pragma: no cover

        logger.debug("Checking that 'x_*' data variable(s) are present")
        if not any([name.startswith("x_") for name in ds.data_vars]):
            raise ValueError("missing data variable starting with x_")  # pragma: no cover

        logger.debug("Checking that all coordinates are present")
        for coord in self.coords:
            if coord not in ds.coords:
                raise ValueError(f"missing coordinate: {coord}")  # pragma: no cover

        logger.debug("Checking that all attributes are present")
        for attr in self.attrs:
            if attr not in ds.attrs:
                raise ValueError(f"missing attribute: {attr}")  # pragma: no cover

        logger.debug("Checking that data variables have the correct dimensions")
        for var, (dims, _, _, _) in self.data_vars.items():
            if set(ds[var].dims) != set(dims):
                raise ValueError(  # pragma: no cover
                    f"incorrect dimensions for {var}. Expected {dims}, "
                    f"got {ds[var].dims}"
                )

        logger.debug("Checking that coordinates have the correct dimensions")
        for coord, (dims, _, _, _) in self.coords.items():
            if set(ds[coord].dims) != set(dims):
                raise ValueError(  # pragma: no cover
                    f"incorrect dimensions for {coord}. Expected {dims}, "
                    f"got {ds[coord].dims}"
                )

        logger.debug("Checking that data variables have the correct units")
        for var, (_, _, units, _) in self.data_vars.items():
            if ds[var].units != units:
                raise ValueError(  # pragma: no cover
                    f"incorrect units for {var}. Expected {units}, "
                    f"got {ds[var].units}"
                )

        logger.debug("Checking that coordinates have the correct units")
        for coord, (_, _, units, _) in self.coords.items():
            if ds[coord].units != units:
                raise ValueError(  # pragma: no cover
                    f"incorrect units for {coord}. Expected {units}, "
                    f"got {ds[coord].units}"
                )

        logger.debug("Checking that attributes have the correct types")
        for attr, typ in self.attrs.items():
            if not isinstance(ds.attrs[attr], typ):
                raise ValueError(  # pragma: no cover
                    f"incorrect type for {attr}. Expected {typ}, "
                    f"got {type(ds.attrs[attr])}"
                )

        logger.debug("Checking that data variables have the correct standard names")
        for var, (_, _, _, standard_name) in self.data_vars.items():
            if ds[var].attrs["standard_name"] != standard_name:
                raise ValueError(  # pragma: no cover
                    f"incorrect standard name for {var}. Expected "
                    f"{standard_name}, got "
                    f"{ds[var].attrs['standard_name']}"
                )

        logger.debug(
            "Checking that all x_* data variables have the correct units and "
            "standard names"
        )
        for var in ds.data_vars:
            if var.startswith("x_"):
                m = var[2:]
                if ds[var].attrs["units"] != "dimensionless":
                    raise ValueError(  # pragma: no cover
                        f"incorrect units for {var}. Expected dimensionless, "
                        f"got {ds[var].attrs['units']}"
                    )
                if ds[var].attrs["standard_name"] != f"{m}_volume_fraction":
                    raise ValueError(  # pragma: no cover
                        f"incorrect standard name for {var}. Expected "
                        f"{m}_volume_fraction, got "
                        f"{ds[var].attrs['standard_name']}"
                    )

        if check_volume_fraction_sum:
            logger.debug(
                "Checking that volume fraction sums are never larger than one"
            )
            vfs = volume_fraction_sum(ds)
            if np.any(vfs.m > 1):
                raise ValueError(  # pragma: no cover
                    "The rescaling factors lead to a profile where the volume "
                    "fraction sum is larger than 1."
                )

        logger.info("Dataset is valid")

        if ret_true_if_valid:  # pragma: no cover
            return True

    def convert(
        self,
        data_vars: t.Mapping[str, pint.Quantity],
        coords: t.Mapping[str, pint.Quantity],
        attrs: t.Mapping[str, str],
    ) -> xr.Dataset:
        """Convert input to schema-compliant dataset.

        Args:
            data_vars: Mapping of data variable names to quantities.
            coords: Mapping of coordinate names to quantities.
            attrs: Mapping of attribute names to values.

        Returns:
            Dataset with schema-compliant data variables, coordinates, and
            attributes.
        """
        logger.debug("converting input to schema-compliant dataset")

        logger.debug("checking that all data variables are present")
        for var in self.data_vars:
            if var not in data_vars:
                raise ValueError(f"missing data variable: {var}")  # pragma: no cover

        logger.debug("checking that there is at least one x_ data variable")
        if not any([name.startswith("x_") for name in data_vars]):
            raise ValueError("missing data variable starting with x_")  # pragma: no cover

        logger.debug("checking that all coordinates are present")
        for coord in self.coords:
            if coord not in coords:
                raise ValueError(f"missing coordinate: {coord}")  # pragma: no cover

        logger.debug("checking that all attributes are present")
        for attr in self.attrs:
            if attr not in attrs:
                raise ValueError(f"missing attribute: {attr}")  # pragma: no cover

        logger.debug("converting data variables to xarray data array tuples")
        for var, (dims, _, units, standard_name) in self.data_vars.items():
            data_vars[var] = (
                dims,
                data_vars[var].m_as(units),
                {
                    "standard_name": standard_name,
                    "long_name": standard_name.replace("_", " "),
                    "units": units,
                },
            )

        logger.debug("converting x_ data variables")
        for var in data_vars:
            if var.startswith("x_"):
                m = var[2:]
                data_vars[var] = (
                    "z",
                    data_vars[var].m_as("dimensionless"),
                    {
                        "standard_name": f"{m}_volume_fraction",
                        "long_name": f"{m} volume fraction",
                        "units": "dimensionless",
                    },
                )

        logger.debug("converting coordinates")
        for attr, (_, _, units, standard_name) in self.coords.items():
            coords[attr] = (
                attr,
                coords[attr].m_as(units),
                {
                    "standard_name": standard_name,
                    "long_name": standard_name.replace("_", " "),
                    "units": units,
                },
            )

        logger.debug("checking that all attributes are present")
        for attr in self.attrs:
            if attr not in attrs:
                raise ValueError(f"missing attribute: {attr}")  # pragma: no cover

        logger.debug("creating dataset")
        return xr.Dataset(
            data_vars=data_vars,
            coords=coords,
            attrs=attrs,
        )

convert(data_vars, coords, attrs)

Convert input to schema-compliant dataset.

Parameters:

Name	Type	Description	Default
data_vars	t.Mapping[str, pint.Quantity]	Mapping of data variable names to quantities.	required
coords	t.Mapping[str, pint.Quantity]	Mapping of coordinate names to quantities.	required
attrs	t.Mapping[str, str]	Mapping of attribute names to values.	required

Returns:

Type	Description
xr.Dataset	Dataset with schema-compliant data variables, coordinates, and
xr.Dataset	attributes.

Source code in src/joseki/profiles/schema.py

def convert(
    self,
    data_vars: t.Mapping[str, pint.Quantity],
    coords: t.Mapping[str, pint.Quantity],
    attrs: t.Mapping[str, str],
) -> xr.Dataset:
    """Convert input to schema-compliant dataset.

    Args:
        data_vars: Mapping of data variable names to quantities.
        coords: Mapping of coordinate names to quantities.
        attrs: Mapping of attribute names to values.

    Returns:
        Dataset with schema-compliant data variables, coordinates, and
        attributes.
    """
    logger.debug("converting input to schema-compliant dataset")

    logger.debug("checking that all data variables are present")
    for var in self.data_vars:
        if var not in data_vars:
            raise ValueError(f"missing data variable: {var}")  # pragma: no cover

    logger.debug("checking that there is at least one x_ data variable")
    if not any([name.startswith("x_") for name in data_vars]):
        raise ValueError("missing data variable starting with x_")  # pragma: no cover

    logger.debug("checking that all coordinates are present")
    for coord in self.coords:
        if coord not in coords:
            raise ValueError(f"missing coordinate: {coord}")  # pragma: no cover

    logger.debug("checking that all attributes are present")
    for attr in self.attrs:
        if attr not in attrs:
            raise ValueError(f"missing attribute: {attr}")  # pragma: no cover

    logger.debug("converting data variables to xarray data array tuples")
    for var, (dims, _, units, standard_name) in self.data_vars.items():
        data_vars[var] = (
            dims,
            data_vars[var].m_as(units),
            {
                "standard_name": standard_name,
                "long_name": standard_name.replace("_", " "),
                "units": units,
            },
        )

    logger.debug("converting x_ data variables")
    for var in data_vars:
        if var.startswith("x_"):
            m = var[2:]
            data_vars[var] = (
                "z",
                data_vars[var].m_as("dimensionless"),
                {
                    "standard_name": f"{m}_volume_fraction",
                    "long_name": f"{m} volume fraction",
                    "units": "dimensionless",
                },
            )

    logger.debug("converting coordinates")
    for attr, (_, _, units, standard_name) in self.coords.items():
        coords[attr] = (
            attr,
            coords[attr].m_as(units),
            {
                "standard_name": standard_name,
                "long_name": standard_name.replace("_", " "),
                "units": units,
            },
        )

    logger.debug("checking that all attributes are present")
    for attr in self.attrs:
        if attr not in attrs:
            raise ValueError(f"missing attribute: {attr}")  # pragma: no cover

    logger.debug("creating dataset")
    return xr.Dataset(
        data_vars=data_vars,
        coords=coords,
        attrs=attrs,
    )

validate(ds, check_volume_fraction_sum=False, ret_true_if_valid=False)

Validate dataset.

Parameters:

Name	Type	Description	Default
ds	xr.Dataset	Dataset to validate.	required
check_volume_fraction_sum	bool	if True, check that volume fraction sums are never larger than one.	False
ret_true_if_valid	bool	make this method return True if the dataset is valid.	False

Raises:

Type	Description
ValueError	If the dataset does not match the schema.

Returns:

Type	Description
t.Optional[bool]	None or bool: If ret_true_if_valid is True, returns True if the dataset is valid, otherwise returns None.

Source code in src/joseki/profiles/schema.py

def validate(
    self,
    ds: xr.Dataset,
    check_volume_fraction_sum: bool = False,
    ret_true_if_valid: bool = False,
) -> t.Optional[bool]:
    """Validate dataset.

    Args:
        ds: Dataset to validate.
        check_volume_fraction_sum: if True, check that volume fraction sums
            are never larger than one.
        ret_true_if_valid: make this method return True if the dataset is
            valid.

    Raises:
        ValueError: If the dataset does not match the schema.

    Returns:
        None or bool: If `ret_true_if_valid` is True, returns True if the 
            dataset is valid, otherwise returns None.
    """
    logger.debug("Validating dataset")

    logger.debug("Checking that all data variables are present")
    for var in self.data_vars:
        if var not in ds.data_vars:
            raise ValueError(f"missing data variable: {var}")  # pragma: no cover

    logger.debug("Checking that 'x_*' data variable(s) are present")
    if not any([name.startswith("x_") for name in ds.data_vars]):
        raise ValueError("missing data variable starting with x_")  # pragma: no cover

    logger.debug("Checking that all coordinates are present")
    for coord in self.coords:
        if coord not in ds.coords:
            raise ValueError(f"missing coordinate: {coord}")  # pragma: no cover

    logger.debug("Checking that all attributes are present")
    for attr in self.attrs:
        if attr not in ds.attrs:
            raise ValueError(f"missing attribute: {attr}")  # pragma: no cover

    logger.debug("Checking that data variables have the correct dimensions")
    for var, (dims, _, _, _) in self.data_vars.items():
        if set(ds[var].dims) != set(dims):
            raise ValueError(  # pragma: no cover
                f"incorrect dimensions for {var}. Expected {dims}, "
                f"got {ds[var].dims}"
            )

    logger.debug("Checking that coordinates have the correct dimensions")
    for coord, (dims, _, _, _) in self.coords.items():
        if set(ds[coord].dims) != set(dims):
            raise ValueError(  # pragma: no cover
                f"incorrect dimensions for {coord}. Expected {dims}, "
                f"got {ds[coord].dims}"
            )

    logger.debug("Checking that data variables have the correct units")
    for var, (_, _, units, _) in self.data_vars.items():
        if ds[var].units != units:
            raise ValueError(  # pragma: no cover
                f"incorrect units for {var}. Expected {units}, "
                f"got {ds[var].units}"
            )

    logger.debug("Checking that coordinates have the correct units")
    for coord, (_, _, units, _) in self.coords.items():
        if ds[coord].units != units:
            raise ValueError(  # pragma: no cover
                f"incorrect units for {coord}. Expected {units}, "
                f"got {ds[coord].units}"
            )

    logger.debug("Checking that attributes have the correct types")
    for attr, typ in self.attrs.items():
        if not isinstance(ds.attrs[attr], typ):
            raise ValueError(  # pragma: no cover
                f"incorrect type for {attr}. Expected {typ}, "
                f"got {type(ds.attrs[attr])}"
            )

    logger.debug("Checking that data variables have the correct standard names")
    for var, (_, _, _, standard_name) in self.data_vars.items():
        if ds[var].attrs["standard_name"] != standard_name:
            raise ValueError(  # pragma: no cover
                f"incorrect standard name for {var}. Expected "
                f"{standard_name}, got "
                f"{ds[var].attrs['standard_name']}"
            )

    logger.debug(
        "Checking that all x_* data variables have the correct units and "
        "standard names"
    )
    for var in ds.data_vars:
        if var.startswith("x_"):
            m = var[2:]
            if ds[var].attrs["units"] != "dimensionless":
                raise ValueError(  # pragma: no cover
                    f"incorrect units for {var}. Expected dimensionless, "
                    f"got {ds[var].attrs['units']}"
                )
            if ds[var].attrs["standard_name"] != f"{m}_volume_fraction":
                raise ValueError(  # pragma: no cover
                    f"incorrect standard name for {var}. Expected "
                    f"{m}_volume_fraction, got "
                    f"{ds[var].attrs['standard_name']}"
                )

    if check_volume_fraction_sum:
        logger.debug(
            "Checking that volume fraction sums are never larger than one"
        )
        vfs = volume_fraction_sum(ds)
        if np.any(vfs.m > 1):
            raise ValueError(  # pragma: no cover
                "The rescaling factors lead to a profile where the volume "
                "fraction sum is larger than 1."
            )

    logger.info("Dataset is valid")

    if ret_true_if_valid:  # pragma: no cover
        return True

volume_fraction_sum(ds)

Compute the sum of volume mixing fractions.

Parameters:

Name	Type	Description	Default
ds	xr.Dataset	Dataset.	required

Returns:

Type	Description
pint.Quantity	The sum of volume fractions.

Source code in src/joseki/profiles/schema.py

def volume_fraction_sum(ds: xr.Dataset) -> pint.Quantity:
    """Compute the sum of volume mixing fractions.

    Args:
        ds: Dataset.

    Returns:
        The sum of volume fractions.
    """
    return (
        sum([ds[c] for c in ds.data_vars if c.startswith("x_")]).values
        * ureg.dimensionless
    )

AFGL (1986)

AFGL 1986 atmosphere's thermophysical profiles.

The profiles are generated from data files stored in joseki/data/afgl_1986. These data files correspond to tables 1a-f and 2a-d of the technical report Anderson+1986.

AFGL1986MidlatitudeSummer

Bases: Profile

AFGL 1986 midlatitude summer atmosphere thermophysical profile.

Source code in src/joseki/profiles/afgl_1986.py

@factory.register(identifier="afgl_1986-midlatitude_summer")
@define
class AFGL1986MidlatitudeSummer(Profile):
    """AFGL 1986 midlatitude summer atmosphere thermophysical profile."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        logger.debug(
            "creating AFGL 1986 midlatitude summer atmosphere thermophysical "
            "profile dataset."
        )
        return to_dataset(
            identifier=Identifier.MIDLATITUDE_SUMMER,
            z=z,
            interp_method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

AFGL1986MidlatitudeWinter

Bases: Profile

AFGL 1986 midlatitude winter atmosphere thermophysical profile.

Source code in src/joseki/profiles/afgl_1986.py

@factory.register(identifier="afgl_1986-midlatitude_winter")
@define
class AFGL1986MidlatitudeWinter(Profile):
    """AFGL 1986 midlatitude winter atmosphere thermophysical profile."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        logger.debug(
            "creating AFGL 1986 midlatitude winter atmosphere thermophysical "
            "profile dataset."
        )
        return to_dataset(
            identifier=Identifier.MIDLATITUDE_WINTER,
            z=z,
            interp_method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

AFGL1986SubarcticSummer

Bases: Profile

AFGL 1986 subarctic summer atmosphere thermophysical profile.

Source code in src/joseki/profiles/afgl_1986.py

@factory.register(identifier="afgl_1986-subarctic_summer")
@define
class AFGL1986SubarcticSummer(Profile):
    """AFGL 1986 subarctic summer atmosphere thermophysical profile."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        logger.debug(
            "creating AFGL 1986 subarctic summer atmosphere thermophysical "
            "profile dataset."
        )
        return to_dataset(
            identifier=Identifier.SUBARCTIC_SUMMER,
            z=z,
            interp_method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

AFGL1986SubarcticWinter

Bases: Profile

AFGL 1986 subarctic winter atmosphere thermophysical profile.

Source code in src/joseki/profiles/afgl_1986.py

@factory.register(identifier="afgl_1986-subarctic_winter")
@define
class AFGL1986SubarcticWinter(Profile):
    """AFGL 1986 subarctic winter atmosphere thermophysical profile."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        logger.debug(
            "creating AFGL 1986 subarctic winter atmosphere thermophysical "
            "profile dataset."
        )
        return to_dataset(
            identifier=Identifier.SUBARCTIC_WINTER,
            z=z,
            interp_method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

AFGL1986Tropical

Bases: Profile

AFGL 1986 tropical atmosphere thermophysical profile.

Source code in src/joseki/profiles/afgl_1986.py

@factory.register(identifier="afgl_1986-tropical")
@define
class AFGL1986Tropical(Profile):
    """AFGL 1986 tropical atmosphere thermophysical profile."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        logger.debug(
            "creating AFGL 1986 tropical atmosphere thermophysical profile dataset."
        )
        return to_dataset(
            identifier=Identifier.TROPICAL,
            z=z,
            interp_method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

AFGL1986USStandard

Bases: Profile

AFGL 1986 US Standard atmosphere thermophysical profile.

Source code in src/joseki/profiles/afgl_1986.py

@factory.register(identifier="afgl_1986-us_standard")
@define
class AFGL1986USStandard(Profile):
    """AFGL 1986 US Standard atmosphere thermophysical profile."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        logger.debug(
            "creating AFGL 1986 US Standard atmosphere thermophysical profile dataset."
        )
        return to_dataset(
            identifier=Identifier.US_STANDARD,
            z=z,
            interp_method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

Identifier

Bases: enum.Enum

AFGL 1986 atmospheric profile identifier enumeration.

Source code in src/joseki/profiles/afgl_1986.py

class Identifier(enum.Enum):
    """AFGL 1986 atmospheric profile identifier enumeration."""

    TROPICAL = "tropical"
    MIDLATITUDE_SUMMER = "midlatitude_summer"
    MIDLATITUDE_WINTER = "midlatitude_winter"
    SUBARCTIC_SUMMER = "subarctic_summer"
    SUBARCTIC_WINTER = "subarctic_winter"
    US_STANDARD = "us_standard"

dataframe_to_dataset(df, identifier, additional_molecules=True)

Convert the output of the parse method to a xarray.Dataset.

Parameters:

Name	Type	Description	Default
df	pd.DataFrame	Atmospheric profile data.	required
identifier	Identifier	Atmospheric profile identifier.	required
additional_molecules	bool	If True, include molecules 8-28 as numbered in Anderson+1986. Else, discard molecules 8-28.	True

Returns:

Type	Description
xr.Dataset	Atmospheric profile dataset.

Notes

Use the z column of the output pandas.DataFrame of read_raw_data as data coordinate and all other columns as data variables. All data variables and coordinates of the returned xarray.Dataset are associated metadata (standard name, long name and units). Raw data units are documented in the technical report AFGL Atmospheric Constituent Profiles (0-120 km), Anderson et al., 1986 Anderson+1986. dataset attributes are added.

Source code in src/joseki/profiles/afgl_1986.py

def dataframe_to_dataset(
    df: pd.DataFrame,
    identifier: Identifier,
    additional_molecules: bool = True,
) -> xr.Dataset:
    """Convert the output of the `parse` method to a `xarray.Dataset`.

    Args:
        df: Atmospheric profile data.
        identifier: Atmospheric profile identifier.
        additional_molecules: If ``True``, include molecules 8-28 as numbered 
            in [Anderson+1986](bibliography.md#Anderson+1986).
            Else, discard molecules 8-28.

    Returns:
        Atmospheric profile dataset.

    Notes:
        Use the ``z`` column of the output pandas.DataFrame of read_raw_data
        as data coordinate and all other columns as data variables.
        All data variables and coordinates of the returned xarray.Dataset are
        associated metadata (standard name, long name and units).
        Raw data units are documented in the technical report *AFGL Atmospheric
        Constituent Profiles (0-120 km)*, Anderson et al., 1986
        [Anderson+1986](bibliography.md#Anderson+1986).
        dataset attributes are added.
    """
    # list molecules
    # molecules labels correspond to column with upper case first letter in
    # raw data DataFrames
    molecules = []
    for column in df.columns:
        if column[0].isupper():
            molecules.append(column)

    if additional_molecules:
        pass
    else:
        molecules = molecules[:7]

    # coordinates
    coords = {"z": ureg.Quantity(df.z.values, "km")}

    # data variables
    data_vars = {}
    data_vars["p"] = ureg.Quantity(df.p.values, "millibar").to("Pa")
    data_vars["t"] = ureg.Quantity(df.t.values, "K")
    data_vars["n"] = ureg.Quantity(df.n.values, "cm^-3").to("m^-3")

    for s in molecules:
        data_vars[f"x_{s}"] = (
            df[s].values * ureg.ppm
        )  # raw data volume fraction are given in ppmv

    # attributes
    pretty_identifier = f"AFGL (1986) {identifier.value.replace('_', '-')}"
    pretty_title = f"{pretty_identifier} atmosphere thernmophysical profile"

    attrs = {
        "Conventions": "CF-1.10",
        "title": pretty_title,
        "institution": INSTITUION,
        "source": SOURCE,
        "history": history(),
        "references": REFERENCE,
        "url": URL,
        "urldate": URLDATE,
    }

    return schema.convert(
        data_vars=data_vars,
        coords=coords,
        attrs=attrs,
    )

get_dataset(identifier, additional_molecules=True)

Read data files for a given atmospheric profile.

Parameters:

Name	Type	Description	Default
identifier	Identifier	Atmospheric profile identifier. See Identifier for possible values.	required
additional_molecules	bool	If True, include molecules 8-28 as numbered in Anderson+1986. Else, discard molecules 8-28.	True

Returns:

Type	Description
xr.Dataset	Atmospheric profile dataset.

Notes

Chain calls to parse and dataframe_to_dataset.

Source code in src/joseki/profiles/afgl_1986.py

def get_dataset(
    identifier: Identifier,
    additional_molecules: bool = True,
) -> xr.Dataset:
    """Read data files for a given atmospheric profile.

    Args:
        identifier: Atmospheric profile identifier.
            See 
            [`Identifier`](reference.md#src.joseki.profiles.afgl_1986.Identifier) 
            for possible values.
        additional_molecules: If ``True``, include molecules 8-28 as numbered in
            [Anderson+1986](bibliography.md#Anderson+1986).
            Else, discard molecules 8-28.

    Returns:
        Atmospheric profile dataset.

    Notes:
        Chain calls to 
        [`parse`](reference.md#src.joseki.profiles.afgl_1986.parse) and
        [`dataframe_to_dataset`](reference.md#src.joseki.profiles.afgl_1986.dataframe_to_dataset).

    """
    df = parse(identifier=identifier)
    return dataframe_to_dataset(
        df=df,
        identifier=identifier,
        additional_molecules=additional_molecules,
    )

parse(identifier)

Parse table data files for a given atmospheric profile.

Parameters:

Name	Type	Description	Default
identifier	Identifier	Atmospheric profile identifier.	required

Returns:

Type	Description
pd.DataFrame	Atmospheric profile dataset.

Notes

Read the relevant raw data files corresponding to the atmospheric profile. These raw data files correspond to tables 1 and 2 from the technical report AFGL Atmospheric Constituent Profiles (0-120 km), Anderson et al., 1986. Each atmospheric profile has 5 tables, i.e. 5 raw data files, associated to it. Only the first of these tables is specific to each atmospheric profile. All 5 raw data files are read into pandas.DataFrame objects and then concatenated after dropping the duplicate columns.

Source code in src/joseki/profiles/afgl_1986.py

def parse(identifier: Identifier) -> pd.DataFrame:
    """Parse table data files for a given atmospheric profile.

    Args:
        identifier: Atmospheric profile identifier.

    Returns:
        Atmospheric profile dataset.

    Notes:
        Read the relevant raw data files corresponding to the atmospheric profile.
        These raw data files correspond to tables 1 and 2 from the
        technical report [*AFGL Atmospheric Constituent Profiles (0-120 km)*,
        Anderson et al., 1986](bibliography.md#Anderson+1986).
        Each atmospheric profile has 5 tables, i.e. 5 raw data files, associated
        to it.
        Only the first of these tables is specific to each atmospheric profile.
        All 5 raw data files are read into `pandas.DataFrame` objects and
        then concatenated after dropping the duplicate columns.
    """
    package = "joseki.data.afgl_1986"
    files = DATA_FILES[identifier]
    dataframes = []
    for file in files:
        csvfile = importlib_resources.files(package).joinpath(file)
        df = pd.read_csv(csvfile)
        dataframes.append(df)
    dataframes[1] = dataframes[1].drop(["H2O", "O3", "N2O", "CO", "CH4"], axis=1)
    for i in range(1, 5):
        dataframes[i] = dataframes[i].drop("z", axis=1)

    return pd.concat(dataframes, axis=1)

to_dataset(identifier, z=None, interp_method=None, conserve_column=False, **kwargs)

Helper Profile.to_dataset() method.

Parameters:

Name	Type	Description	Default
identifier	Identifier	AFGL 1986 atmosphere thermophysical profile identifier. See Identifier for possible values.	required
z	t.Optional[pint.Quantity]	New level altitudes. If None, return the original dataset Else, interpolate the dataset to the new level altitudes. Default is None.	None
interp_method	t.Mapping[str, str]	Interpolation method for each data variable. Default is None.	None
conserve_column	bool	If True, ensure that column densities are conserved during interpolation.	False
kwargs	t.Any	Additional arguments passed to get_dataset.	{}

Returns:

Type	Description
xr.Dataset	Atmosphere thermophysical profile dataset.

Source code in src/joseki/profiles/afgl_1986.py

def to_dataset(
    identifier: Identifier,
    z: t.Optional[pint.Quantity] = None,
    interp_method: t.Mapping[str, str] = None,
    conserve_column: bool = False,
    **kwargs: t.Any,
) -> xr.Dataset:
    """
    Helper Profile.to_dataset() method.

    Args:
        identifier: AFGL 1986 atmosphere thermophysical profile identifier.
            See 
            [`Identifier`](reference.md#src.joseki.profiles.afgl_1986.Identifier) 
            for possible values.
        z: New level altitudes.
            If ``None``, return the original dataset
            Else, interpolate the dataset to the new level altitudes.
            Default is ``None``.
        interp_method: Interpolation method for each data variable. Default is 
            ``None``.
        conserve_column: If `True`, ensure that column densities are conserved
            during interpolation.
        kwargs: Additional arguments passed to 
            [`get_dataset`](reference.md#src.joseki.profiles.afgl_1986.get_dataset).

    Returns:
        Atmosphere thermophysical profile dataset.
    """
    # Get additional_molecules from kwargs
    additional_molecules = kwargs.get("additional_molecules", True)

    # kwargs different than 'additional_molecules' are ignored
    if len([x for x in kwargs.keys() if x != "additional_molecules"]) > 0:
        logger.warning(
            "Ignoring kwargs different than 'additional_molecules'. "
            "(got %s)"
            "Use 'additional_molecules' to include molecules 8-28 "
            "as numbered in Anderson et al. (1986).",
            kwargs,
        )

    # Get the original dataset
    ds = get_dataset(
        identifier=identifier,
        additional_molecules=additional_molecules,
    )

    # Interpolate if necessary
    if z is not None:
        method = interp_method if interp_method is not None else DEFAULT_METHOD
        ds = interp(
            ds=ds,
            z_new=z,
            method=method,
            conserve_column=conserve_column,
        )
        return ds
    else:
        return ds

MIPAS (2007)

MIPAS atmosphere thermophysical profiles.

Remedios et al. (2007) define a set of 5 "standard atmospheres" representing the atmosphere at different latitudes and seasons or times of day:

• midlatitude day
• midlatitude night
• polar winter
• polar summer
• tropical

MIPAS standard atmospheres were intended to provide an updated set of pro- files for characteristic atmospheric states such as the AFGL Atmospheric constituent profiles.

Identifier

Bases: enum.Enum

MIPAS atmosphere thermophysical profile identifier enumeration.

Source code in src/joseki/profiles/mipas_2007.py

class Identifier(enum.Enum):
    """MIPAS atmosphere thermophysical profile identifier enumeration."""

    MIDLATITUDE_DAY = "midlatitude_day"
    MIDLATITUDE_NIGHT = "midlatitude_night"
    POLAR_WINTER = "polar_winter"
    POLAR_SUMMER = "polar_summer"
    TROPICAL = "tropical"

MIPASMidlatitudeDay

Bases: Profile

MIPAS midlatitude day reference atmosphere.

Source code in src/joseki/profiles/mipas_2007.py

@factory.register("mipas_2007-midlatitude_day")
@define
class MIPASMidlatitudeDay(Profile):
    """MIPAS midlatitude day reference atmosphere."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        return to_dataset(
            identifier=Identifier.MIDLATITUDE_DAY,
            z=z,
            method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

MIPASMidlatitudeNight

Bases: Profile

MIPAS Midlatitude night reference atmosphere.

Source code in src/joseki/profiles/mipas_2007.py

@factory.register("mipas_2007-midlatitude_night")
@define
class MIPASMidlatitudeNight(Profile):
    """MIPAS Midlatitude night reference atmosphere."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        return to_dataset(
            identifier=Identifier.MIDLATITUDE_NIGHT,
            z=z,
            method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

MIPASPolarSummer

Bases: Profile

MIPAS Polar summer reference atmosphere.

Source code in src/joseki/profiles/mipas_2007.py

@factory.register("mipas_2007-polar_summer")
@define
class MIPASPolarSummer(Profile):
    """MIPAS Polar summer reference atmosphere."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        return to_dataset(
            identifier=Identifier.POLAR_SUMMER,
            z=z,
            method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

MIPASPolarWinter

Bases: Profile

MIPAS Polar winter reference atmosphere.

Source code in src/joseki/profiles/mipas_2007.py

@factory.register("mipas_2007-polar_winter")
@define
class MIPASPolarWinter(Profile):
    """MIPAS Polar winter reference atmosphere."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        return to_dataset(
            identifier=Identifier.POLAR_WINTER,
            z=z,
            method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

MIPASTropical

Bases: Profile

MIPAS Tropical reference atmosphere.

Source code in src/joseki/profiles/mipas_2007.py

@factory.register("mipas_2007-tropical")
@define
class MIPASTropical(Profile):
    """MIPAS Tropical reference atmosphere."""

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        return to_dataset(
            identifier=Identifier.TROPICAL,
            z=z,
            method=interp_method,
            conserve_column=conserve_column,
            **kwargs,
        )

get_dataset(identifier)

Read MIPAS reference atmosphere data files into an xarray.Dataset.

Parameters:

Name	Type	Description	Default
identifier	Identifier	Atmospheric profile identifier. See Identifier for possible values.	required

Returns:

Type	Description
xr.Dataset	Atmospheric profile.

Source code in src/joseki/profiles/mipas_2007.py

def get_dataset(identifier: Identifier) -> xr.Dataset:
    """Read MIPAS reference atmosphere data files into an xarray.Dataset.

    Args:
        identifier: Atmospheric profile identifier.
            See 
            [`Identifier`](reference.md#src.joseki.profiles.mipas_2007.Identifier) 
            for possible values.

    Returns:
        Atmospheric profile.
    """
    content = read_file_content(identifier=identifier)
    quantities = parse_content(content.splitlines())

    # Coordinates
    coords = {"z": quantities.pop("z")}

    # Data variables
    data_vars = {}
    p = quantities.pop("p")
    data_vars["p"] = p
    t = quantities.pop("t")
    data_vars["t"] = t
    n = p / (K * t)  # perfect gas equation
    data_vars["n"] = n
    data_vars.update(quantities)

    logger.debug("data variables: %s", data_vars.keys())

    # Attributes
    pretty_id = identifier.value.replace("_", " ")
    pretty_title = f"MIPAS {pretty_id} Reference Atmosphere"
    attrs = {
        "Conventions": "CF-1.10",
        "history": history(),
        "title": pretty_title,
        "source": SOURCE,
        "institution": INSTITUTION,
        "references": REFERENCE,
        "url": URL,
        "urldate": URL_DATE,
    }

    # Dataset
    ds = schema.convert(
        data_vars=data_vars,
        coords=coords,
        attrs=attrs,
    )

    return ds

parse_content(lines)

Parse lines.

Source code in src/joseki/profiles/mipas_2007.py

def parse_content(lines: t.List[str]) -> t.Dict[str, pint.Quantity]:
    """Parse lines."""
    logger.debug("Parsing file content")
    iterator = iter(lines)
    line = next(iterator)

    quantities: t.Dict[str, pint.Quantity] = {}

    def _add_to_quantities(quantity: pint.Quantity, name: str) -> None:
        if name not in ["z", "p", "t", "n"]:
            name = f"x_{name}"

        if quantity.check(""):
            quantities[name] = quantity.to("dimensionless")
        else:
            quantities[name] = quantity

    var: str = ""
    units: str = ""
    values: t.List[str] = []
    while line != "*END":
        if line.startswith("!"):
            pass  # this is a comment, ignore the line
        elif line.startswith("*"):
            # convert previously read values (if any) and units to quantity
            if len(values) > 0:
                quantity = ureg.Quantity(
                    np.array(values, dtype=float),
                    units,
                )
                _add_to_quantities(quantity=quantity, name=var)

            # this is a variable line, parse variable name and units
            var, units = parse_var_line(line)

            # following lines are the variables values so prepare a variable
            # to store the values
            values = []
        else:
            if "!" in line:
                # this the line with the number of profile levels, ignore it
                pass
            else:
                # this line contains variable values
                values += parse_values_line(line)
        line = next(iterator)

    # include last array of values before the '*END' line
    quantity = ureg.Quantity(np.array(values, dtype=float), units)
    _add_to_quantities(quantity=quantity, name=var)

    return quantities

parse_units(s)

Parse units.

Source code in src/joseki/profiles/mipas_2007.py

def parse_units(s: str) -> str:
    """Parse units."""
    if s.startswith("[") and s.endswith("]"):
        units = s[1:-1]
        if units == "mb":
            return "millibar"
        else:
            return units
    else:
        raise ValueError(f"Cannot parse units '{s}'")

parse_values_line(s)

Parse a line with numeric values.

Source code in src/joseki/profiles/mipas_2007.py

def parse_values_line(s: str) -> t.List[str]:
    """Parse a line with numeric values."""
    if "," in s:  # delimiter is comma and whitespace combined
        s_strip = s.strip()
        if s_strip[-1] == ",":
            s_strip = s_strip[:-1]
        return [x.strip() for x in s_strip.split(",")]
    else:  # delimiter is whitespace
        return s.split()

parse_var_line(s)

Parse a line with the declaration of a variable and its units.

Source code in src/joseki/profiles/mipas_2007.py

def parse_var_line(s: str) -> t.Tuple[str, str]:
    """Parse a line with the declaration of a variable and its units."""
    parts = s[1:].strip().split()
    if len(parts) == 2:
        var_name, units_s = parts
    elif len(parts) == 3:
        var_name, _, units_s = parts
    else:
        raise ValueError(f"Invalid line format: {s}")
    var = parse_var_name(var_name)
    units = parse_units(units_s)
    return var, units

parse_var_name(n)

Parse variable name.

Source code in src/joseki/profiles/mipas_2007.py

def parse_var_name(n: str) -> str:
    """Parse variable name."""
    translate = {"HGT": "z", "PRE": "p", "TEM": "t"}
    if n in translate.keys():
        return translate[n]
    else:
        return to_chemical_formula(n)

read_file_content(identifier)

Read data file content.

Parameters:

Name	Type	Description	Default
identifier	Identifier	Atmospheric profile identifier. See Identifier for possible values.	required

Returns:

Type	Description
str	file content, URL, URL date.

Source code in src/joseki/profiles/mipas_2007.py

def read_file_content(identifier: Identifier) -> str:
    """
    Read data file content.

    Args:
        identifier: Atmospheric profile identifier.
            See 
            [`Identifier`](reference.md#src.joseki.profiles.mipas_2007.Identifier) 
            for possible values.

    Returns:
        file content, URL, URL date.
    """
    package = "joseki.data.mipas_2007"
    file = f"{identifier.value}.atm"
    logger.debug(f"Reading file {file}")
    return importlib_resources.files(package).joinpath(file).read_text()

to_chemical_formula(name)

Convert to chemical formula.

Parameters:

Name	Type	Description	Default
name	str	Molecule name.	required

Returns:

Type	Description
str	Molecule formula.

Notes

If molecule name is unknown, returns name unchanged.

Source code in src/joseki/profiles/mipas_2007.py

def to_chemical_formula(name: str) -> str:
    """Convert to chemical formula.

    Args:
        name: Molecule name.

    Returns:
        Molecule formula.

    Notes:
        If molecule name is unknown, returns name unchanged.
    """
    try:
        return translate_cfc(name)
    except ValueError:
        return name

to_dataset(identifier, z=None, method=None, conserve_column=False, **kwargs)

Helper for Profile.to_dataset method

Source code in src/joseki/profiles/mipas_2007.py

def to_dataset(
    identifier: Identifier,
    z: t.Optional[pint.Quantity] = None,
    method: t.Optional[t.Mapping[str, str]] = None,
    conserve_column: bool = False,
    **kwargs: t.Any,
) -> xr.Dataset:
    """Helper for `Profile.to_dataset` method"""
    # no kwargs are expected
    if len(kwargs) > 0:  # pragma: no cover
        logger.warning("Unexpected keyword arguments: %s", kwargs)

    # get original MIPAS midlatitude day reference atmosphere
    logger.debug("Get original MIPAS midlatitude day reference atmosphere")
    ds = get_dataset(identifier=identifier)

    # Interpolate to new vertical grid if necessary
    if z is not None:
        method = DEFAULT_METHOD if method is None else method
        ds = interp(
            ds=ds,
            z_new=z,
            method=method,
            conserve_column=conserve_column,
        )
        return ds
    else:
        return ds

translate_cfc(name)

Convert chlorofulorocarbon name to corresponding chemical formula.

Parameters:

Name	Type	Description	Default
name	str	Chlorofulorocarbon name.	required

Returns:

Type	Description
str	Chlorofulorocarbon chemical formula.

Raises:

Type	Description
ValueError	If the name does not match a known chlorofulorocarbon.

Source code in src/joseki/profiles/mipas_2007.py

def translate_cfc(name: str) -> str:
    """Convert chlorofulorocarbon name to corresponding chemical formula.

    Args:
        name: Chlorofulorocarbon name.

    Returns:
        Chlorofulorocarbon chemical formula.

    Raises:
        ValueError: If the name does not match a known chlorofulorocarbon.
    """
    for formula, names in CFC_FORMULAE.items():
        if name in names:
            return formula
    raise ValueError("Unknown chlorofulorocarbon {name}")

US Standard Atmosphere (1976)

Module to compute the U.S. Standard Atmosphere 1976.

The U.S. Standard Atmosphere 1976 is a Earth atmosphere thermophysical model described in the technical report NOAA+1976.

USSA1976

Bases: Profile

Class to compute the U.S. Standard Atmosphere 1976.

The U.S. Standard Atmosphere 1976 is a Earth atmosphere thermophysical model described in the technical report NOAA+1976.

Source code in src/joseki/profiles/ussa_1976.py

@factory.register(identifier="ussa_1976")
@define
class USSA1976(Profile):
    """
    Class to compute the U.S. Standard Atmosphere 1976.

    The U.S. Standard Atmosphere 1976 is a Earth atmosphere thermophysical model
    described in the technical report [NOAA+1976](bibliography.md#NOAA+1976).
    """

    def to_dataset(
        self,
        z: t.Optional[pint.Quantity] = None,
        interp_method: t.Optional[t.Mapping[str, str]] = None,
        conserve_column: bool = False,
        **kwargs: t.Any,
    ) -> xr.Dataset:
        # Since the ussa_1976 model can be evaluated at any altitude, both
        # interp_method and conserve_column are ignored.

        # kwargs are ignored
        if kwargs:
            logger.warning(  # pragma: no cover
                "value of the 'kwargs' parameter will be ignored."
            )

        # variable to compute with the ussa1976 package
        variables = [
            "p",
            "t",
            "n_tot",
            "n",
        ]
        # compute profile
        if z is None:
            logging.debug("Computing profile with ussa1976 package")
            ds = ussa1976.compute(variables=variables)
        else:
            logging.debug("Computing profile with ussa1976 package")
            logging.debug("z=%s", z)
            ds = ussa1976.compute(z=z.m_as("m"), variables=variables)

        # extract data
        coords = {"z": to_quantity(ds["z"]).to("km")}

        data_vars = {}
        data_vars["p"] = to_quantity(ds["p"]).to("Pa")
        data_vars["t"] = to_quantity(ds["t"]).to("K")
        data_vars["n"] = to_quantity(ds["n_tot"]).to("m^-3")

        # compute volume fraction
        for s in ds["s"].values:
            nx = to_quantity(ds["n"].sel(s=s))
            n_tot = to_quantity(ds["n_tot"])
            data_vars[f"x_{s}"] = (nx / n_tot).to("dimensionless")

        attrs = {
            "Conventions": "CF-1.10",
            "title": "U.S. Standard Atmosphere 1976",
            "institution": "NOAA",
            "source": ds.attrs["source"],
            "history": ds.attrs["history"] + "\n" + history(),
            "references": ds.attrs["references"],
            "url": "https://ntrs.nasa.gov/citations/19770009539",
            "urldate": "2022-12-08",
        }

        ds = schema.convert(
            data_vars,
            coords,
            attrs,
        )

        return ds

Command line interface

Command-line interface.

main(file_name, identifier, altitudes, altitude_units, represent_in_cells, p_interp_method, t_interp_method, n_interp_method, x_interp_method)

Joseki command-line interface.

Source code in src/joseki/__main__.py

@click.command()
@click.option(
    "--file-name",
    "-f",
    help="Output file name.",
    default="ds.nc",
    show_default=True,
    type=click.Path(writable=True),
)
@click.option(
    "--identifier",
    "-i",
    help="Atmospheric profile identifier.",
    required=True,
    type=click.Choice(
        choices=IDENTIFIER_CHOICES,
        case_sensitive=True,
    ),
)
@click.option(
    "--altitudes",
    "-a",
    help=(
        "Path to level altitudes data file. The data file is read with "
        "pandas.read_csv. The data file must be a column named 'z'."
    ),
    default=None,
    show_default=True,
)
@click.option(
    "--altitude-units",
    "-u",
    help="Altitude units",
    default="km",
    show_default=True,
)
@click.option(
    "--represent-in-cells",
    "-r",
    help=(
        "Compute the cells representation of the atmospheric profile. The "
        "initial altitude values are used to define the altitude bounds of "
        "each cell. The pressure, temperature, number density and mixing "
        "ratio fields are interpolated at the cells' center altitudes."
    ),
    is_flag=True,
)
@click.option(
    "--p-interp-method",
    "-p",
    help="Pressure interpolation method.",
    type=click.Choice(
        INTERPOLATION_METHOD_CHOICES,
        case_sensitive=True,
    ),
    default="linear",
    show_default=True,
)
@click.option(
    "--t-interp-method",
    "-t",
    help="Temperature interpolation method.",
    type=click.Choice(
        INTERPOLATION_METHOD_CHOICES,
        case_sensitive=True,
    ),
    default="linear",
    show_default=True,
)
@click.option(
    "--n-interp-method",
    "-n",
    help="Number density interpolation method.",
    type=click.Choice(
        INTERPOLATION_METHOD_CHOICES,
        case_sensitive=True,
    ),
    default="linear",
    show_default=True,
)
@click.option(
    "--x-interp-method",
    "-x",
    help="Volume mixing ratios interpolation method.",
    type=click.Choice(
        INTERPOLATION_METHOD_CHOICES,
        case_sensitive=True,
    ),
    default="linear",
    show_default=True,
)
@click.version_option()
def main(
    file_name: str,
    identifier: str,
    altitudes: t.Optional[str],
    altitude_units: str,
    represent_in_cells: bool,
    p_interp_method: str,
    t_interp_method: str,
    n_interp_method: str,
    x_interp_method: str,
) -> None:
    """Joseki command-line interface."""
    # read altitude grid
    if altitudes is not None:
        df = pd.read_csv(pathlib.Path(altitudes))
        z = df["z"].values * ureg(altitude_units)
    else:
        z = None

    # make dataset
    ds = make(
        identifier=identifier,
        z=z,
        p_interp_method=p_interp_method,
        t_interp_method=t_interp_method,
        n_interp_method=n_interp_method,
        x_interp_method=x_interp_method,
        represent_in_cells=represent_in_cells,
    )

    # write dataset
    ds.to_netcdf(file_name)

Core

Core module.

make(identifier, z=None, interp_method=DEFAULT_METHOD, represent_in_cells=False, conserve_column=False, **kwargs)

Create a profile with the specified identifier.

Parameters:

Name	Type	Description	Default
identifier	str	Profile identifier.	required
z	t.Optional[pint.Quantity]	Altitude values.	None
interp_method	t.Optional[t.Mapping[str, str]]	Mapping of variable and interpolation method.	DEFAULT_METHOD
represent_in_cells	bool	If True, compute the altitude layer centers and interpolate the profile on the layer centers, and return the interpolated profile.	False
conserve_column	bool	If True, ensure that column densities are conserved during interpolation.	False
kwargs	t.Any	Additional keyword arguments passed to the profile constructor.	{}

Returns:

Type	Description
xr.Dataset	Profile as xarray.Dataset.

Source code in src/joseki/core.py

def make(
    identifier: str,
    z: t.Optional[pint.Quantity] = None,
    interp_method: t.Optional[t.Mapping[str, str]] = DEFAULT_METHOD,
    represent_in_cells: bool = False,
    conserve_column: bool = False,
    **kwargs: t.Any,
) -> xr.Dataset:
    """
    Create a profile with the specified identifier.

    Args:
        identifier: Profile identifier.
        z: Altitude values.
        interp_method: Mapping of variable and interpolation method.
        represent_in_cells: If `True`, compute the altitude layer centers and 
            interpolate the profile on the layer centers, and return the 
            interpolated profile.
        conserve_column: If `True`, ensure that column densities are conserved
            during interpolation.
        kwargs: Additional keyword arguments passed to the profile constructor.

    Returns:
        Profile as xarray.Dataset.
    """
    logger.info("Creating profile %s", identifier)
    logger.debug("z: %s", z)
    logger.debug("interp_method: %s", interp_method)
    logger.debug("represent_in_cells: %s", represent_in_cells)
    logger.debug("conserve_column: %s", conserve_column)
    logger.debug("kwargs: %s", kwargs)

    profile = factory.create(identifier)

    logger.debug("exporting profile to xarray.Dataset")
    ds = profile.to_dataset(
        z=z,
        interp_method=interp_method,
        conserve_column=conserve_column,
        **kwargs,
    )

    if represent_in_cells:
        ds = represent_profile_in_cells(
            ds,
            method=interp_method,
            conserve_column=conserve_column,
        )

    return ds

Units

Units module.

to_quantity(da)

Convert a xarray.DataArray to a pint.Quantity.

Notes

The array's attrs metadata mapping must contain a units field. This function can also be used on coordinate variables.

Parameters:

Name	Type	Description	Default
da	xr.DataArray	xarray.DataArray instance which will be converted.	required

Raises:

Type	Description
ValueError	If the xarray.DataArray's `attrs field does not contain a units key.

Returns:

Type	Description
pint.Quantity	The corresponding quantity.

Source code in src/joseki/units.py

def to_quantity(da: xr.DataArray) -> pint.Quantity:
    """Convert a `xarray.DataArray` to a `pint.Quantity`.


    Notes:
        The array's `attrs` metadata mapping must contain a `units` field.
        This function can also be used on coordinate variables.

    Args:
        da: xarray.DataArray instance which will be converted.

    Raises:
        ValueError: If the `xarray.DataArray`'s ``attrs` field does not 
            contain a `units` key.

    Returns:
        The corresponding quantity.
    """
    try:
        units = da.attrs["units"]
    except KeyError as e:
        raise ValueError("this DataArray has no 'units' metadata field") from e
    else:
        return ureg.Quantity(da.values, units)

Util

Utility module.

utcnow()

Get current UTC time.

Returns:

Type	Description
str	ISO 8601 formatted UTC timestamp.

Source code in src/joseki/profiles/util.py

def utcnow() -> str:
    """Get current UTC time.

    Returns:
        ISO 8601 formatted UTC timestamp.
    """
    return datetime.datetime.utcnow().replace(microsecond=0).isoformat()