tactus.tasks package

Tasks submodule.

Submodules

tactus.tasks.addpert module

Addpert.

class Addpert(config)

Bases: Task

Addpert, add SLAF perturbation to unperturbed boundary file.

execute()

Run task.

Define run sequence.

tactus.tasks.archive module

ARCHIVEHOUR // ARCHIVESTATIC.

class ArchiveFDB(config)

Bases: ArchiveTask

Archving task for time dependent data dedicated for FDB.

class ArchiveHour(config)

Bases: ArchiveTask

Archving task for time dependent data.

class ArchiveMergedSQLites(config)

Bases: ArchiveTask

Archving task for time dependent data.

class ArchiveStatic(config)

Bases: ArchiveTask

Archving task for static data.

class ArchiveStaticMember(config)

Bases: ArchiveTask

Archving task for static data.

class ArchiveTask(config, datatype=None, include=None, exclude=None)

Bases: Task

Archving data.

execute()

Loops over archive choices.

tactus.tasks.base module

Base site class.

class ReferenceCheck(config)

Bases: Task

ReferenceCheck class.

class Task(config, name)

Bases: object

Base Task class.

archive_logs(files, target=None)

Archive files in a log directory.

Parameters:

• files (str,list) – File(s) to be archived

• target (str) – Target directory for archiving

change_to_wdir()

Change to task working dir.

create_wdir()

Create task working directory and check for unix group and set permissions.

create_wrkdir()

Create a cycle working directory.

execute()

Do nothing for base execute task.

get_binary(binary_name, task_name=None)

Determine binary path from task or system config section.

Parameters:

• binary_name (str) – Name of binary

• task_name (str, Optional) – Optional name for task specific setting

Returns:

full path to binary

Return type:

bindir (str)

get_task_setting(setting)

Get task setting.

Parameters:

setting (str) – Setting to find in task.{self.name}

Returns:

Found setting

Return type:

value

post(source=None, target=None)

Do default postfix.

E.g. clean

prep()

Do default preparation before execution.

E.g. clean

remove_wdir(wdir=None)

Remove working directory.

rename_wdir(prefix='Failed_task_', source=None, target=None)

Rename failed/completed working directory.

run()

Run task.

Define run sequence.

class UnitTest(config)

Bases: Task

Base Task class.

tactus.tasks.batch module

Batch process.

class BatchJob(rte, wrapper='')

Bases: object

Batch job.

run(cmd, logfile=None)

Run command.

Parameters:

• cmd (str) – Command to run.

• logfile (io.IOBase, str, optional) – Object to tee stdout into in addition to sys.stdout (e.g. open(“oops.log”, “a”)). If a string is given it will be used as a file name.

Raises:

• TypeError – Checks if cmd is of type str

• ValueError – If logfile is not a open file handle

• CalledProcessError – Execution error

tactus.tasks.bator module

Bator task — ODB subbase creation via the BATOR binary.

class Bator(config)

Bases: OdbIngestionTask

Run BATOR for one observation type to produce an ECMA ODB subbase.

tactus.tasks.boundaries module

Boundary interpolation.

class C903(config)

Bases: InterpolateBoundaries

E927.

class E927(config)

Bases: InterpolateBoundaries

E927.

class InterpolateBoundaries(config)

Bases: Task

Boundary interpolation task.

execute()

Run task.

Define run sequence.

tactus.tasks.c903light module

C903Light.

class C903Light(config)

Bases: Task

C903Light, just symlink the target file to one produced elsewhere.

execute()

Run task.

Define run sequence.

tactus.tasks.canari module

Canari.

class Canari(config)

Bases: Task

Canari task.

execute()

Execute task.

tactus.tasks.cleaning_tasks module

Clean tactus file systems.

class Cleaning(config)

Bases: Task

Interface class to the cleaning.

execute()

Run the cleaning.

tactus.tasks.collectlogs module

CollectLogs.

class CollectLogs(config, config_label='staticlogs')

Bases: Task

Collect task and scheduler logfiles and store them as tarfiles.

execute()

Execute collect logs .

scan_logs(tarlog, parent, target, pattern='', exclude='')

Search for files matching a pattern and add them to a tar file.

Parameters:

• tarlog (tarfile object) – Tar file used

• parent (str) – Top search directory

• target (str) – Main search directory

• pattern (str) – Optional search pattern

• exclude (str) – Optional string for files to be excluded

Raises:

FileNotFoundError – If parent directory does not exist

class CollectLogsHour(config)

Bases: CollectLogs

Collectlog task for static data.

class CollectLogsStatic(config)

Bases: CollectLogs

Collectlog task for static data.

tactus.tasks.compilation module

Compialtion tasks.

class IALClone(config)

Bases: Task

Bundle Fetch task.

execute()

Execute task.

class TactusBundleBuild(config)

Bases: Task

tactus bundle build.

execute()

Execute task.

get_install_subpath()

Build install subpath by using the location of the env.sh file.

The arch build directory (<bundle_dir>/source/arch/<arch>) may contain a default symlink pointing at the actual build used, e.g. one selected by ecbundle based on compiler/toolchain. When that symlink exists, it is resolved and the path components coming after the arch directory name are returned, giving the subpath under which the build was actually installed (e.g. <toolchain>/<build_type>). When there is no such symlink, the subpath is empty (.), meaning installs go directly under the arch directory.

Returns:

Subpath (relative to the arch directory) to append to

the install/build directories.

Return type:

Path

make_install_arch_symlink()

Creates a “default” symlink in the install tree.

Creates a symlink named default in the compilers installation folder to point to the full path of the compiler’s default architecture if the architecture provided was a symlink.

class TactusBundleCreate(config)

Bases: Task

tactus create bundle.

execute()

Execute task.

tactus.tasks.creategrib module

CreateGrib.

class CreateGrib(config)

Bases: GlGrib

Convert time dependent files.

class CreateGribStatic(config)

Bases: GlGrib

Convert static files.

class GlGrib(config, name='')

Bases: Task

Create grib files.

convert2grib(infile, outfile, filetype)

Convert FA to grib.

Namelist arguments are given in the creategrib.NAME config part per filetype :param infile: Input file :type infile: str :param outfile: Output file :type outfile: str :param filetype: filetype to look for in rules :type filetype: str

Raises:

FileNotFoundError – Could not find file

create_list(input_template, output_settings)

Create list of files to process.

execute()

Execute creategrib.

tactus.tasks.discover_task module

Discover tasks.

add_plugins_to_sys_path(reg: TactusPluginRegistry) → None

Add plugin root paths to sys.path so their task modules can be imported.

Each plugin whose tasks_path exists is inserted at the front of sys.path (if not already present), making the plugin’s packages importable by their fully-qualified module names.

Parameters:

reg (TactusPluginRegistry) – Registry of active tactus plugins.

available_tasks(reg: TactusPluginRegistry)

Create a list of available tasks.

Parameters:

reg (TactusPluginRegistry) – tactus plugin registry

Returns:

Task objects

Return type:

known_types (list)

create_task_index(config)

Create a task index file.

Parameters:

config (ConfigParse) – Config

Returns:

Dict of known tasks, and their location

Return type:

known_types(dict)

discover(package, base)

Discover task classes.

Plugin classes are discovered in a given namespace package, deriving from a given base class. The base class itself is ignored, as are classes imported from another module (based on cls.__module__). Each discovered class is identified by the class name by changing it to lowercase and stripping the name of the base class, if it appears as a suffix.

Parameters:

• package (types.ModuleType) – Namespace package containing the plugins

• base (type) – Base class for the plugins

Returns:

type): Discovered plugin classes

Return type:

(dict of str

discover_modules(package, what='plugin')

Discover plugin modules.

Parameters:

• package (types.ModuleType) – Namespace package containing the plugins

• what (str, optional) – String describing what is supposed to be discovered. Defaults to “plugin”.

Yields:

tuple – str: Name of the imported module types.ModuleType: The imported module

get_task(name, config) → Task

Create a tactus.tasks.Task object from configuration.

Parameters:

• name (_type_) – _description_

• config (_type_) – _description_

Returns:

The task object with name name. The task object has to be a subclass of Task to be retrievable.

Return type:

Task

Raises:

NotImplementedError – If task name is not amongst the known task names.

load_task_index(config)

Load a task index file.

Parameters:

config (ConfigParse) – Config

Returns:

Dict of known tasks, and their location

Return type:

known_types(dict)

tactus.tasks.e923 module

E923.

class E923(config, taskname=None)

Bases: Task

Methods for the e923 work.

constant_part(constant_file)

Run the constant part of e923.

Parameters:

constant_file – filename of the resulting file

execute_part(part_nr: int, month: int | None = None)

Generates namelist for E923 part and execute binary.

Parameters:

• part_nr – the number of the E923 part to execute

• month – (optional) month number

link_data_files(label: int | str, month: int | None = None, gunzip_flags: str | None = '') → List[str]

Link/copy input files for E923 part.

Parameters:

• label – dataset label in input_data (or integer to get related part)

• month – month number

• gunzip_flags – Flags passed to gunzip when decompressing compressed (*.Z) files

Returns:

List of files where links can be removed for cleanup

monthly_part(constant_file)

Run the monthly part of e923.

Parameters:

constant_file – filename of the input constant file

myexec(cmd, i)

Execute binary task.

Parameters:

• cmd (str) – Command to run

• i (int) – sequence number

remove_links(link)

Remove link.

Parameters:

link (list) – List of links to remove

class E923Constant(config)

Bases: E923

E923Constant task.

execute()

Run task.

Define run sequence.

class E923Monthly(config)

Bases: E923

E923Monthly task.

execute()

Run task.

Define run sequence.

class PgdUpdate(config)

Bases: Task

PgdUpdate.

Ensure consistency in orography between the atmosphere and surfex Read gridpoint orography from the atmospheric file and insert it in the PGD file Const.Clim.sfx.

execute()

Run task.

tactus.tasks.e923_update module

E923Update.

class E923Update(config)

Bases: Task

Methods for updating climfile for ALARO.

For mor information see doc/markdown/e923_update.md.

execute()

Run task.

Define run sequence.

tactus.tasks.forecast module

Forecast.

class FirstGuess(config)

Bases: Task

FirstGuess.

execute()

Find initial file.

class Forecast(config)

Bases: PySurfexBaseTask

Forecast task.

accelerator_device_input()

Copy the input files for GPU execution.

• parallel_method: input file with parallelisation technique for each algorithm

• synchost: input file defining optional device-to-host memory transfers

• select_gpu: wrapper file for sbatch, binding GPU to MPI rank

archive_output(filetype, periods)

Archive forecast model output.

Parameters:

• filetype (str) – Filename template

• periods (str) – Output list

execute()

Execute forecast.

merge_output(filetype, periods)

Merge distributed forecast model output.

Parameters:

• filetype (str) – File type (history, surfex, fullpos)

• periods (str) – Output list

Final result is the expected output file name in the working directory (as if there was no IO server). NOTE: This function has been replaced by io_merge tasks. It is only called if n_io_merge=0.

post()

Do special post for Foreast.

sst_sic_input(intp_bddir)

Link SST & SIC input file to forecast directory.

wfp_input()

Add wind turbine files to forecast directory.

class PrepareCycle(config)

Bases: Task

Task.

tactus.tasks.generatewfptabfile module

Task to generate location/turbine tab-files as input for windfarm parameterization.

class GenerateWfpTabFile(config)

Bases: Task

Generates TAB-file for WFP.

execute()

Run task.

Run sequence: define domain, call json-2-tab, store tab-output.

tactus.tasks.glbd module

GlBd.

class GlBd(config)

Bases: Task

Interpolate boundary files with GL.

execute()

Run task.

Define run sequence.

get_bdfile(bdfile_template, local_file, basetime, validtime, s3_path_template=None)

Helper function to possibly download a file via S3 before trying to find it.

Parameters:

• bdfile_template – str

• local_file – Path

• basetime – datetime

• validtime – datetime

• s3_path_template – str

tactus.tasks.gribmodify module

AddCalculatedFields.

class AddCalculatedFields(config)

Bases: Task

Create grib files.

add_field_to_grib(fnames, params, operation, output_params, layer_weights, physical_range)

Add fields to the same grib following specified modification.

Parameters:

• fnames (list) – list of grib files

• params (list) – list of parameter dictionaries of input parameters

• operation (str) – operation to compute new field

• output_params (dict) – list of parameter dictionaries of output parameters

• layer_weights (dict) – dict of level weights with level numbers as keys

• physical_range (list) – list of physical range for each parameter

Raises:

NotImplementedError – Operation not implemented yet”

calc_patch_averaging(params, layer_weights, values_list, bitmap_list, physical_range, nature_weighting)

Calculate patch averaging.

Bitmaps are used to mask out invalid values. This is necessary, as the values for a parameter may be missing in some tiles despite the fact that the tile fraction is not 0. This is solved by updating the layer weights to always sum to 1 while excluding missing data.

execute()

Execute gribmodify.

expand_input_grib_id(rules)

Expand input_grib_id entries into separate instances.

find_additional_files(additional_files, validtime)

Find multiple additional files.

find_in_files(param, fname, additional_files)

Check if field with specified parameters exists.

Checks the main file and all additional files.

Parameters:

• param – parameter dictionary,

• fname – main grib file

• additional_files – list of additional grib files

Returns:

True if field exists in any file

Return type:

bool

find_par(param, fname)

Check if field with specified parameters already exists in grib.

Parameters:

• param – parameter dictionary,

• fname – grib file

Returns:

True if field exists

Return type:

bool

get_unique_values(dicts, key, exclude=None)

Get unique values of a key in a list of dictionaries.

get_value_for_params(params, match_dict, values_list, bitmap_list, apply_bitmap=False, return_bitmap=False)

Get the index of the dict in the list that matches all key-value pairs.

safe_codes_get(gid, key)

Safely get the value for a key from a GRIB message.

Returning None if the key is not found.

set_while_retaining_time_unit(gid_res, key, value)

Set a key in a GRIB message while retaining the time unit.

tactus.tasks.interpolsstsic module

InterpolSstSic.

class InterpolSstSic(config)

Bases: Task

Interpolate SST/SIC from the host model to the model geometry.

Supported host models:

• IFS

execute()

Run task.

Run sequence: set input files, create gl namelist and run gl.

Raises:

NotImplementedError – If an unsupported SST model is used

tactus.tasks.iomerge module

IO merge task.

class IOmerge(config)

Bases: Task

IO merge task.

check_fc_path()

Check if fc_path exists.

Raises:

FileNotFoundError – If directory is not present

execute()

Execute IO merge.

merge_output(lt, filetype)

Merge distributed forecast model output.

Parameters:

• lt (timeDelta) – lead time

• filetype (str) – Type of file

static wait_for_file(filename, age_limit=15)

Wait until a file is at least N seconds old.

Parameters:

• filename (str) – full path to file

• age_limit (int) – minimum age of last modification

wait_for_io(filetype, lt, maxtries=2)

Wait for all io_server output to be stable.

Parameters:

• filetype (str) – kind of output file

• lt (timeDelta) – lead time

• maxtries (int) – maximum number of file search trials

Returns:

List of files expected

Return type:

file_list (list)

Raises:

RuntimeError – In case of erroneous number of files

tactus.tasks.marsprep module

Marsprep.

class Marsprep(config)

Bases: Task

Marsprep task.

execute()

Run task.

Define run sequence.

Raises:

RuntimeError – If there is an issue with the work folder.

get_bdmember_fetch_list(bddir: str, bdfile: str)

Builds the list of missing files to fetch from MARS.

get_geopotential_latlon(first_member, bdmember_list)

Get geopotential in lat/lon.

Parameters:

• first_member (int) – First bdmember in ensemble

• bdmember_list (List[int]) – Members to fetch data for

get_gg_data(tag: str, steps: List[int], members_dict: Dict[str, List[int]])

Get gridpoint surface data.

get_gg_snow_data(tag: str, steps: List[int], members_dict: Dict[str, List[int]], param)

Get soil gridpoint data.

get_gg_soil_data(tag: str, params: dict)

Get soil gridpoint data.

get_grid_point_surface_data()

Get grid point surface data.

get_grid_point_upper_air_data()

Get gridpoint upper air data.

get_lat_lon_data(bdmember_list: List[int])

Get Lat/Lon data.

Parameters:

bdmember_list (List[int]) – Members to fetch data for

get_lat_lon_sst_data(missing_steps_per_member: Dict[int, List[int]], members_dict: Dict[str, List[int]])

Get lat/lon data for sst.

get_sfx_data()

Get SFX data.

get_sh_data(tag: str, steps: List[int], members_dict: Dict[str, List[int]])

Get SH data.

get_sh_temperature_data(tag: str, steps: List[int], members_dict: Dict[str, List[int]], param)

Get soil gridpoint data.

get_shz_data(tag: str)

Get geopotential in spherical harmonics.

get_spectral_harmonic_data()

Get spectral harmonic data.

get_sst_data()

Get SST data.

get_ua_data(tag: str, steps: List[int], members_dict: Dict[str, List[int]])

Get upper air data.

property mars: dict

Get mars selection.

update_data_request(request: BaseRequest, prefetch: bool, specify_domain: bool, bdmember: List[int] | None = None, grid: str | None = None, source: str | None = None, fieldset: str | None = None)

Create ECMWF MARS system request.

Parameters:

• request – BaseRequest object to update

• prefetch – Retrieve or stage

• specify_domain – Use lat/lon and rotation or use global (default)

• bdmember – Boundary members to retrieve in case of eps.

• grid – Specific grid for some request. Default None.

• source – Sorce for retrieve data from disk. Defaults None.

• fieldset – Name of fieldset. Defaults None.

tactus.tasks.obs_ingestion module

Shared base for ODB observation-ingestion tasks (Bator, Obsconvert).

class OdbIngestionTask(config)

Bases: Task

Abstract base for tasks that run an ODB observation-ingestion binary.

Subclasses must set the four class attributes below.

execute()

Run the configured obs-ingestion binary for self.obstype.

tactus.tasks.obsconvert module

Obsconvert task — ODB subbase creation via the OBSCONVERT binary.

class Obsconvert(config)

Bases: OdbIngestionTask

Run OBSCONVERT for one observation type to produce an ECMA ODB subbase.

tactus.tasks.obsprep module

Observation preparation task.

Observation provider selection

da.obs_provider names the active provider convention. All provider definitions live in configuration under da.providers.<name>:

e.g. obs_provider = “LACE”

Multi-file merging

All candidates found in the archive are merged into a single file per obs type.

• BUFR / GRIB: files are concatenated byte-for-byte.

• OBSOUL: files are merged via obsoul_merge.pl (configured via da.obsoul_merge_script).

• NETCDF: only the first file is used; a warning is issued when more than one is found.

• HDF: No merging performed - radar files should be linked as site${i}

Temporal windowing

obs_step in the active provider block (minutes, default 0 = disabled) activates collection across multiple slots within the assimilation window. When enabled, ObsPrep computes the slots covered by [basetime + window_shift, basetime + window_shift + window_len] and searches each slot’s date directory.

Example: basetime 00 UTC, window_shift=-90 min, window_len=180 min → slots 23, 00, 01 should be searched (three different hours, possibly across two calendar dates).

class ObsPrep(config)

Bases: Task

Observation-preparation task.

Supported families as defined by da_stream: - surface stages only surface (SYNOP) observations for surface assimilation - upper_air stages all configured obs types for upper_air

DEFAULT_OBS_3DVAR = ['synop', 'gpssol', 'amdar', 'geowind', 'hrwind', 'temp', 'wp', 'seviri', 'amsua', 'amsub', 'iasi', 'ascat', 'radar']

DEFAULT_OBS_SURFACE = ['synop']

execute()

Stage observation files and write obstypes availability list.

For each obs type, collects all matching files from all archive slots within the assimilation window, merges them, and copies the result into the working directory. Writes obstypes_YYYYMMDDRR with the list of successfully staged types — used by the Bator tasks.

tactus.tasks.odbmerge module

ODB merge task — merges per-obstype ECMA subbases into a single ECMA database.

Uses the SHUFFLE binary and merge_ioassign.

class OdbMerge(config)

Bases: Task

Merge all successfully completed BATOR subbases into one ECMA ODB.

The merged ECMA database is archived to the DA scratch directory so that downstream tasks (surface or upper-air analysis) can access it.

execute()

Merge BATOR subbases and archive merged ECMA ODB.

tactus.tasks.pertana module

Pertana.

class Pertana(config)

Bases: Task

Pertana, add perturbations from the nesting file to the initial file.

execute()

Run task.

tactus.tasks.pertsurf module

PertSurf.

class PertSurf(config)

Bases: Task

Perturb surface parameters in SURFEX initial file.

execute()

Execute the PertSurf task.

tactus.tasks.prep_run module

Preparatory task for a tactus suite.

class PrepRun(config)

Bases: Task

Preparatory run/cleaning task.

create_famodeldefs(target_eccodes_definition_path: str)

Create faModelName.def in tactus_eccodes_path.

Parameters:

target_eccodes_definition_path (str) – Path to the where model name definitions should be created.

Raises:

• FileNotFoundError – If the source YAML file is not found.

• ValueError – If the YAML file does not contain the required keys.

• yaml.YAMLError – If the YAML file cannot be loaded.

execute()

Execute the task, including cleaning if enabled.

tactus.tasks.serial module

Serial task.

class Serial(config)

Bases: Task

Serial task.

execute()

Execute task.

tactus.tasks.sfx module

Surfex tasks.

class Pgd(config)

Bases: PySurfexBaseTask

Task.

execute()

Execute.

class PgdFilterTownFrac(config)

Bases: Task

PgdFilterTownFrac.

Reduce Town frac in Pgd file made with eccoclimapII.

execute()

Run task.

class Prep(config)

Bases: PySurfexBaseTask

Prep.

execute()

Execute.

class PySurfexBaseTask(config, name)

Bases: Task

Base task class for pysurfex tasks.

execute()

Execute.

tactus.tasks.sqlite module

ExtractSQLite.

class ExtractSQLite(config)

Bases: Task

Extract sqlite point files.

execute()

Execute ExtractSQLite on all output files.

class MergeSQLites(config)

Bases: Task

Extract sqlite point files.

execute()

Execute MergeSQLite on all member FC files.

tactus.tasks.toposoil module

Topography and SOILGRID data preparation tasks.

class Soil(config)

Bases: Task

Prepare soil data task for PGD.

static check_domain_validity(domain_properties: dict) → None

Check if domain is valid.

Parameters:

domain_properties (dict) – Dict with domain properties

Raises:

ValueError – If domain is outside soilgrid data area

static coordinates_for_cutting_dataset(domain_properties: dict, halo: float = 5.0) → tuple

Get coordinates for cutting dataset.

Parameters:

• domain_properties (dict) – Dict with domain properties

• halo (float) – Halo. Defaults to 5.0.

Returns:

Coordinates for cutting dataset

Return type:

tuple

execute()

Run task.

Define run sequence.

Raises:

FileNotFoundError – If no tif files are found.

static write_soil_header_file(header_file, soiltype, hdr_north, hdr_south, hdr_west, hdr_east, hdr_rows, hdr_cols, nodata=0, bits=8, write_fact=False, fact=10) → None

Write header file.

Parameters:

• header_file (str) – Header file

• soiltype (str) – Soil type

• hdr_north (float) – North

• hdr_south (float) – South

• hdr_west (float) – West

• hdr_east (float) – East

• hdr_rows (int) – Number of rows

• hdr_cols (int) – Number of columns

• nodata (int) – No data value. Defaults to 0.

• bits (int) – Number of bits. Defaults to 8.

• write_fact (bool) – Write factor. Defaults to False.

• fact (int) – Factor. Defaults to 10

class Topography(config)

Bases: Task

Topography data preparation.

define_gmted_input(domain_properties: dict) → tuple

Define GMTED input files.

Parameters:

domain_properties (dict) – Domain properties

Returns:

GMTED input files

Return type:

tuple

execute()

Run task.

Define run sequence.

gmted_header_coordinates(east: float, west: float, south: float, north: float) → tuple

Get GMTED header coordinates.

Parameters:

• east (float) – East

• west (float) – West

• south (float) – South

• north (float) – North

Returns:

Header coordinates

Return type:

tuple

static tif2bin(gd, bin_file) → None

Convert tif file to binary file used by surfex.

Parameters:

• gd – gdal dataset

• bin_file (str) – Binary file

static write_gmted_header_file(header_file, hdr_north, hdr_south, hdr_west, hdr_east, hdr_rows, hdr_cols) → None

Write header file.

Parameters:

• header_file (str) – Header file

• hdr_north (float) – North

• hdr_south (float) – South

• hdr_west (float) – West

• hdr_east (float) – East

• hdr_rows (int) – Number of rows

• hdr_cols (int) – Number of columns