import numpy as np
import datetime
import os
import pandas as pd
from logging import info
from .....utils.datastores.dump import dump_datastore
from ..utils.flexpart_header import read_header, Flexpartheader
from .inputs.fluxes import flux_contribution
from .inputs.inicond import inicond_contribution
from .....utils.check.errclass import CifFileNotFoundError
[docs]
def outputs2native(self, data2dump, input_type,
di, df, runsubdir, mode='fwd', onlyinit=False,
check_transforms=False,
**kwargs):
"""Reads outputs to pycif objects.
Does nothing for now as we instead read Lagrangian output
inside loop over observations in obsoper.py
"""
# If no data to extract, pass
if data2dump == {} or onlyinit:
return data2dump
ddi = min(di, df)
# Specific behaviour in batch computation
batch_computation = hasattr(self, "perturbed_species")
dataout = {}
for spec in self.chemistry.acspecies.attributes:
trid = ("concs", spec)
if trid not in data2dump:
continue
# Skip if reference species already processed
# in case of batch computing
ref_spec = getattr(self, "perturbed_species", {}).get(spec, spec)
if ref_spec in self.process_sample_species[ddi]:
continue
self.process_sample_species[ddi].append(ref_spec)
# Load all samples at once in batch computation
dataobs = {spec: self.dataobs[ddi][spec]}
if batch_computation:
ref_species = self.perturbed_species[spec]
sample_species = [
s for s in self.chemistry.acspecies.attributes
if self.perturbed_species[s] == ref_species
]
dataobs = {
s: self.dataobs[ddi][s]
for s in sample_species
}
nobs = len(dataobs[spec])
subdir = ddi.strftime(self.footprint_dir_format)
if self.footprint_type == "STILT":
fp_header_nest = Flexpartheader()
fp_header_glob = Flexpartheader()
fp_header_nest.outheight = [1]
else:
# Ref station ID for header
ref_header = getattr(
self, "ref_header_ID",
dataobs[spec].head(1)["metadata"]['station'].values[0].upper())
# Initialize header
fp_header_glob = None
header_nest = "header"
if self.domain.nested:
fp_header_glob = read_header(
self, os.path.join(
self.run_dir_glob,
ref_header,
subdir, 'header')
)
header_nest = "header_nest"
if self.force_read_nest and not self.domain.nested:
header_nest = "header_nest"
file_header_nest = os.path.join(
self.run_dir_nest,
ref_header,
subdir, header_nest)
try:
fp_header_nest = read_header(
self,
file_header_nest
)
except FileNotFoundError:
raise CifFileNotFoundError(
f"Could not find {file_header_nest} "
"with following arguments: \n"
f"- run_dir_nest: {self.run_dir_nest}\n"
f"- ref_header_ID: {getattr(self, 'ref_header_ID', None)}\n"
f"- ref_header: {ref_header}\n"
f"- footprint_dir_format: {getattr(self, 'footprint_dir_format', None)}\n"
"Looking for a file with format: \n"
f"run_dir_nest / ref_header_ID or station_ID / footprint_dir_format / header_nest"
)
fp_header_init = None
if self.read_background:
file_header_init = os.path.join(
self.run_dir_bg,
ref_header,
subdir, 'header')
try:
fp_header_init = read_header(
self, file_header_init
)
except FileNotFoundError:
raise CifFileNotFoundError(
f"Could not find {file_header_init} "
"with following arguments: \n"
f"- run_dir_bg: {self.run_dir_bg}\n"
f"- ref_header_ID: {getattr(self, 'ref_header_ID', None)}\n"
f"- footprint_dir_format: {getattr(self, 'footprint_dir_format', None)}\n"
"Looking for a file with format: \n"
f"run_dir_bg / ref_header_ID or station_ID / footprint_dir_format / header_nest"
)
# Nest domain definition
ix1 = self.domain.ix1
ix2 = self.domain.ix2
iy1 = self.domain.iy1
iy2 = self.domain.iy2
# Save to datastore for debugging purposes
obs_ghg = np.nan * np.empty(nobs)
obs_bkg = np.nan * np.empty(nobs)
obs_sim = np.nan * np.empty(nobs)
obs_model = np.nan * np.empty(nobs)
obs_check = np.nan * np.empty(nobs)
obs_bkgerr = np.nan * np.empty(nobs)
obs_err = np.nan * np.empty(nobs)
info(f"di, df: {di}, {df}, {datetime.datetime.now()}")
# Apply fluxes contribution
if self.read_surface_sensitivity:
flux_contribution(
self, mode, dataobs,
fp_header_nest, fp_header_glob, spec, ddi,
batch_computation=batch_computation
)
# Apply sensitivity to background if required
if self.read_background:
inicond_contribution(
self, mode, dataobs,
fp_header_init, spec, ddi,
batch_computation=hasattr(self, "perturbed_species")
)
# Update data2dump
dataout[trid] = dataobs[spec]
if batch_computation:
ref_species = self.perturbed_species[spec]
sample_species = [
s for s in self.chemistry.acspecies.attributes
if self.perturbed_species[s] == ref_species
]
for s in sample_species:
dataout[("concs", s)] = dataobs[s]
# Dump temporary datastore
if not self.dump_debug:
continue
file_name = f"{runsubdir}/debug_monitor_{spec}.nc"
col2dump = ["nest", "nest_tl",
"glob", "glob_tl",
"background", "background_tl"]
dump_datastore(dataobs[spec],
file_monit=file_name,
mode='w', dump_type="nc",
col2dump=col2dump)
return dataout
