from ....utils.path import init_dir
from ....utils.classes.domains import Domain
import os
import numpy as np
import pandas as pd
from shutil import copytree, ignore_patterns, rmtree, copy
[docs]
def ini_mapper(model, transform_type, general_mapper={}, backup_comps={},
transforms_order=[], ref_transform="", **kwargs):
"""Build the data-flow mapper for the Lagrangian (FLEXPART/STILT) model.
Declares:
* **Flux inputs** — one surface flux field per active species, with
date windows that extend backward by ``model.backward_trajdays`` to
cover the full footprint integration period.
* **Background inputs** (optional) — a background concentration field
read from the period shifted backward by ``backward_trajdays``.
* **Outputs** — one sampled concentration field per active species.
Args:
model: Lagrangian model plugin instance (carries ``input_dates``,
``input_background``, ``tstep_dates``, ``domain``, and
``chemistry.acspecies``).
transform_type (str): unused; kept for API compatibility.
general_mapper (dict): unused.
backup_comps (dict): unused.
transforms_order (list): unused.
ref_transform (str): unused.
**kwargs: unused.
Returns:
dict: mapper with ``inputs``, ``outputs``, and ``outputs2inputs``.
"""
# Input intervals come from input_dates, but account for footprint durations
# by adding an extra input before the start of the month
input_intervals = {
ddi: np.concatenate([
model.input_dates[ddi][:-1, np.newaxis],
model.input_dates[ddi][1:, np.newaxis]],
axis=1)
for ddi in model.input_dates}
background_intervals = {
ddi: np.concatenate([
model.input_background[ddi][:-1, np.newaxis],
model.input_background[ddi][1:, np.newaxis]],
axis=1)
for ddi in model.input_background
}
output_intervals = {
ddi: np.append(
model.tstep_dates[ddi][:-1, np.newaxis],
model.tstep_dates[ddi][1:, np.newaxis],
axis=1)
for ddi in model.tstep_dates}
default_dict = {'input_dates': input_intervals, 'force_loadin': True,
'force_dump': True, "sparse_data": False, "sampled": False}
dict_background = dict(
default_dict,
**{'domain': model.domain, "input_dates": background_intervals})
# For emissions, generate a dictionary with one surface level
domain_in = model.domain
domain_out = Domain(nlon=domain_in.nlon, nlat=domain_in.nlat,
zlon=domain_in.zlon, zlat=domain_in.zlat,
zlonc=domain_in.zlonc, zlatc=domain_in.zlatc,
nlon_side=domain_in.nlon_side, nlat_side=domain_in.nlat_side,
zlon_side=domain_in.zlon_side, zlat_side=domain_in.zlat_side,
zlonc_side=domain_in.zlonc_side, zlatc_side=domain_in.zlatc_side,
nlev=1, pressure_unit="Pa",
unstructured_domain=domain_in.unstructured_domain,
sigma_b=np.array([1]), sigma_a=np.array([0]),
sigma_b_mid=np.array([1]), sigma_a_mid=np.array([0]))
dict_emis = dict(default_dict, **{"domain": domain_out, "unit": "ng/m2/s"})
# Executable
mapper = {'inputs': {},
'outputs': {('concs', s): {"force_loadout": True,
"domain": model.domain,
"input_dates": output_intervals,
"continuous_hdomain": True,
"continuous_vdomain": True,
"sampled": True,
"sparse_data": False
}
for s in model.chemistry.acspecies.attributes}
}
# Updating mapper if using background
if model.read_background:
mapper["inputs"].update({
('inicond', s): dict_background
for s in model.chemistry.acspecies.attributes})
# Update mapper if using surface fluxes
if model.read_surface_sensitivity:
mapper["inputs"].update({
('flux', s): dict_emis
for s in model.chemistry.acspecies.attributes})
# Save propagation of perturbations
inputs = ["inicond", "flux", "restart_inicond"]
mapper["outputs2inputs"] = {
("concs", s):
[(cmp, s) for cmp in inputs]
for s in model.chemistry.acspecies.attributes
}
return mapper
