# -*- coding: utf-8 -*-
"""
ensemble.py
-----------
Algo Components for ensemble Surfex simulations.
.. inheritance-diagram:: vortex_cen.algo.ensemble
:top-classes: vortex_cen.algo.components._CenParaBlindRun, vortex_cen.algo.components._CenTaylorRun,
vortex_cen.algo.components._CenTaylorVortexWorker, vortex_cen.algo.components._CenWorkerBlindRun,
vortex.algo.components.ParaExpresso, vortex.tools.parallelism.TaylorVortexWorker
vortex.algo.components.AlgoComponent, vortex.algo.components.Parallel, vortex.algo.components.TaylorRun
:private-bases:
:parts: 1
.. autoclass:: S2MComponent
:no-members:
:show-inheritance:
.. autoclass:: SurfexComponent
:no-members:
:show-inheritance:
.. autoclass:: SurfexComponentMultiDates
:no-members:
:show-inheritance:
.. autoclass:: SurfexWorker
:no-members:
:show-inheritance:
.. autoclass:: PrepareForcingComponent
:no-members:
:show-inheritance:
.. autoclass:: PrepareForcingWorker
:no-members:
:show-inheritance:
.. autoclass:: S2MReanalysis
:no-members:
:show-inheritance:
.. autoclass:: S2MReforecast
:no-members:
:show-inheritance:
.. autoclass:: ShadowsForcingWorker
:no-members:
:show-inheritance:
"""
import glob
from bronx.fancies import loggers
from bronx.stdtypes.date import Date, Period, tomorrow
from bronx.syntax.externalcode import ExternalCodeImportChecker
import footprints
from vortex.algo.components import DelayedAlgoComponentError
from vortex_cen.algo.components import _CenParaBlindRun, _CenWorkerBlindRun, _CenTaylorRun
from vortex_cen.algo.components import _CenTaylorVortexWorker
from vortex.syntax.stdattrs import a_date
from vortex.tools.systems import ExecutionError
logger = loggers.getLogger(__name__)
echecker = ExternalCodeImportChecker('snowtools')
with echecker:
from snowtools.tools.change_prep import prep_tomodify
from snowtools.utils.resources import save_file_period, save_file_date
from snowtools.tools.update_namelist import update_surfex_namelist_object
from snowtools.tools.change_forcing import forcinput_select, forcinput_applymask
from snowtools.utils.infomassifs import infomassifs
from snowtools.utils.ESCROCsubensembles import ESCROC_subensembles
from snowtools.utils.FileException import TimeListException, MultipleValueException
from vortex_cen.algo.deterministic import SurfexMixIn
class S2MExecutionError(ExecutionError):
"""Execution Error in S2M algo component that should be catched and delayed"""
def __init__(self, model, deterministic, subdir, datebegin, dateend):
self.model = model
self.deterministic = deterministic # Key used for delayed exception management
self.subdir = subdir
self.datebegin = datebegin
self.dateend = dateend
super().__init__(self.model + ' execution failed.')
def __str__(self):
return ("Error while running " + self.model + " for member " + self.subdir + " for period " +
self.datebegin.ymdh + " - " + self.dateend.ymdh)
def __reduce__(self):
red = list(super().__reduce__())
red[1] = tuple([self.model, self.deterministic, self.subdir, self.datebegin,
self.dateend]) # Les arguments qui seront passes a __init__
return tuple(red)
class S2MMissingDeterministicError(DelayedAlgoComponentError):
"""
Exception raised when no resource is found for a mandatory role although
fatal is False in toolbox inputs.
"""
def __init__(self, role):
self.role = role
self.deterministic = True
def __str__(self):
return "Unable to find the mandatory resource of role : " + self.role
def __reduce__(self):
red = list(super().__reduce__())
red[1] = tuple([self.role, self.deterministic])
return tuple(red)
[docs]
@echecker.disabled_if_unavailable
class SurfexWorker(_CenWorkerBlindRun, SurfexMixIn):
"""
This algo component is designed to run a SURFEX experiment without
MPI parallelization.
"""
_footprint = dict(
info = 'AlgoComponent designed to run a SURFEX experiment without MPI parallelization.',
attr = dict(
datebegin = a_date,
dateend = a_date,
dateinit = a_date,
kind = dict(
values = ['deterministic', 'escroc', 'ensmeteo', 'ensmeteonodet', 'ensmeteo+sytron', 'ensmeteo+escroc',
'croco'],
# Les cas "escroc" et "croco" s
# * escroc = 1 FORCING (no subdirs) / N physiques Crocus
# * croco = N FORCINGs (--> N subdirs) / N physiques Crocus
),
threshold = dict(
info = "Threshold to apply to SWE",
type = int,
optional = True,
default = -999
),
physical_options = dict(
info = "Dictionnary of ESCROC physical options",
type = dict,
optional = True,
default = {}
),
snow_parameters = dict(
info = "Dictionnary of ESCROC snow physical parameters",
type = dict,
optional = True,
default = {}
),
subdir = dict(
info = 'work in this particular subdirectory',
optional = False
),
geometry_in=dict(
info="Area information in case of an execution on a massif geometry",
type=footprints.stdtypes.FPList,
),
geometry_out=dict(
info="The resource's massif geometry.",
type=str,
),
daily = dict(
info = "If True, split simulations in daily runs",
type = bool,
optional = True,
default = False,
),
dailynamelist = dict(
info = "If daily is True, possibility to provide a list of namelists to change each day",
type = list,
optional = True,
default = [],
),
)
)
def modify_prep(self, datebegin_this_run):
"""
The PREP file needs to be modified if the init date differs from the starting
date or if a threshold needs to be applied on snow water equivalent.
"""
modif_swe = self.threshold > 0 and datebegin_this_run.month == 8 and datebegin_this_run.day == 1
modif_date = datebegin_this_run == self.datebegin and self.datebegin != self.dateinit
modif = modif_swe or modif_date
if modif:
prep = prep_tomodify("PREP.nc")
if modif_swe:
print("APPLY THRESHOLD ON SWE.")
prep.apply_swe_threshold(self.threshold)
if modif_date:
print("CHANGE DATE OF THE PREP FILE.")
prep.change_date(self.datebegin)
prep.close()
else:
print("DO NOT CHANGE THE PREP FILE.")
def _commons(self, rundir, thisdir, rdict, **kwargs):
# TODO : tout cela pourrait être géré plus proprement à partir des inputs du contexte
if len(self.dailynamelist) > 1:
list_files_copy = self.dailynamelist
else:
list_files_copy = ["OPTIONS.nam"]
list_files_link = ["PGD.nc", "METADATA.xml", "ecoclimapI_covers_param.bin",
"ecoclimapII_eu_covers_param.bin", "drdt_bst_fit_60.nc"]
if self.kind in ['escroc', 'croco'] and (self.datebegin != self.dateinit or self.threshold > 0):
list_files_copy_ifnotprovided = ["PREP.nc"]
list_files_link_ifnotprovided = []
else:
list_files_copy_ifnotprovided = []
list_files_link_ifnotprovided = ["PREP.nc"]
# here Bertrand also created links towards the forcings but in a non-standard way --> to be checked
for required_copy in list_files_copy:
self.copy_if_exists(self.system.path.join(rundir, required_copy), required_copy)
for required_link in list_files_link:
self.link_in(self.system.path.join(rundir, required_link), required_link)
for required_link in list_files_link_ifnotprovided:
self.link_ifnotprovided(self.system.path.join(rundir, required_link), required_link)
# For reforecast:
self.link_ifnotprovided(self.system.path.join(self.system.path.dirname(thisdir), required_link),
required_link)
for required_copy in list_files_copy_ifnotprovided:
self.copy_ifnotprovided(self.system.path.join(rundir, required_copy), required_copy)
self.copy_ifnotprovided(self.system.path.join(self.system.path.dirname(thisdir), required_copy),
required_link)
rdict = self._surfex_task(rundir, thisdir, rdict)
self.postfix()
return rdict
def _surfex_task(self, rundir, thisdir, rdict):
# ESCROC cases: each member will need to have its own namelist
# meteo ensemble cases: the forcing modification must be applied to all members and the namelist
# generation requires that the forcing generation has already be done. Therefore, preprocessing
# is done in the offline algo in all these cases
# Determinstic cases : the namelist is prepared in the preprocess algo component in order to allow
# to build PGD and PREP
# MV : Cela justifierai de faire des algos distincts...
namelist_ready = self.kind == 'deterministic' # Cas oper uniquement
need_other_run = True
need_other_forcing = True
need_save_forcing = False
updateloc = True
datebegin_this_run = self.datebegin
sytron = self.kind == "ensmeteo+sytron" and self.subdir == "mb036" # MV : A sortir de l'algo (S2M-specific)
changenamelistdaily = self.daily and (len(self.dailynamelist) > 1) # MV : Faire un aglo spécifique
while need_other_run:
# Modification of the PREP file
self.modify_prep(datebegin_this_run)
if need_other_forcing:
# MV : L'unique différence entre "escroc" et "croco" est la valeur de "forcingdir"
# TODO : externalise les lignes suivantes de la boucle "need_other_run"
# TODO : gérer le cas "SYTRON" plus proprement (algo / footprint spécifique pour remplaçer)
# la valeur en dur"/mb035"
# TODO : Utiliser les "effective_inputs" pour définir "forcingdir"
if self.kind == "escroc": # MV : Faire un algo spécifqique
# ESCROC only: the forcing files are in the father directory (same forcing for all members)
forcingdir = rundir
elif sytron: # MV : S2M-spécifique, faire un algo distinct
# ensmeteo+sytron: the forcing files are supposed to be in the subdirectories
# of each member except for the sytron member
forcingdir = rundir + "/mb035"
else: # MV : Faire des algos distincts pour chaque cas
# ensmeteo or ensmeteo+escroc or croco: the forcing files are supposed to be in the subdirectories
# of each member
# determinstic case: the forcing file(s) is/are in the only directory
forcingdir = thisdir
if len(self.geometry_in) > 1: # MV : A remplacer par la tâche d'aggrégation de forçages
print("FORCING AGGREGATION")
forcinglist = []
# MV: La boucle qui suit semble très S2M-spécifique et devrait être externalisée
# --> utiliser la tâche d'aggrégation
for massif in self.geometry_in:
try:
dateforcbegin, dateforcend, forcingname = self.find_forcing(datebegin_this_run,
self.dateend)
self.link_in(self.system.path.join(forcingdir, massif, forcingname), 'FORCING.nc')
except (FileNotFoundError, MultipleValueException) as e:
rdict['rc'] = e
return rdict # Note than in the other case return rdict is at the end
forcingname = "FORCING_" + massif + ".nc"
self.system.mv("FORCING.nc", forcingname)
forcinglist.append(forcingname)
print(forcinglist)
try:
forcinput_applymask(forcinglist, "FORCING.nc", **self.reprod_info)
except TimeListException:
deterministic = self.subdir == "mb035"
rdict['rc'] = S2MExecutionError("merge of forcings", deterministic, self.subdir,
dateforcbegin, dateforcend)
return rdict # Note than in the other case return rdict is at the end
need_save_forcing = True
else:
# Get the first file covering part of the whole simulation period
# MV : Et pour une période de simulation correspondant à une sous-période
# d'un forçage ?
# La stratégie d'essayer de deviner les FORCINGs présent à partir de périodes pré-définies
# ne semble pas optimale: à ce stade on connait précisément les FORCINGs présents.
# On peut donc:
# 1. Récupérer la liste des datebegin/dateend des FORCINGs présents
# 2. Construire la liste des itérations nécéssaires pour couvrir la période de simulation
# 3. Boucler sur cette liste
print("LOOK FOR FORCING")
try:
dateforcbegin, dateforcend, forcingname = self.find_forcing(datebegin_this_run, self.dateend)
self.link_in(self.system.path.join(forcingdir, forcingname), 'FORCING.nc')
except (FileNotFoundError, MultipleValueException) as e:
rdict['rc'] = e
return rdict
print("FORCING FOUND")
# MV : S2M-spécifique, à externaliser
# --> utiliser la tâche "AddSlopes"
if "flat" in self.geometry_in[0] and "allslopes" in self.geometry_out:
print("FORCING EXTENSION")
liste_massifs = infomassifs().dicArea[self.geometry_in[0]]
liste_aspect = infomassifs().get_list_aspect(8, ["0", "20", "40"])
self.mv_if_exists("FORCING.nc", "FORCING_OLD.nc")
forcinput_select('FORCING_OLD.nc', 'FORCING.nc', liste_massifs, 0, 5000,
["0", "20", "40"], liste_aspect, **self.reprod_info)
need_save_forcing = True
if self.daily: # MV : reforecast-spécifique, faire un algo dédié
dateend_this_run = min(tomorrow(base=datebegin_this_run), min(self.dateend, dateforcend))
need_other_forcing = False
else:
dateend_this_run = min(self.dateend, dateforcend)
if not namelist_ready:
if sytron:
self.copy_if_exists(self.system.path.join(rundir, "OPTIONS_sytron.nam"), "OPTIONS.nam")
available_namelists = self.find_namelists()
if len(available_namelists) > 1:
print("WARNING SEVERAL NAMELISTS AVAILABLE !!!")
for namelist in available_namelists:
# Update the contents of the namelist (date and location)
# Location taken in the FORCING file.
print("MODIFY THE NAMELIST:" + namelist.container.basename)
newcontent = update_surfex_namelist_object(
namelist.contents,
datebegin_this_run,
dateend=dateend_this_run,
updateloc=updateloc,
physicaloptions=self.physical_options,
snowparameters=self.snow_parameters)
newnam = footprints.proxy.container(filename=namelist.container.basename)
newcontent.rewrite(newnam)
newnam.close()
if self.daily:
updateloc = True
if changenamelistdaily:
# Change the namelist
self.link_in(self.dailynamelist.pop(0), "OPTIONS.nam")
# Run surfex offline
list_name = self.system.path.join(thisdir, 'offline.out')
try:
self.local_spawn(list_name)
# Uncomment these lines to test the behaviour in case of failure of 1 member
# if self.subdir == "mb006": # To test warnings
# if self.subdir == "mb035": # To test errors
# deterministic = self.subdir == "mb035"
# # print("DEBUGINFO")
# # print(dir(self))
# rdict['rc'] = S2MExecutionError(self.progname, deterministic,
# self.subdir, datebegin_this_run, dateend_this_run)
# return rdict
except ExecutionError:
deterministic = self.subdir == "mb035"
rdict['rc'] = S2MExecutionError(self.progname, deterministic, self.subdir,
datebegin_this_run, dateend_this_run)
return rdict # Note than in the other case return rdict is at the end
# Copy the SURFOUT file for next iteration
self.system.cp("SURFOUT.nc", "PREP.nc")
# Post-process
self.offline_postprocess(datebegin_this_run, dateend_this_run)
# Rename outputs with the dates
save_file_date(".", "SURFOUT", dateend_this_run, newprefix="PREP")
# Prepare next iteration if needed
datebegin_this_run = dateend_this_run
need_other_run = dateend_this_run < self.dateend
print(dateend_this_run, self.dateend)
print("INFO SAVE FORCING", need_save_forcing, need_other_run, need_other_forcing)
if need_save_forcing and not (need_other_run and not need_other_forcing):
save_file_period(".", "FORCING", dateforcbegin, dateforcend)
# Remove the symbolic link for next iteration (not needed since now we rename the forcing just before
# self.system.remove("FORCING.nc")
return rdict
[docs]
@echecker.disabled_if_unavailable
class PrepareForcingWorker(_CenTaylorVortexWorker):
"""This algo component is designed to prepare a SURFEX Forcing file (change of geometry)."""
_footprint = dict(
info = 'AlgoComponent designed to run a SURFEX experiment without MPI parallelization.',
attr = dict(
kind = dict(
values = ['prepareforcing'],
),
subdir = dict(
info = 'work in this particular subdirectory',
optional = False
),
geometry_in = dict(
info = "Area information in case of an execution on a massif geometry",
# WARNING : geometry_in contient la liste des TAGs des géométries, et pas de objets
# "geometry"
type = footprints.stdtypes.FPList,
optional = True,
default = None
),
geometry_out = dict(
# WARNING : geometry_ est le TAG de l'objet geometry
info = "The resource's massif geometry.",
type = str,
),
reprod_info = dict(
info="Informations that must be stored in output files for reproductibility",
type=dict,
optional=True,
default=dict(),
)
)
)
def _commons(self, rundir, thisdir, rdict, **kwargs):
rdict = self._prepare_forcing_task(rundir, thisdir, rdict)
self.postfix()
return rdict
def forcingdir(self, rundir, thisdir):
return rundir
def _prepare_forcing_task(self, rundir, thisdir, rdict):
# TODO (MV) : séparer en 2 algo
# 1. Concatenate_forcings
# 2. AddSlopes
if len(self.geometry_in) > 1:
logger.info("FORCING AGGREGATION")
forcinglist = []
for massif in self.geometry_in:
forcingname = "FORCING_" + massif + ".nc"
forcinglist.append(forcingname)
logger.info("FORCING EXTENSION")
forcinput_applymask(forcinglist, "FORCING_OUT.nc", **self.reprod_info)
else:
logger.info("FORCING EXTENSION")
geoin = self.geometry_in[0]
if "flat" in geoin and "allslopes" in self.geometry_out:
list_slopes = ["0", "20", "40"]
liste_massifs = infomassifs().dicArea[self.geometry_in[0]]
liste_aspect = infomassifs().get_list_aspect(8, list_slopes)
forcinput_select(f'FORCING_{geoin}.nc', 'FORCING_OUT.nc', liste_massifs,
0, 5000, list_slopes, liste_aspect, **self.reprod_info)
return rdict
def postfix(self):
self.system.subtitle('{:s} : directory listing (post-run)'.format(self.kind))
for line in self.system.dir():
print(line)
[docs]
class S2MComponent(_CenParaBlindRun):
"""
This Algo Component is designed to manage any Surfex execution without MPI
parallelization (deterministic or ensemble-like simulations).
Ensemble-like simulations include real ensemble simulations (many executions
of the same simulation with different initial conditions or configurations)
and multi-year simulations (associating each year to one member) that can run
in parallel.
The different members of an ensemble simulation are identified by an input
resource that differ between the members (defined by the method
**role_ref_namebuilder** that can be overloaded). For each identified
member, a worker object (whose class inherits from :class:`_CenWorkerBlindRun`)
is generated and the different workers run in parallel.
The :class:`S2MComponent` class (that relies on the :mod:`taylorism`
package):
* allocates the different executions of the same binary to workers
* analyses their feedbacks to look for execution errors
(thanks to the inherited :meth:`~ParaBlindRun._default_post_execute`
method): should the `rc` entry of the dictionary returned by the
worker's :meth:`_CenWorkerBlindRun.vortex_task` method be an Exception, it is
captured and stored.
When the execution of all members finishes, the captured exceptions (see
above) are wrapped in a
:class:`~vortex.algo.components.DelayedAlgoComponentError` exception that
is ultimately raised.
Consequently, we can rely on that to:
* filter/ignore some errors,
* send notifications...
For CEN R&D needs, this can be easily implemented in the Task classes
provided they inherit from the :class:`vortex_cen.layout.nodes.CENTaskMixIn`
mixin. For example:
* To filter execution errors, just define
``filter_execution_error`` in your Task class:
``filter_execution_error = CENTaskMixIn.s2moper_filter_execution_error``
* To send notification emails about fatal and non-fatal errors:
``report_execution_warning = CENTaskMixIn.s2moper_report_execution_warning``
and
``report_execution_errors = CENTaskMixIn.s2moper_report_execution_errors``
"""
_footprint = dict(
info = 'AlgoComponent that runs several executions in parallel.',
attr = dict(
kind = dict(
values = ['safrane', 'syrpluie', 'syrmrr', 'sytist', 'sypluie', 'syvapr',
'syvafi', 'intercep'],
),
datebegin = a_date,
dateend = a_date,
execution = dict(
values = ['analysis', 'forecast'],
optional = True,
),
reprod_info=dict(
info="Informations that must be stored in output files for reproductibility",
type=dict,
optional=True,
default=dict(),
)
)
)
def execute(self, rh, opts):
"""Loop on the various initial conditions provided."""
self._default_pre_execute(rh, opts)
# Update the common instructions
common_i = self._default_common_instructions(rh, opts)
cpl_model = self.get_origin(rh, opts)
subdirs = self.get_subdirs(rh, opts)
self._add_instructions(common_i, dict(subdir=subdirs, deterministic=cpl_model))
self._default_post_execute(rh, opts)
def get_subdirs(self, rh, opts):
"""Get the different member's subdirectories.
One member is associated to each 'effective input' (inputs that where
actually retrieved during the fetch step) Section with a role matching
the one defined by the **role_ref_namebuilder** method.
"""
deterministic_member = self.context.sequence.effective_inputs(role=self.role_deterministic_namebuilder())
# Produce a delayed algo component error if no deterministic member in order to let the members run
# but crash at the end
if len(deterministic_member) < 1 and self.kind != 'ensmeteonodet':
self.delayed_exception_add(S2MMissingDeterministicError(self.role_deterministic_namebuilder()),
traceback=True)
avail_members = deterministic_member +\
self.context.sequence.effective_inputs(role=self.role_members_namebuilder())
subdirs = list()
for am in avail_members:
if am.rh.container.dirname not in subdirs:
subdirs.append(am.rh.container.dirname)
# Add a sytron member (only for child SurfexComponent but done here to test availability of deterministic mb)
if self.kind == "ensmeteo+sytron" and len(deterministic_member) == 1:
subdirs.append('mb036')
# Ca partait d'une bonne idee mais en pratique il y a plein de cas particuliers
# pour lesquels ca pose probleme : reanalyse safran, surfex postes, etc
# self.algoassert(len(set(subdirs)) == len(set([am.rh.provider.member for am in avail_members])))
# Crash if subdirs is an empty list (it means that there is not any input available)
# TODO : modifier en python3
self.algoassert(len(subdirs) >= 1)
return subdirs
def get_origin(self, rh, opts):
"""Get the status (essential or secondary) of the different members.
Ultimately:
* A failure on an essential member automatically leads to the
crash of the whole task (for example the deterministic member of
the PEARP-S2M operational chain)
* A failure on a secondary member is not automatically fatal
for the task (it depends on the number of secondary members that
failed)
Each member is associated to a specific input section (see method
:meth:`get_subdirs`) which is also a VORTEX Section object. This method
uses the optional 'source_conf' footprint attribute of this object to
descriminate the essential member(s) (with source_conf='4dvarfr'
whereas for standard PEARP members, source_conf='pearp'
in the case of the PEARP-S2M operational chain).
This method returns a list of boolean (with an order matching the one of
the different members) where a `True` value indicates the essential(s)
member(s).
In case there is no 'source_conf' attribute, all members are considered
as essential members.
"""
deterministic_member = self.context.sequence.effective_inputs(role=self.role_deterministic_namebuilder())
avail_members = deterministic_member +\
self.context.sequence.effective_inputs(role=self.role_members_namebuilder())
subdirs = list()
cpl_model = list()
for am in avail_members:
if am.rh.container.dirname not in subdirs:
subdirs.append(am.rh.container.dirname)
if hasattr(am.rh.resource, 'source_conf'):
cpl_model.append(am.rh.resource.source_conf == '4dvarfr')
else:
# If the origin of the guess is not given the execution is in
# 'deterministic' mode (monthly reanalysis)
cpl_model.append(True)
return cpl_model
def role_deterministic_namebuilder(self):
"""
Defines the role of the effective inputs to take as reference to define
the deterministic member.
"""
return 'Ebauche_Deterministic'
def role_members_namebuilder(self):
"""
Defines the role of the effective inputs to take as reference to define
the different members.
"""
return 'Ebauche'
[docs]
class S2MReanalysis(S2MComponent):
"""AlgoComponent that runs several SAFRAN reanalyses in parallel."""
_footprint = dict(
info = 'AlgoComponent that runs several executions in parallel.',
attr = dict(
execution = dict(
values = ['reanalysis'],
optional = False,
),
),
)
def role_ref_namebuilder(self):
return 'Observations'
def get_subdirs(self, rh, opts):
avail_members = self.context.sequence.effective_inputs(role=self.role_ref_namebuilder())
subdirs = [am.rh.container.dirname for am in avail_members]
return list(set(subdirs))
def get_list_seasons(self, rh, opts):
list_dates_begin_input = list()
list_dates_end_input = list()
datebegin_input = self.datebegin
if self.datebegin.month >= 8:
dateend_input = min(Date(self.datebegin.year + 1, 8, 1, 6, 0, 0), self.dateend)
else:
dateend_input = min(Date(self.datebegin.year, 8, 1, 6, 0, 0), self.dateend)
list_dates_begin_input.append(datebegin_input)
list_dates_end_input.append(dateend_input)
while dateend_input < self.dateend:
datebegin_input = dateend_input
dateend_input = min(datebegin_input.replace(year=datebegin_input.year + 1), self.dateend)
list_dates_begin_input.append(datebegin_input)
list_dates_end_input.append(dateend_input)
list_dates_begin_input.sort()
list_dates_end_input.sort()
return list_dates_begin_input, list_dates_end_input
def _default_common_instructions(self, rh, opts):
"""Create a common instruction dictionary that will be used by the workers."""
ddict = super(S2MComponent, self)._default_common_instructions(rh, opts)
for attribute in self.footprint_attributes:
if attribute not in ['datebegin', 'dateend']:
ddict[attribute] = getattr(self, attribute)
return ddict
def execute(self, rh, opts):
"""Loop on the various initial conditions provided."""
self._default_pre_execute(rh, opts)
# Update the common instructions
common_i = self._default_common_instructions(rh, opts)
# Note: The number of members and the name of the subdirectories could be
# auto-detected using the sequence
subdirs = self.get_subdirs(rh, opts)
deterministic = [True] * len(subdirs)
# WARNING : The current method implies that the different seasons directories are sorted
# One way to ensure that is to use the begin year as directory name.
subdirs.sort()
# list_dates_begin, list_dates_end = self.get_list_seasons(rh, opts)
# subdirs must be named yyyymmddhh_YYYYMMDDHH
list_dates_begin = [item.split('_')[0] for item in subdirs]
list_dates_end = [item.split('_')[1] for item in subdirs]
self._add_instructions(common_i, dict(subdir=subdirs,
datebegin=list_dates_begin,
dateend=list_dates_end,
deterministic=deterministic))
self._default_post_execute(rh, opts)
[docs]
class S2MReforecast(S2MComponent):
"""AlgoComponent that runs several SAFRAN reforecasts in parallel."""
_footprint = dict(
info = 'AlgoComponent that runs several executions in parallel.',
attr = dict(
execution = dict(
values = ['reforecast'],
optional = False,
),
datebegin = dict(
optional = True,
),
dateend = dict(
optional = True,
),
),
)
def _default_common_instructions(self, rh, opts):
"""Create a common instruction dictionary that will be used by the workers."""
ddict = super(S2MComponent, self)._default_common_instructions(rh, opts)
for attribute in self.footprint_attributes:
if attribute not in ['datebegin', 'dateend']:
ddict[attribute] = getattr(self, attribute)
return ddict
def execute(self, rh, opts):
"""Loop on the various initial conditions provided."""
self._default_pre_execute(rh, opts)
# Update the common instructions
common_i = self._default_common_instructions(rh, opts)
subdirs, list_dates_begin, list_dates_end = self.get_individual_instructions(rh, opts)
deterministic = [True] * len(subdirs) # Each simulation day is important
self._add_instructions(common_i, dict(subdir=subdirs,
datebegin=list_dates_begin,
dateend=list_dates_end,
deterministic=deterministic))
self._default_post_execute(rh, opts)
def get_individual_instructions(self, rh, opts):
avail_members = self.context.sequence.effective_inputs(role=self.role_members_namebuilder())
subdirs = list()
list_dates_begin = list()
list_dates_end = list()
for am in avail_members:
# am.rh.container.dirname = YYYYMMDDHH
for subdir in glob.glob(f'{am.rh.container.dirname}/mb*'):
if subdir not in subdirs:
subdirs.append(subdir)
list_dates_begin.append(am.rh.resource.datebegin)
list_dates_end.append(am.rh.resource.dateend)
return subdirs, list_dates_begin, list_dates_end
[docs]
@echecker.disabled_if_unavailable
class SurfexComponent(S2MComponent):
"""AlgoComponent that runs several SURFEX executions in parallel."""
_footprint = dict(
info = 'AlgoComponent that runs several executions in parallel.',
attr = dict(
kind = dict(
values = ['escroc', 'ensmeteo', 'ensmeteonodet', 'ensmeteo+sytron', 'croco', 'ensmeteo+escroc',
'prepareforcing']
),
# Usages:
# * ensmeteo: S2M oper and prosnow
# * ensmeteonodet: S2M reforecast and prosnow
# * ensmeteo+sytron: S2M oper
# * ensmeteo+escroc seems unsued
dateinit = dict(
info = "The initialization date if different from the starting date.",
type = Date,
optional = True,
default = '[datebegin]',
),
threshold = dict(
info = "Threshold to apply to the snow water equivalent (in kg/m2) each 1st August",
type = int,
optional = True,
default = -999
),
members = dict(
info = "The members that will be processed",
type = footprints.stdtypes.FPList,
optional = True, # WARNING : mandatory if kind in ['escroc', 'croco']
# TODO : faires des algos distincts pour gérer proprement l'aspect non-optionel pour les
# algos croco et escroc
),
subensemble = dict(
info = "Name of the escroc subensemble (define which physical options are used)",
values = ["E1", "E2", "Crocus", "E1tartes", "E1notartes", "E2open", "E2MIP", "E2tartes", "E2MIPtartes",
"E2B21", "E2MIPB21", "E1B21"],
optional = True,
),
geometry_in=dict(
info="Area information in case of an execution on a massif geometry",
type=footprints.stdtypes.FPList,
# S2M (SAFRAN) specific
),
geometry_out=dict(
info="The resource's massif geometry.",
type=str,
# S2M (SAFRAN) specific
),
daily = dict(
info = "If True, split simulations in daily runs",
type = bool,
optional = True,
default = False,
),
dailynamelist = dict(
info = "If daily is True, possibility to provide a list of namelists to change each day",
type = list,
optional = True,
default = [],
# Used in prosnow only
),
multidates = dict(
info = "If True, several dates allowed",
type = bool,
optional = True,
default = False,
values = [False]
# Reforecast-specific
),
startmbnode = dict(
info = 'first member rep of the node for example 1,41,81 etc.',
type = int,
optional = True,
default = 1,
# Croco-specific
# Cet argument est utilisé uniquement pour trouver les "membres" des simulations
# dans la méthode 'get_subdirs'.
# --> A remplacer par une méthode plus robuste
),
)
)
def execute(self, rh, opts):
"""Loop on the various initial conditions provided."""
self._default_pre_execute(rh, opts)
# Update the common instructions
common_i = self._default_common_instructions(rh, opts)
subdirs = self.get_subdirs(rh, opts)
if self.subensemble:
escroc = ESCROC_subensembles(self.subensemble, self.members)
physical_options = escroc.physical_options
snow_parameters = escroc.snow_parameters
self._add_instructions(common_i, dict(subdir=subdirs,
physical_options=physical_options,
snow_parameters=snow_parameters))
else:
self._add_instructions(common_i, dict(subdir=subdirs))
self._default_post_execute(rh, opts)
def get_subdirs(self, rh, opts):
if self.kind == "escroc":
subdirs = ['mb{:04d}'.format(m) for m in self.members]
elif self.kind == 'croco':
# La ligne qui suit est un moyen bien compliqué de renvoyer la liste des "subdirs" des membres à traiter
# lorsque cette liste diffère de self.conf.members (cas d'un tirage aléatoire de membres)
# --> A remplaçer par une méthode plus robuste (par exemple en assurant en amont que les membres tirés
# aléatoirement sont rangés dans des répertoires 1 à N).
subdirs = ['mb{:04d}'.format(m) for m in range(self.startmbnode, self.startmbnode + len(self.members))]
else:
subdirs = super().get_subdirs(rh, opts)
if len(self.geometry_in) > 1:
# In the case of a postes geometry, there are 3 effective inputs with forcing file role
# (They are concatenated)
# Therefore it is necessary to reduce subdirs to 1 single element for each member
subdirs = list(set(map(self.system.path.dirname, subdirs)))
return subdirs
def role_members_namebuilder(self):
return 'Forcing'
def role_deterministic_namebuilder(self):
return 'Forcing_Deterministic'
[docs]
@echecker.disabled_if_unavailable
class PrepareForcingComponent(_CenTaylorRun):
"""
AlgoComponent that prepares several forcing files in parallel (changes of geometry).
"""
_footprint = dict(
info = 'AlgoComponent that runs several executions in parallel.',
attr = dict(
kind = dict(
values = ['prepareforcing', 'extractforcing', 'shadowsforcing']
),
# Inutile (parallélisation sur les années) ?
datebegin = dict(
info = "The list of begin dates of the forcing files",
type = footprints.stdtypes.FPList,
),
# Inutile (parallélisation sur les années) ?
dateend = dict(
info = "The list of begin dates of the forcing files",
type = footprints.stdtypes.FPList,
),
geometry_in = dict(
info = "Area information in case of an execution on a massif geometry",
type = footprints.stdtypes.FPList,
optional = True,
default = None
),
geometry_out = dict(
info = "The resource's massif geometry.",
type = str,
optional = True,
default = None
),
reprod_info=dict(
info="Informations that must be stored in output files for reproductibility",
type=dict,
optional=True,
default=dict(),
)
)
)
def execute(self, rh, opts):
"""Loop on the various initial conditions provided."""
self._default_pre_execute(rh, opts)
# Update the common instructions
common_i = self._default_common_instructions(rh, opts)
# Note: The number of members and the name of the subdirectories could be
# auto-detected using the sequence
subdirs = self.get_subdirs(rh, opts)
self._add_instructions(common_i, dict(subdir=subdirs, datebegin=self.datebegin,
dateend=self.dateend))
self._default_post_execute(rh, opts)
[docs]
@echecker.disabled_if_unavailable
class SurfexComponentMultiDates(SurfexComponent):
"""AlgoComponent that runs several SURFEX in parallel (including several dates for reforecasts)."""
_footprint = dict(
info = 'AlgoComponent that runs several executions in parallel.',
attr = dict(
multidates = dict(
info = "If True, several dates allowed",
type = bool,
values = [True]
)
)
)
def get_dates(self, subdirs):
# Pour l'instant je fais le porc parce que le passage de dateend ne marche pas du tout
duration = Period(days=4)
listdatebegin = []
listdateend = []
for datebegin_str in subdirs:
datebegin = Date(datebegin_str.split('/')[0])
listdatebegin.append(datebegin)
listdateend.append(datebegin + duration)
return listdatebegin, listdateend
def execute(self, rh, opts):
"""Loop on the various initial conditions provided."""
self._default_pre_execute(rh, opts)
# Update the common instructions
common_i = self._default_common_instructions(rh, opts)
subdirs = self.get_subdirs(rh, opts)
listdatebegin, listdateend = self.get_dates(subdirs)
listdateinit = listdatebegin[:]
if self.subensemble:
escroc = ESCROC_subensembles(self.subensemble, self.members)
physical_options = escroc.physical_options
snow_parameters = escroc.snow_parameters
self._add_instructions(common_i, dict(subdir=subdirs,
datebegin=listdatebegin,
dateinit=listdateinit,
dateend=listdateend,
physical_options=physical_options,
snow_parameters=snow_parameters))
else:
self._add_instructions(common_i, dict(subdir=subdirs,
datebegin=listdatebegin,
dateinit=listdateinit,
dateend=listdateend))
self._default_post_execute(rh, opts)
def _default_common_instructions(self, rh, opts):
"""Create a common instruction dictionary that will be used by the workers."""
ddict = super()._default_common_instructions(rh, opts)
ddict.pop('datebegin')
ddict.pop('dateend')
ddict.pop('dateinit')
return ddict
[docs]
@echecker.disabled_if_unavailable
class ShadowsForcingWorker(PrepareForcingWorker):
"""
It only applies shadows to a forcing file without any change of geometry
(worker for 1 member).
"""
_footprint = dict(
info = 'Apply shadows',
attr = dict(
kind = dict(
values = ['shadowsforcing']
),
geometry_out = dict(
info="The resource's massif geometry.",
type=str,
optional = True,
default = None
),
datebegin=a_date,
dateend=a_date,
)
)
def _prepare_forcing_task(self, rundir, thisdir, rdict):
self.system.mv("FORCING.nc", "FORCING_OLD.nc")
forcinput_applymask(["FORCING_OLD.nc"], "FORCING.nc")
save_file_period(thisdir, "FORCING", self.datebegin, self.dateend)
return rdict