import copy
import os
from logging import info
import numpy as np
import pandas as pd
from ......utils.datastores.empty import init_empty
from ......utils.netcdf import readnc
from ......utils.check.errclass import CifError
[docs]
def read_sim(self, data2load, runsubdir, mode, ddi, ddf):
"""Read the mod.txt file as simulated by CHIMERE"""
# If species to extract are empty, pass
# extract = len([trid for trid in data2load if len(
# data2load[trid][ddi]) > 0]) > 0
# if not extract:
# return data2load
# If no simulated concentration is available just pass
sim_file = f"{runsubdir}/mod.txt"
if not os.path.isfile(sim_file):
raise CifError(
f"CHIMERE did not produce a mod.txt file in {runsubdir}. Please check the directory. If using the option autorestart in the observation operator, always remove the first non-valid sub-directory"
)
if os.stat(sim_file).st_size == 0:
info(
"CHIMERE ran without any observation "
"to be compared with for sub-simu "
"only CHIMERE's outputs are available"
)
# Filling simulations with 0
for trid in data2load:
data2load[trid][ddi] = init_empty()
data2load[trid][ddi].loc[:, "sim"] = np.nan
return data2load
# Read simulated concentrations
data_ref = pd.read_csv(
sim_file,
sep="\s+",
header=None,
index_col=None,
usecols=range(5, 12) if mode == "tl" else range(5, 11),
names=["param", "sim", "pmid", "dp", "airm", "hlay"]
+ (["simfwd"] if mode == "tl" else []),
)
# Loop over species
avail_species = data_ref["param"].unique()
dataout = {}
for trid in data2load:
spec = trid[1].replace("__sample#", "")
if spec not in avail_species:
continue
# Use sub-dataset as determine in make_obs
inds = self.auxiliary_indexes[ddi]["concs"][spec]
data = data_ref.iloc[inds[1]: inds[1] + inds[0]][
["sim", "pmid", "dp", "airm", "hlay"] +
(["simfwd"] if mode == "tl" else [])
]
dataloc = init_empty()
# Check whether different transforms use this extraction
if trid[0] != "concs":
if len(self.chunk_indexes[ddi]["concs"][spec]) > 0:
out_transforms = np.unique(
self.chunk_indexes[ddi]["concs"][spec])
valid_transforms = [
i
for i in out_transforms
if (self.chunk_indexes[ddi]["concs"][spec] == i).sum()
== self.auxiliary_indexes[ddi][trid[0]][spec]
]
if len(valid_transforms) == 1:
data = data[
self.chunk_indexes[ddi]["concs"][spec] == valid_transforms[0]
]
else:
raise CifError(
"I could not determine what chunk should "
"be extracted. This is unlikely to happen, "
"but if does, please report to developpers"
)
# Putting values to the local data store
column = "spec" if mode == "fwd" else "incr"
dataloc.loc[:, ("maindata", column)] = data.loc[:, "sim"].values
dataloc.loc[:, ("metadata", "pressure")] = data.loc[:, "pmid"].values
dataloc.loc[:, ("metadata", "dp")] = data.loc[:, "dp"].values
dataloc.loc[:, ("metadata", "airm")] = data.loc[:, "airm"].values
dataloc.loc[:, ("metadata", "hlay")] = data.loc[:, "hlay"].values
if column == "incr":
dataloc.loc[:, ("maindata", "spec")] = data.loc[:, "simfwd"].values
else:
dataloc.loc[:, ("maindata", "incr")] = 0.0
# Put simulated value into correct column
# Different case if concs, or other parameters such as pressure
# Put pressure and other auxiliary data into spec column for later
# interpolation
if trid[0] != "concs":
# Column name
col = trid[0]
if trid[0] == "dpressure":
col = "dp"
dataloc.loc[:, ("maindata", "spec")] = copy.deepcopy(
dataloc.loc[:, ("metadata", col)].values
)
dataloc.loc[:, ("maindata", "incr")] = 0.0
dataout[trid] = copy.deepcopy(dataloc)
del dataout[trid][("metadata", "pressure")]
del dataout[trid][("metadata", "dp")]
del dataout[trid][("metadata", "airm")]
del dataout[trid][("metadata", "hlay")]
return dataout
[docs]
def read_obs(runsubdir, mode="fwd"):
"""Read obs.txt file"""
obs_file = f"{runsubdir}/obs.txt"
# Read observations' characteristics
columns = ["ihourrun", "np", "ime", "izo", "ivert", "spec"]
if mode == "adj":
columns += ["dy"]
data = pd.read_csv(
obs_file,
sep="\s+",
header=0,
names=columns,
)
return data
[docs]
def write_sim(data, runsubdir):
"""Write mock mod.txt file"""
sim_file = f"{runsubdir}/mod.txt"
data = data.assign(sim=0.0)
data = data.assign(p_mid=100000.0)
data = data.assign(ddp=1000.0)
data = data.assign(airmloc=2.5e20)
data = data.assign(thlayloc=10000.0)
data = data.assign(sim_ref=0.0)
data.to_csv(sim_file, sep=' ', header=None, index=False)
[docs]
def write_obs(data, runsubdir):
"""Write mock obs.txt file"""
obs_file = f"{runsubdir}/obs.txt"
nbobs, nbdatatot = pd.read_csv(
obs_file, sep="\s+", header=None, nrows=1).values[0]
with open(obs_file, "w") as f:
f.write(str(int(nbobs)) + " " + str(int(nbdatatot)) + "\n")
data = data.assign(dy=0.0)
data.to_csv(f, sep=' ', header=None, index=False)
[docs]
def readend(ficin):
"""Read end.nc file"""
list_spec = readnc(ficin, ['species']).astype('str')
list_names_spec = []
end = []
for spec in list_spec:
strsp = ''
i = 0
while i < len(spec):
if spec[i] != ' ':
strsp = strsp + spec[i]
else:
i = len(spec)
i = i + 1
if strsp.find('_o') == -1:
list_names_spec.append(strsp)
conc = readnc(ficin, [strsp])
end.append(conc)
return list_names_spec, end
[docs]
def readmeteo(ficin, list_var):
"""Read variables in METEO.nc file"""
met = []
for var in list_var:
var = readnc(ficin, [var])
met.append(var)
return met
