# -*- coding: utf-8 -*-
"""
research_task_base.py
---------------------
Main class used for all CEN HPC research tasks.
.. inheritance-diagram:: vortex_cen.tasks.research_task_base
:top-classes: mkjob.nodes.Node
:private-bases:
:parts: 4
.. autoclass:: _CenResearchTask
:members:
:show-inheritance:
"""
import vortex
from mkjob.nodes import Task
from vortex_cen.tasks.oper_research_mixin import CENTaskMixIn
from bronx.stdtypes.date import Date
from footprints.stdtypes import FPDict
from footprints.util import rangex
from vortex_cen.tools.monitoring import InputReportContext, OutputReportContext
from vortex_cen.tools.monitoring import AlgoReportContext, TestReportContext
from snowtools.utils.dates import get_list_dates_files, get_dic_dateend
[docs]
class _CenResearchTask(Task, CENTaskMixIn):
"""
Abstract class defining the common sequence of actions for CEN's vortex tasks.
A vortex task is the sequence of actions to execute a single algo component.
It always follows this procedure (definied in the main `process` method) :
1. fetch all necessary input resources (files):
- from an archive machine --> on a transfert node ('early-fetch')
- from the local machine --> on a compute node ('fetch')
2. execute the algo component (an executable, a script or·a sequence of instructions)
of inputs resources necessary to run an algo component (an executable, a script or a
sequence of instructions) --> 'compute'
3. save output resources (files produced or modified by the algo component)
- to the local machine --> on a compute node ('backup')
- to an archive machine --> on a compute node ('late-backup')
To implement a new task inheriting from this abstract class, implement a subset of the following methods:
* get_remote_inputs
* get_local_inputs
* algo
* launch_algo
* put_outputs
See their respective documentation for more details.
Doc:
http://intra.cnrm.meteo.fr/algopy/trainings/vortex_dev2022_1/presentation/beamer/vortex_dev_jobs2_presentation.pdf
"""
def __init__(self, **kw):
"""
Initialise class attributes for dynamic documentation.
"""
super().__init__(**kw)
self.MANDATORY_CONFIGURATION_VARIABLES = [
"datebegin",
"dateend",
"xpid",
"geometry",
]
self.OPTIONAL_CONFIGURATION_VARIABLES = [
"date",
"test",
"localtest",
"debug",
"io_duration",
"namespace_out",
"diff",
"allow_path",
]
[docs]
def update_attributes(self, mandatory, optional, overwrite=None):
"""
Update class attributes for dynamic documentation
"""
if isinstance(overwrite, list):
for var in overwrite:
# Warning : "remove" removes only 1 element in case of duplicated values.
# However, there should never be duplicates in MANDATORY_CONFIGURATION_VARIABLES and
# OPTIONAL_CONFIGURATION_VARIABLES
if var in self.MANDATORY_CONFIGURATION_VARIABLES:
self.MANDATORY_CONFIGURATION_VARIABLES.remove(var)
if var in self.OPTIONAL_CONFIGURATION_VARIABLES:
self.OPTIONAL_CONFIGURATION_VARIABLES.remove(var)
self.MANDATORY_CONFIGURATION_VARIABLES.extend(
[x for x in mandatory if x not in self.MANDATORY_CONFIGURATION_VARIABLES]
)
self.OPTIONAL_CONFIGURATION_VARIABLES.extend(
[x for x in optional if x not in self.OPTIONAL_CONFIGURATION_VARIABLES]
)
[docs]
def defaults(self, extras):
"""
Set toolbox defaults, extended with actual arguments ``extras``.
:param extras: items to add to vortex.defaults.
:type extras: dict
"""
t = vortex.ticket()
if 'localtest' in self.conf:
vortex.active_now = False
vortex.defaults(
# namespace = self.conf.get('namespace', Namespace('vortex.multi.fr')),
# namespace = Namespace('vortex.multi.fr'),
# date = '[dateend]', # WARNING : research only
# TODO : the 'date' footprint is to be removed for research applications
# experiment = self.conf.xpid,
# geometry = self.conf.geometry,
# vapp = self.conf.vapp,
# vconf = '[geometry:tag]', # TODO : à modifier après changement de convention
# model = self.conf.model,
# namebuild = 'flat@cen', # WARNING : research only !
# nativefmt = 'netcdf',
)
for optk in ('cutoff', 'geometry', 'cycle', 'vortex_set_aside'):
if optk in self.conf:
value = self.conf.get(optk)
if isinstance(value, dict):
value = FPDict(value)
vortex.defaults[optk] = value
# Le nombre de process et de tâches peut être associé à la géométrie via un dictionnaire, on récupère
# maintenant la bonne valeur
# TODO : Sortir ce qui suit de research_task_base et essayer de simplifier
if 'ntasks' in self.conf:
if (isinstance(self.conf.ntasks, dict) and self.conf.geometry.tag in self.conf.ntasks.keys()):
self.conf.ntasks = self.conf.ntasks[self.conf.geometry.tag]
else:
# Default value from s2m.
# Maybe it would be better to crash and ask the user to set an explicit value ?
self.conf.ntasks = 80
if 'nprocs' in self.conf:
if (isinstance(self.conf.nprocs, dict) and self.conf.geometry.tag in self.conf.nprocs.keys()):
self.conf.nprocs = self.conf.nprocs[self.conf.geometry.tag]
else:
# No hyperthreading
self.conf.nprocs = self.conf.ntasks
if 'nnodes' in self.conf:
if (isinstance(self.conf.nnodes, dict) and self.conf.geometry.tag in self.conf.nnodes.keys()):
self.conf.nnodes = self.conf.nnodes[self.conf.geometry.tag]
else:
self.conf.nnodes = 1
# Format uenv properly : "uenv:{uenv_name}@user" in cas only {uenv_name} is provided
for key, value in self.conf.items():
if "uenv" in key:
if isinstance(value, bool):
continue
if ':' not in value:
value = f"uenv:{value}"
if '@' not in value:
value = f'{value}@{t.env()["USER"]}'
self.conf[key] = value
# Define a namespace_out variable to apply to all outputs set as the *namespace_out*
# configuration variable if provided by the user or 'vortex.multi.fr' by default
# self.namespace = self.namespace_out
vortex.defaults(**extras)
self.header('Toolbox defaults')
vortex.defaults.show()
@property
def namespace_out(self):
"""
Namespace for output toolboxes
"""
return self.conf.get('namespace_out', 'vortex.multi.fr')
[docs]
def force_configuration_variables(self):
"""
Implement this method to force the value of some configuration variables in specific use cases of the task.
In particular, if an input for the task comes from the output of a previous task in the driver, the "block" of
the input is imposed by the output block of the producing task.
"""
pass
@property
def allow_path(self):
return self.conf.get("allow_path", False)
@property
def debug(self):
"""
Enter 'debug' mode to preserve the working directory even after a succesfull execution.
**Associated configuration variable:**
* ``debug`` Enter 'debug' mode, default : False
type debug: bool
"""
if 'debug' in self.conf:
return self.conf.debug
else:
return False
[docs]
def preprocess(self):
"""
Pre-processing step to set usefull class variables.
**Associated (optional) configuration variables:**
* ``io_duration`` Argument similar to the one of the `get_list_dates_files` method in
snowtools/utils/dates.py. It is used to retrieve the list of *datebegin* and
*dateend* footprints for IO covering sub-periods.
Possible values : "yearly", "monthly" or "full"
type io_duration: str
"""
self.get_list_dates(duration=self.conf.get('io_duration', 'yearly'))
self.force_configuration_variables()
[docs]
def process(self):
"""
Main method definig the task's sequence of actions
"""
t = self.ticket
self.preprocess()
if 'early-fetch' in self.steps:
# In a multi step job (MTOOL, ...), this step will be run on a TRANSFER NODE.
# Consequently, data that may be missing from the local cache must be fetched here.
# e.g. GCO's genv, data from the mass archive system, ...
# Note: most of the data should be retrieved here since the use of transfer node is costless.
with InputReportContext(self, t):
self.get_remote_inputs()
if 'fetch' in self.steps:
# In a multi step job (MTOOL, ...), this step will be run, on a COMPUTE NODE,
# just before the beginning of computations. It is the appropriate place to fetch data produced
# by a previous task (the so-called previous task will have to use the 'backup' step
# in order to make such data available in the local cache).
with InputReportContext(self, t):
self.get_local_inputs()
if 'compute' in self.steps:
# The actual computations... (usually a call to the run method of an AlgoComponent)
# This is executed on a COMPUTE NODE.
with AlgoReportContext(self, t):
algo = self.algo()
if 'localtest' not in self.conf:
self.launch_algo(algo)
if 'backup' in self.steps or 'late-backup' in self.steps:
# In a multi step job (MTOOL, ...), this step will be run on a TRANSFER NODE.
# Consequently, most of the data should be archived here.
with OutputReportContext(self, t):
self.put_outputs()
if 'late-backup' in self.steps:
# Reproducibility check with reference output (retrieved from the archive on a transfer node only)
if 'test' in self.conf and 'localtest' not in self.conf:
with TestReportContext(self, t):
self.unittest()
elif 'diff_xpid' in self.conf:
self.diff()
if self.debug:
# Debug mode : make the job crash at the end to preserve the working directory
print('============================================================================')
print('============================================================================')
raise Exception('INFO :The execution went well, do not take into account the following error')
[docs]
def algo(self):
"""
Implement this method to call your task's algo component.
This method should return a valid AlgoComponent object.
"""
raise NotImplementedError("method 'algo' returning a valid AlgoComponent object should be"
"implemented in child class.")
[docs]
def launch_algo(self, algo, **kw):
"""
Implement this method in your task's algo component.
The implementation should define how to run the algo component, or call one of the standard
methods: `launch_MPI_executable()`, `launch_python_algo()`
:param algo: AlgoComponent object
:param kw:
"""
raise NotImplementedError("the method 'launch_algo' should be implemented in child class and might call "
"'launch_MPI_executable()' or 'launch_python_algo()' if appropriate.")
[docs]
def launch_MPI_executable(self, algo, mpiopts=None):
"""
Run executable with MPI.
:param algo: AlgoComponent object
:param mpiopts: dict with MPI options nnodes=..., nprocs=..., ntasks=...
"""
# Pour un exécution de binaire, il faut donner l'objet "exécutable" associé (récupéré par la commande
# vortex.executable(...))
# Il est possible de récupérer cet objet avec la ligne suivante :
executable = [tbx.rh for tbx in self.ticket.context.sequence.executables()]
# TODO : les valeurs de mpiopts sont définies par défaut dans la méthode "component_runner" de mkjob/nodes.py
# à partir des variables de configuration self.conf.nnodes, self.conf.ntasks, self.conf.nprocs
# --> Réfléchir à la pertinence de faire 2 méthodes "launch_MPI_executable" et "launch_executable" distinctes
self.component_runner(algo, executable, mpiopts=mpiopts)
[docs]
def launch_executable(self, algo):
"""
run executable without MPI.
:param algo: AlgoComponent object
"""
executable = [tbx.rh for tbx in self.ticket.context.sequence.executables()]
# Security : force following configuration variables to 1 because
# mkjob crashes in case of inconsistency
self.conf.nnodes = 1
self.conf.nprocs = 1
self.conf.ntasks = 1
self.component_runner(algo, executable, mpiopts=dict(nnodes=1, nprocs=1, ntasks=1))
[docs]
def launch_python_algo(self, algo, **kw):
"""
Run your task's algo component. For algo components consisting of python code.
:param algo: AlgoComponent object
:param kw: keyword arguments dict
"""
if algo is not None:
algo.run(**kw)
[docs]
def put_outputs(self):
"""
Implement this method in your task to save resources remotely (on Hendrix, sxcen,...) from a transfer node.
"""
# raise NotImplementedError()
pass
[docs]
def unittest(self):
"""
Implement this method in unittest tasks to monitor the test results.
"""
if 'diff_xpid' in self.conf and self.conf.diff_xpid:
self.diff()
[docs]
def diff(self):
"""
Implement this method in your task to compare output with a reference file.
"""
pass
[docs]
def get_list_dates(self, duration='yearly'):
"""
Get the list of datebegin/dateend corresponding to the different time periods covered by IO files
from the actual simulation's datebegin/dateend arguments.
:param duration: Time period covered by individual files.
:type duration: str
"""
if 'datebegin' in self.conf and 'dateend' in self.conf:
# Get FORCING input dates
self.list_dates_begin, list_dates_end, self.list_dates_begin_pro, self.list_dates_end_pro = \
get_list_dates_files(Date(self.conf.datebegin), Date(self.conf.dateend), duration)
self.dict_dates_end = get_dic_dateend(self.list_dates_begin, list_dates_end)
self.dict_dates_end_pro = get_dic_dateend(self.list_dates_begin_pro, self.list_dates_end_pro)
elif 'rundate' in self.conf: # Real-time only --> make a specific default class ?
# TODO : Trouver une meilleur solution que ce comportement implicite
self.conf.datebegin = self.conf.rundate
self.conf.dateend = self.conf.rundate
self.list_dates_begin = [self.conf.rundate]
self.dict_dates_end = {self.conf.rundate: self.conf.rundate}
elif 'date' in self.conf: # Real-time only --> make a specific default class ?
# TODO : Trouver une meilleur solution que ce comportement implicite
self.conf.datebegin = self.conf.date
self.conf.dateend = self.conf.date
self.list_dates_begin = [self.conf.date]
self.dict_dates_end = {self.conf.date: self.conf.date}
else:
# TODO
pass
[docs]
def get_list_members(self):
"""
Return the complete list of ensemble members from either
- the exact 'member' value (int) --> returns [member]
- the 'members' list (FPList or list) --> returns the list of members
- the number of members 'nmembers' (int) --> returns the list of 'nmembers' values starting from 1
"""
if self.conf.member is not None:
return rangex(self.conf.member)
elif 'members' in self.conf:
# members is the list of ensemble members, ex : range(35), '0-35-1', [1, 2, 3]
return rangex(self.conf.members)
elif 'nmembers' in self.conf:
# nmembers is the number of ensemble members (int)
return rangex(1, self.conf.nmembers)
[docs]
def get_forcing(self, localname='FORCING_[datebegin:ymdh]_[dateend:ymdh].nc', namespace='vortex.multi.fr',
fatal=True):
"""
Method to get meteorological forcing file(s) covering the simulation period.
Look for files covering sub-periods defined by the `io_duration` configuration variable (current values:
"monthly", "yearly", "full")
:param localname: *local* footprint (how to name the file in the working directory).
This is an algo/task-specific argument. Default name depends on the actual datebegin/dateend of each file.
WARNING : in case a unique value is provided the user should ensure that a single
file will be retrieved
:type localname: str
:param namespace: namespace for fetching the forcing files. Default: vortex.multi.fr
:type namespace: str
**Mandatory configuration variables:**
* ``forcing_datebegin`` *datebegin* footprint, default self.conf.datebegin
type forcing_datebegin: str, footprints.stdtypes.FPList
* ``forcing_dateend`` *dateend* footprint, default self.conf.dateend
type forcing_dateend: str, footprints.stdtypes.FPList
* ``forcing_xpid`` Experiment identifier, default self.conf.xpid
type forcing_xpid: str
* ``forcing_geometry`` *geometry* footprint, default self.conf.geometry
type forcing_geometry: str, footprints.stdtypes.FPList
* ``forcing_vapp`` *vapp* footprint, default self.conf.vapp
type forcing_vapp: str
* ``forcing_vconf`` *vconf* footprint, default self.conf.vconf
type forcing_vconf: str
* ``forcing_block`` *block* footprint, default "meteo"
type forcing_vconf: str
* ``forcing_namespace`` *namespace* footprint, default "vortex.multi.fr" (hendrix + local cache)
type forcing_namespace: str
**Optional configuration variables:**
* ``forcing_member`` *member* footprint, default None (or *member* if provided)
type forcing_member: int, footprints.stdtypes.FPList
* ``forcing_namebuild`` *namebuild* footprint, default "flat@cen" (will change soon)
type forcing_namebuild: str, values: "flat@cen", "date@cen", "date@std"
* ``forcing_intent`` *intent* footprint (local file permissions), default "in"
Possible values : "in" (read-only), "inout" (read-write)
type forcing_intent: str
* ``forcing_date`` *date* footprint corresponding to the rundate of the forcing.
Useful to get the forcing of the monthly reanalysis for example. default is *forcing_dateend*.
* ``forcing_source_app`` *source_app* footprint, default None
type forcing_source_app: str, footprints.stdtypes.FPList
* ``forcing_source_conf`` *source_conf* footprint, default None
type forcing_source_conf: str, footprints.stdtypes.FPList
* ``forcing_source`` Retrieve *source_app* and *source_conf* footrprints dictionnaries for S2M reanalysis
Possible values : 'era5', 'era40'
type forcing_source: str
* ``forcing_cutoff`` *cutoff* footprint (to be made optional for SurfaceIO objects), default None
type forcing_cutoff: str
* ``io_duration`` Argument similar to the one of the `get_list_dates_files` method in
snowtools/utils/dates.py.
Used to retrieve the list of *datebegin* and *dateend* for inputs covering sub-periods.
Possible values : "yearly", "monthly" or "full"
type io_duration: str
* ``forcing_vortex1`` Boolean to identify resources produced with vortex1 (filename without geometry)
type forcing_vortex1: bool
TODO : prévoir un mécanisme pour rendre des déclarer les arguments obligatoires / optionnels pour
chaque tâche (ex: member)
"""
t = self.ticket
forcing_datebegin = self.conf.get('forcing_datebegin', self.conf.get('datebegin', None))
forcing_dateend = self.conf.get('forcing_dateend', self.conf.get('dateend', None))
forcing_xpid = self.conf.get('forcing_xpid', self.conf.xpid)
forcing_user = self.conf.get('forcing_user', None)
forcing_vapp = self.conf.get('forcing_vapp', self.conf.vapp)
forcing_vconf = self.conf.get('forcing_vconf', self.conf.vconf)
forcing_block = self.conf.get('forcing_block', 'meteo')
forcing_member = self.conf.get('forcing_member', self.conf.get('member', None))
# forcing_geometry value may depend on the task's output 'geometry' value
if 'forcing_geometry' in self.conf:
if isinstance(self.conf.forcing_geometry, dict):
forcing_geometry = self.conf.forcing_geometry[self.conf.geometry.tag]
else:
forcing_geometry = self.conf.forcing_geometry
else:
forcing_geometry = self.conf.geometry
# Security : in case of an ensemble of forcing files, get the FORCING of each member in a
# separate directory to avoid overwrinting files.
if (isinstance(forcing_member, list) and len(forcing_member) > 1 and 'member' not in localname):
localname = f'mb[member%04d]/{localname}'
# TODO : modifier le namebuilder par defaut lorsque le nouveau incluant la
# géométrie sera disponible
forcing_namebuild = self.conf.get('forcing_namebuild', 'flat@cen')
forcing_intent = self.conf.get('forcing_intent', 'in')
# TODO : ne pas utiliser de source_app / source_conf à l'avenir
forcing_source_app = self.conf.get('forcing_source_app', None)
forcing_source_conf = self.conf.get('forcing_source_conf', None)
forcing_cutoff = self.conf.get('forcing_cutoff', None)
vortex1 = self.conf.get('forcing_vortex1', False)
duration = self.conf.get('io_duration', 'yearly')
list_dates_begin, list_dates_end, _, _ = get_list_dates_files(Date(forcing_datebegin),
Date(forcing_dateend), duration)
dict_dates_end = get_dic_dateend(list_dates_begin, list_dates_end)
# Verrue pour gérer les footprints *source_app* et *source_conf* de la réanalyse S2M
if 'forcing_source' in self.conf and forcing_source_app is None and forcing_source_conf is None:
if vortex1: # pour la rétro-compatibilité
forcing_source_app, forcing_source_conf = \
self.get_safran_sources(list_dates_begin, era5=self.conf.forcing_source == 'era5')
else:
forcing_source_app, forcing_source_conf = None, None
self.sh.title(f'Input forcing ({duration} duration)')
forcing = vortex.input(
role = 'Forcing', # Used for parallelisation and alternates only
kind = 'MeteorologicalForcing',
nativefmt = 'netcdf',
datebegin = list_dates_begin,
dateend = dict_dates_end,
experiment = forcing_xpid, # default : self.conf.xpid
username = forcing_user,
geometry = forcing_geometry, # default : self.conf.geometry
local = localname,
vapp = forcing_vapp, # default : self.conf.vapp
vconf = forcing_vconf, # default : self.conf.vconf
block = forcing_block, # default : 'meteo' ?
member = forcing_member, # default : None
intent = forcing_intent, # default : 'in' ?
namespace = namespace, # default : 'vortex.multi.fr',
namebuild = forcing_namebuild, # default recherche : 'flat@cen', defaut oper : None
vortex1 = vortex1,
date = self.conf.get('forcing_date', '[dateend]'), # pas supprimer! C'est necessaire pour reforecast.
source_app = forcing_source_app, # default = None (ne pas refaire l'erreur)
source_conf = forcing_source_conf, # default = None (ne pas refaire l'erreur)
cutoff = forcing_cutoff, #
fatal = fatal,
),
print(t.prompt, 'FORCING =', forcing)
print()