tactus package

Package to run the Destination Earth on Demand Extremes system.

class GeneralConstants(*_args, **_kwargs)[source]

Bases: QuasiConstant

General package-related constants.

PACKAGE_DIRECTORY = PosixPath('/home/runner/work/tactus/tactus/tactus')

PACKAGE_NAME = 'tactus'

VERSION = '1.2.0'

Subpackages

Submodules

tactus.archive module

Archive.

class Archive(config, datatype=None, include=None, exclude=None)[source]

Bases: object

Handle archiving.

archive(pattern, inpath, outpath, archive_type=None, newname=None, tarname=None)[source]

Send files to the file manager.

Parameters:

pattern (str,list) – string of list of patterns to search for
inpath (str) – Full path on the input archive
outpath (str) – relative path on the output archive
archive_type (str, optional) – Archive type. Defaults to None.
newname (str, optional) – Forces a rename of an identified file. Defaults to None.
tarname (str, optional) – Name of output tarname

Raises:

FileNotFoundError – If file not found
RuntimeError – Erroneous tarname

compress_files(pattern, inpath, outpath, tarname)[source]

Create tarfile.

Creates a tarfile for files matching pattern. If no files are found no tarfile is created and a warning is issued.

Parameters:

pattern (str,list) – string of list of patterns to search for
inpath (str) – Full path on the input archive
outpath (str) – relative path on the output archive
tarname (str, optional) – Name of output tarname

execute()[source]: Loops over archive choices.

tactus.argparse_wrapper module

Wrappers for argparse functionality.

add_expand_config(parser_object)[source]

Add object args.

Parameters:: parser_object (args oject) – args object to update

add_keep_def_file(parser_object, help_message='Keep suite definition file after submission')[source]

Add object args.

Parameters:

parser_object (args oject) – args object to update
help_message (str) – Help text

add_namelist_args(parser_object)[source]

Add namelist args.

Parameters:: parser_object (args oject) – args object to update
Returns:: updated args object
Return type:: parser_object (args oject)

get_args_parser(program_name='tactus')[source]

Build and return the argument parser.

Parameters:: program_name (str) – The name of the program.
Returns:: The configured argument parser.
Return type:: argparse.ArgumentParser

get_common_parser()[source]

Build and return the common argument parser shared by all subcommands.

Returns:

Parser with common arguments (config-file,: host-file, etc.).

Return type:

argparse.ArgumentParser

tactus.aux_types module

Aux types used in the package.

class BaseMapping(*args, **kwargs)[source]

Bases: Mapping

Immutable mapping that will serve as basis for all config-related classes.

copy(update: Mapping | Callable[[Mapping], Any] | None = None, validate=True)[source]: Return a copy of the instance, optionally updated according to update.

property data: Return the underlying data stored by the instance.

dict()[source]

Returns a dict deepcopy of the internal data.

Returns:: The dictionary representation

dumps(section='', style: Literal['toml', 'json', 'yaml'] = 'toml', toml_formatting_function: Callable | None = None)[source]: Get a nicely printed version of the container’s contents.

update(key: str, value)[source]: Return a copy of the instance, with updated key=value according to argument.

class QuasiConstant(*_args, **_kwargs)[source]

Bases: object

Inheriting from this will make the class’ attributes (almost) immutable.

class QuasiConstantMetaclass(*args, **kwargs)[source]

Bases: type

Metaclass to help making a class behave as if it had quasi-constant attributes.

dict()[source]: Return a dict form of the instance, with nested instances also converted.

property type_conversions: Type conversions to be performed on the attributes.

tactus.boundary_utils module

Boundary handling support classes.

class Boundary(config)[source]

Bases: object

Boundary class.

tactus.cleaning module

Clean tactus file systems.

class CleanTactus(config, defaults=None, basetime=None)[source]

Bases: object

Clean data.

clean()[source]: Perform the cleaning.

prep_cleaning(choices, basetime=None, dry_run=None)[source]

Prepare tasks for cleaning.

Parameters:

choices (dict) – Dict with cleaning settings
basetime (dateTime object) – Reference time
dry_run (boolean) – Flag for dry runs

wipe_ecfs(ecfs_path)[source]: Remove a full ecfs directory tree.

tactus.commands_functions module

Implement the package’s commands.

class RunTaskNamespace(**kwargs)[source]

Bases: Namespace

Namespace for the ‘run’ command.

members: List[int] | None = None

output: Path

tactus_home: str

task: str

task_job: Path

template_job: Path

troika: str

create_exp(args, config)[source]

Implement the ‘case’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

doc_config(args, config: ParsedConfig)[source]

Implement the ‘doc_config’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (ParsedConfig) – Parsed config file contents.

namelist_convert(args, config: ParsedConfig)[source]

Implement the ‘namelist convert’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

namelist_format(args, config: ParsedConfig)[source]

Implement the ‘namelist format’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

namelist_integrate(args, config)[source]

Implement the ‘namelist integrate’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

Raises:

SystemExit # noqa – DAR401

remove_cases(args, config)[source]: Remove output from cases.

replace_node(args, config)[source]

Implement the ‘replace’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

run_task(args: RunTaskNamespace, config: ParsedConfig)[source]

Implement the ‘run’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

show_config(args, config)[source]

Implement the ‘show_config’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

show_config_schema(args, config)[source]

Implement the show config-schema command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

show_host(args, config)[source]

Implement the show host command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

show_namelist(args, config)[source]

Implement the ‘show_namelist’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

show_paths(args, config)[source]: Implement the ‘show_paths’ command.

ssh_cmd(host, user, cmd)[source]

SSH to remote server and execute basic commands.

Parameters:

host – Host name or server IP address
user – Username (string) to login to server with
cmd – Command to be executed

Returns:

Message to notify if failed and why or successfully completed.

start_suite(args, config)[source]

Implement the ‘start suite’ command.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.
config (.config_parser.ParsedConfig) – Parsed config file contents.

Raises:

SystemExit – If error occurs while transferring files.

tactus.config_parser module

Registration and validation of options passed in the config file.

class BasicConfig(*args, _metadata=None, **kwargs)[source]

Bases: BaseMapping

Base class for configs. Arbitrary entries allowed: no validation is performed.

property data: Return the underlying data stored by the instance.

classmethod from_file(path, **kwargs)[source]

Retrieve configs from a file in miscellaneous formats.

Parameters:

path (Union[pathlib.Path, str]) – Path to the config file.
**kwargs – Arguments passed to the class constructor.

Returns:

Configs retrieved from the specified path.

Return type:

cls

get(item, default=None)[source]

Get dictionary stored at key as a BasicConfig.

Parameters:

item (str) – Key to retrieve (can be nested, e.g. “general.times”)
default – Value to return if key is not found

Returns:

data stored at key as a BasicConfig, or default if key not found

Return type:

BasicConfig

get_as_dict(item, default=None)[source]

Get dictionary stored at key as a dict.

Parameters:

item (str) – Key to retrieve (can be nested, e.g. “general.times”)
default – Value to return if key is not found

Returns:

data stored at key as a dict, or default if key ot found

Return type:

dict

property metadata: Get the metadata associated with the instance.

save_as(config_file)[source]

Save config file.

Parameters:: config_file (str) – Path to config file
Raises:: TypeError – when unknown filetype as config_file is given.

static save_dictionary_as(dictionary, config_file)[source]

Save config file.

Parameters:

dictionary – dictionary to save
config_file (str) – Path to config file

Raises:

TypeError – when unknown filetype as config_file is given.

exception ConfigFileValidationError[source]

Bases: Exception

Error to be raised when parsing the input config file fails.

class ConfigParserDefaults(*_args, **_kwargs)[source]

Bases: QuasiConstant

Defaults related to the parsing of config files.

CONFIG_DIRECTORY = PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files')

CONFIG_PATH = PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files/config.toml')

DATA_DIRECTORY = PosixPath('/home/runner/work/tactus/tactus/tactus/data')

MAIN_CONFIG_JSON_SCHEMA = {'$gribmodify_rules': {'csc_specific': {'description': 'The CSCs for which the specified conversion should happen.', 'items': {'type': 'string'}, 'title': 'CSC specific selection', 'type': 'array'}, 'minumum_frequency': {'description': 'Minimum output frequency expected', 'format': 'duration', 'title': 'Minimum frequency'}, 'output_name': {'description': 'Chosen name of output parameter.', 'title': 'Output parameter name', 'type': 'string'}}, '$output_setting_rules': {'anyOf': [{'maxLength': 0, 'type': 'string'}, {'format': 'duration', 'type': 'string'}, {'items': {'anyOf': [{'format': 'duration_slice', 'type': 'string'}]}, 'type': 'array'}]}, '$schema': 'http://json-schema.org/draft-07/schema#', 'additionalProperties': True, 'definitions': {'boundaries': {'description': 'Settings for control of boundary treatment.', 'properties': {'bd_has_surfex': {'description': 'Set to true if the host model runs with surfex', 'type': 'boolean'}, 'bdint': {'description': 'Boundary interval', 'format': 'duration', 'type': 'string'}, 'bdmodel': {'default': 'ifs', 'description': 'Coupling model', 'enum': ['lam', 'ifs'], 'type': 'string'}, 'bdmodel_sfx': {'default': 'from_grib', 'description': 'SURFEX prep input file', 'enum': ['from_grib', 'from_fa', 'from_nc'], 'type': 'string'}, 'bdshift': {'description': "Shift of boundary initial time. E.g. 'PT3H' would mean use 3H old boundaries", 'format': 'duration', 'type': 'string'}, 'bdshift_sfx': {'description': "Shift of SURFEX prep input initial time. E.g. 'PT24H' would mean use 24H old file", 'format': 'duration', 'type': 'string'}, 'do_slaf': {'description': 'Set to true if using the SLAF scheme for ensemble member boundary files', 'type': 'boolean'}, 'humi_gp': {'description': 'Value of TFP_Q%LLGP in domain namelist for c903. True for gridpoint humidity.', 'type': 'boolean'}, 'ifs': {'description': "Model for the 'general.times' section.", 'properties': {'bdmember': {'anyOf': [{'type': 'integer'}, {'type': 'string'}, {'type': 'array'}], 'default': [], 'description': "Specification of boundary members to nest in if selection is IFS. The order of bdmembers determines which member uses what bdmember. The bdmembers can be specified as integer, string slice or a list of integers. String slices have to be in the format 'start:end[:step]'. Use empty list for no bdmembers."}, 'selection': {'description': 'Selection', 'enum': ['atos_bologna_DT', 'lumi_DT', 'ATOS_DT', 'HRES', '@HOST@_DT', 'DT12', 'i5qp', 'i7u4', 'i7ye', 'i8hy', 'IFSENS', 'iit7', 'ilv7', 'is6g', 'irok', 'iyzp'], 'type': 'string'}}, 'title': 'IFS related settings', 'type': 'object'}, 'lam': {'description': 'LAM related boundary settings', 'properties': {'bdcycle': {'default': 'PT24H', 'description': 'lam boundary model cycle interval', 'pattern': '^P(?!$)(\\d+Y)?(\\d+M)?(\\d+W)?(\\d+D)?(T(?=\\d+[HMS])(\\d+H)?(\\d+M)?(\\d+S)?)?$', 'type': 'string'}, 'bdcycle_start': {'default': 'PT0H', 'description': 'lam boundary model cycle start hour', 'pattern': '^P(?!$)(\\d+Y)?(\\d+M)?(\\d+W)?(\\d+D)?(T(?=\\d+[HMS])(\\d+H)?(\\d+M)?(\\d+S)?)?$', 'type': 'string'}}, 'title': 'LAM related boundary settings', 'type': 'object'}, 'lbc_per_task': {'default': 7, 'description': 'Number of LBCs processed in a single task', 'type': 'number'}, 'max_interpolation_tasks': {'description': 'Number of interpolate tasks run in parallel', 'type': 'number'}, 'slafdiff': {'description': "SLAF difference of boundary initial times. E.g. 'PT6H' would mean a 6H difference", 'format': 'duration', 'type': 'string'}, 'slafk': {'description': 'SLAF difference scaling factor', 'type': 'number'}, 'slaflag': {'description': "SLAF shift of boundary initial time. E.g. 'PT3H' would mean use 3H old(er) boundaries", 'format': 'duration', 'type': 'string'}, 'sst_is_lsm': {'default': 'auto', 'description': 'Switch for using SST as LSM', 'enum': ['lsm', 'sst', 'auto'], 'type': 'string'}, 'sstmodels': {'description': 'Coupling model for SST & SIC', 'items': {'enum': ['IFS'], 'type': 'string'}, 'type': 'array'}}, 'title': 'Boundary Section', 'type': 'object'}, 'clean_old_data': {'decription': 'Option for cleaning old data', 'properties': {'cases_period': {'default': 'P7D', 'description': 'Age for old cases in days', 'pattern': '^P(?!$)(\\d+Y)?(\\d+M)?(\\d+W)?(\\d+D)?$', 'type': 'string'}, 'dw_case_prefixes': {'decription': 'List of prefixes of the daily runs', 'default': ['AUTOMATED_RUN', 'HARMONIE_AROME', 'AROME', 'ALARO'], 'type': 'array'}, 'force_delete_period': {'decription': "ISO-8601 duration string used to determine the fallback age of a suite for deletion if its files are missing. Defaults to approximately one year ('P365D').", 'default': 'P365D', 'pattern': '^P(?!$)(\\d+Y)?(\\d+M)?(\\d+W)?(\\d+D)?$', 'type': 'string'}, 'ignore': {'decription': 'File not to remove', 'default': [], 'type': 'array'}, 'scratch_ext': {'decription': 'Extention of scratch for operational user.', 'default': 'tactus', 'type': 'string'}}, 'title': 'Cleaning old data', 'type': 'object'}, 'collectlogs': {'description': 'Settings for control of log collection.', 'properties': {'hourlogs': {'description': 'Collection settings for time dependent tasks', 'properties': {'joboutdir': {'description': 'Top output dir to collect logs from ', 'type': 'string'}, 'tarname': {'description': 'Name of the resulting tar file', 'type': 'string'}, 'task_logs': {'description': 'Search path for task specific logs', 'type': 'string'}}, 'type': 'object'}, 'staticlogs': {'description': 'Collection settings for static tasks', 'properties': {'joboutdir': {'description': 'Top output dir to collect logs from ', 'type': 'string'}, 'tarname': {'description': 'Name of the resulting tar file', 'type': 'string'}, 'task_logs': {'description': 'Search path for task specific logs', 'type': 'string'}}, 'type': 'object'}}, 'title': 'CollectLogs Section', 'type': 'object'}, 'compile': {'properties': {'arch': {'default': 'source/ial-source/bundle/arch/ecmwf/hpc2020', 'description': 'Compiler configuration', 'type': 'string'}, 'bundle_file': {'default': '@CASEDIR@/IAL/bundle/bundle.yml', 'description': 'bundle used for compiling binaries', 'type': 'string'}, 'bundle_update': {'default': True, 'description': 'whether to update bundle_file with another bundle', 'type': 'boolean'}, 'cache': {'default': False, 'description': 'whether to use chache build/install folders', 'type': 'boolean'}, 'cache_dir': {'default': '@REFERENCE_DATA@/bundle_cache', 'description': 'whether to use chache build/install folders', 'type': 'string'}, 'clean_build': {'default': False, 'description': 'whether to clean build folder before compilation', 'type': 'boolean'}, 'dir': {'default': '@CASEDIR@/bundle', 'description': 'Directory where compilation will occur', 'type': 'string'}, 'git_token': {'default': '', 'description': 'Git token for IAL source access', 'type': 'string'}, 'ial_dir': {'default': '@CASEDIR@/IAL', 'description': 'Local path to IAL source dir', 'type': 'string'}, 'ial_git_branch': {'default': 'develop', 'description': 'Branch/tag', 'type': 'string'}, 'ial_git_repo': {'default': 'git@github.com:ACCORD-NWP/IAL', 'description': 'Git repo for IAL source', 'type': 'string'}, 'ninja': {'default': False, 'description': 'whether to use ninja for compilation', 'type': 'boolean'}, 'skip_build': {'default': False, 'description': 'whether to skip build step', 'type': 'boolean'}, 'update_bundle_file': {'default': '@TACTUS_HOME@/data/compilation/@CYCLE@/bundle.yml', 'description': 'bundle used for compiling binaries', 'type': 'string'}}, 'title': 'Compilation configuration', 'type': 'object'}, 'fullpos': {'description': 'Settings for control of fullpos selection.', 'properties': {'config_path': {'default': '@TACTUS_HOME@/data/namelist_generation_input/@CYCLE@/fullpos', 'title': 'Path to fullpos config files', 'type': 'string'}, 'domain_name': {'default': 'DOMAIN', 'description': 'Name of your domain as used by fullpos', 'maxLength': 20, 'type': 'string'}, 'main': {'default': ['rules', 'namfpc_header'], 'description': 'List of mandatory rules', 'items': {'type': 'string'}, 'type': 'array'}, 'selection': {'additionalProperties': {'items': {'type': 'string'}, 'type': 'array'}, 'description': 'List of selections to include', 'type': 'object'}}, 'title': 'Fullpos Section', 'type': 'object'}, 'general': {'description': "Model for the 'general' section.", 'properties': {'accept_static_namelists': {'description': 'Allow usage of static namelists as input for the tasks. The namelist should be located in the directory specified by platform.namelists and named with the convention namelist_{kind}_{task} where kind could be surfex or master and task forecast/pgd or similar. For the extraction of namelist see [namelists.md](../docs/namelists.md).', 'type': 'boolean'}, 'case': {'default': 'tactus_case', 'description': 'Experiment name', 'pattern': '^[^0-9][^.]*$', 'type': 'string'}, 'cnmexp': {'default': 'DEOD', 'description': 'Experiment short name', 'maxLength': 4, 'minLength': 4, 'type': 'string'}, 'csc': {'default': 'AROME', 'description': 'CSC', 'enum': ['AROME', 'ALARO', 'HARMONIE_AROME'], 'type': 'string'}, 'cycle': {'description': 'IAL cycle', 'enum': ['CY50t2', 'CY49t2', 'CY48t3', 'CY46h1'], 'type': 'string'}, 'e923_optional_sections': {'default': [], 'description': "Option to include/exclude part 5 and 6 of e923. Set to '[5,6]' to retain traditional climate files.", 'items': {'type': 'integer'}, 'type': 'array'}, 'famodel': {'description': 'faModelName for handling grib output settings. Needed to specify encodings of ensemble members as well as sub-hourly outputs. Do not replace default name.', 'type': 'string'}, 'fullpos_windpower': {'default': True, 'description': 'Switch for enabeling wind power production in fullpos output. Normally no need to provide but wind power output via full pos is not supported in CY48.', 'type': 'boolean'}, 'initfile': {'description': 'Initial file for the forecast object. Normally no need to provide unless there are special requirements or the forecast object is used in stand alone mode.', 'type': 'string'}, 'initfile_sfx': {'description': 'Initial surfex file for the forecast object. Normally no need to provide unless there are special requirements or the forecast object is used in stand alone mode.', 'type': 'string'}, 'keep_workdirs': {'description': 'Do not remove working directories', 'type': 'boolean'}, 'loglevel': {'description': 'Logger output level', 'enum': ['CRITICAL', 'ERROR', 'DEBUG', 'INFO', 'WARNING'], 'type': 'string'}, 'member': {'default': 0, 'description': 'Member number of a member in the ensemble', 'type': 'integer'}, 'member_str': {'default': 'mbr000', 'description': 'String identifier of a member in the ensemble', 'pattern': '^mbr\\d{3}$', 'type': 'string'}, 'namelist': {'description': 'Namelist settings. Various settings for the forecast namelist that have no other place to go.', 'properties': {'nrazts': {'description': 'Reset times for instantaneous fluxes ( e.g. min/max temperature )', 'type': 'string'}, 'radiation_frequency': {'description': 'Time interval for radiation compitations in the model.', 'type': 'string'}}, 'title': 'Namelist', 'type': 'object'}, 'output_settings': {'description': 'Output settings. Specified as duration for any of interval, endtime:interval, starttime:endtime:interval or a list of these options', 'properties': {'ddh_dl': {'$ref': '#/$output_setting_rules', 'description': 'Output list for limited area DDH files'}, 'ddh_gl': {'$ref': '#/$output_setting_rules', 'description': 'Output list for global DDH files'}, 'ddh_zo': {'$ref': '#/$output_setting_rules', 'description': 'Output list for zonal DDH files'}, 'fullpos': {'$ref': '#/$output_setting_rules', 'description': 'Output list for fullpos files'}, 'history': {'$ref': '#/$output_setting_rules', 'description': 'Output list for history files'}, 'surfex': {'$ref': '#/$output_setting_rules', 'description': 'Output list for surfex files'}}, 'title': 'Output Settings', 'type': 'object'}, 'remove_sections': {'default': [], 'description': 'Sections to remove prior to config merge', 'type': 'array'}, 'surfex': {'description': 'SURFEX switch', 'type': 'boolean'}, 'surfex_sea_ice': {'default': 'sice', 'description': 'Treatment of sea ice in surfex', 'enum': ['sice', 'none'], 'type': 'string'}, 'times': {'description': "Model for the 'general.times' section.", 'oneOf': [{'anyOf': [{'not': {'required': ['list']}, 'required': ['start']}, {'not': {'anyOf': [{'required': ['end']}, {'required': ['list']}, {'required': ['start']}]}, 'required': ['basetime']}]}, {'not': {'anyOf': [{'required': ['start']}, {'required': ['end']}]}, 'required': ['list']}], 'properties': {'basetime': {'description': 'Initial date and time for the current task.', 'format': 'date-time', 'type': 'string'}, 'cycle_length': {'default': 'PT3H', 'description': 'Cycle length defined as duration', 'format': 'duration', 'type': 'string'}, 'end': {'description': 'End time if the run covers several cycles. Date time format or duration relative start.', 'oneOf': [{'format': 'date-time'}, {'format': 'duration'}], 'type': 'string'}, 'forecast_range': {'description': 'Forecast range', 'format': 'duration', 'type': 'string'}, 'list': {'description': 'List of dates, placeholder for future needs.', 'items': {'format': 'date-time', 'type': 'string'}, 'type': 'array'}, 'start': {'description': "Start time, given as a date-time string 'YYYY-MM-DDTHH:mm:ssZ' or duration, e.g. '-P1D' for yesterday, which is relative to the current date at 00Z.", 'oneOf': [{'format': 'date-time'}, {'format': 'duration'}], 'type': 'string'}, 'validtime': {'description': 'Final date and time for the current task.', 'format': 'date-time', 'type': 'string'}}, 'title': 'Time-Related Entries', 'type': 'object'}, 'upd_sst_sic': {'default': False, 'description': 'Switch to update SST/SIC during forecast', 'type': 'boolean'}, 'windfarm': {'default': False, 'description': 'Switch for enabeling wind farm parameterisation', 'type': 'boolean'}}, 'required': ['times'], 'title': 'General Section', 'type': 'object'}, 'gribmodify': {'description': 'Rules for modifications to be made to the grib output', 'properties': {'conversions': {'default': [], 'description': 'List of file types to be converted', 'type': 'array'}, 'fullpos.1': {'description': 'Calculates total rain from components', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Total rain', 'type': 'object'}, 'fullpos.2': {'description': 'Calculates total snow from components', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Total snow', 'type': 'object'}, 'fullpos.3': {'description': 'Calculates total precipitation from components', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Total precpipation', 'type': 'object'}, 'fullpos.4': {'description': 'Calculates 10m wind speed', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': '10m wind speed', 'type': 'object'}, 'fullpos.5': {'description': 'Calculates 100m wind speed', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': '100m wind speed', 'type': 'object'}, 'fullpos.6': {'description': 'Calculates 10m gust wind speed', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': '10m gust wind speed', 'type': 'object'}, 'fullpos.7': {'description': 'Calculates 2m temperature based on Lawrence 2005, BAMS, doi: 10.1175/BAMS-86-2-225', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': '2m dewpoint temperature', 'type': 'object'}, 'gribmodify_rules_file': {'description': 'File specifying the gribmodify rules for the specified conversions', 'type': 'string'}, 'surfex.1': {'description': 'Calculates nature-weighted leaf area index', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Leaf area index', 'type': 'object'}, 'surfex.2': {'description': 'Calculates albedo', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Albedo', 'type': 'object'}, 'surfex.3': {'description': 'Calculates the nature-weighted patch-averaged soil moisture in the top 20 cm', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Top 20 cm average soil moisture', 'type': 'object'}, 'surfex.4': {'description': 'Calculates the nature-weighted patch-averaged soil moisture in the top 100 cm', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Top 100 cm average soil moisture', 'type': 'object'}, 'surfex.5': {'description': 'Calculates the patch-averaged soil temperature in the top 20 cm', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Top 20 cm average soil temperature', 'type': 'object'}, 'surfex.6': {'description': 'Calculates the patch-averaged soil temperature in the top 100 cm', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Top 100 cm soil temperature', 'type': 'object'}, 'surfex.7': {'description': 'Calculates root depth soil moisture', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Root depth soil moisture', 'type': 'object'}, 'surfex.8': {'description': 'Calculates the patch-averaged canopy water', 'properties': {'$ref': '#/$gribmodify_rules'}, 'title': 'Canopy water', 'type': 'object'}}, 'title': 'Section for modify grib', 'type': 'object'}, 'include': {'description': "Sections to be included from other config files. Specified as 'key = val' pairs, where the keys are the section names and the vals are the paths to the files containing them.", 'properties': {}, 'title': 'Include Section', 'type': 'object'}, 'perturbations': {'description': 'Settings for control of perturbations.', 'properties': {'pertana': {'default': False, 'title': 'Perturbation of initial atmospheric conditions with boundary data.', 'type': 'boolean'}, 'pertsurf': {'default': False, 'title': 'Perturbation of initial surfex conditions', 'type': 'boolean'}}, 'title': 'Perturbation Section', 'type': 'object'}, 'pgd': {'description': 'Settings for control of Pgd selection.', 'properties': {'npatch': {'default': 3, 'title': 'Number of patches for surfex variables', 'type': 'integer'}, 'one_decade': {'default': True, 'title': 'Option for using one decade', 'type': 'boolean'}, 'use_osm': {'default': False, 'title': 'Option to use Open Street map input data', 'type': 'boolean'}}, 'title': 'Pgd Section', 'type': 'object'}, 'plugin_registry': {'properties': {'plugins': {'title': 'Loaded plugins', 'type': 'object'}}, 'title': 'Plug-in registry', 'type': 'object'}, 'suite_control': {'description': 'Suite control options', 'properties': {'compile': {'default': False, 'description': 'Compile IAL source code', 'type': 'boolean'}, 'create_static_data': {'description': 'Activate the generation of PGD and monthly climate files in the suite', 'type': 'boolean'}, 'create_time_dependent_suite': {'description': 'Activate the time dependent part of the suite', 'type': 'boolean'}, 'do_addcalculatedfields': {'description': 'Activates additional GRIB postprocessing', 'type': 'boolean'}, 'do_cleaning': {'default': False, 'description': 'Do file cleaning', 'type': 'boolean'}, 'do_creategrib_static': {'default': False, 'description': 'Activate grib conversion for static files', 'type': 'boolean'}, 'do_extractsqlite': {'description': 'Activates extraction of selected points to SQLite files for verification.', 'type': 'boolean'}, 'do_interpolsstsic': {'description': 'Activates the SST & SIC boundary interpolation', 'type': 'boolean'}, 'do_marsprep': {'description': 'Run marsprep or skip extraction', 'type': 'boolean'}, 'do_mergesqlite': {'description': 'Activates merging of extracted sqlite files. Requires do_extractsqlite to be true, and number of ensemble members to be greater than 1, to have an effect.', 'type': 'boolean'}, 'do_pgd': {'description': 'Activate generation of a PGD file', 'type': 'boolean'}, 'do_soil': {'description': 'Activate preparation of SOILGRID data as input for PGD', 'type': 'boolean'}, 'interpolate_boundaries': {'description': 'Activates the boundary interpolation', 'type': 'boolean'}, 'max_static_data_tasks': {'default': 10, 'description': 'Limit for max number of monthly tasks, run at same time', 'type': 'integer'}, 'member_specific_mars_prep': {'default': False, 'description': 'Whether to do member specific marsprep (true) or common marsprep', 'type': 'boolean'}, 'member_specific_static_data': {'default': False, 'description': 'Whether to create member specific static data (true) or not', 'type': 'boolean'}, 'mirror_globalDT': {'default': False, 'description': 'Mirror globalDT suite (true) and trigger if globalDT is available', 'type': 'boolean'}, 'mirror_host_case': {'default': False, 'description': 'Mirror a host suite and trigger if available.', 'type': 'boolean'}, 'mirror_offline': {'default': False, 'description': 'Mirror offlineSFX suite and trigger if offline is available', 'type': 'boolean'}, 'mirror_suite': {'default': False, 'description': 'Mirror system.host_suite and trigger when the PostMortem task is completed', 'type': 'boolean'}, 'mode': {'description': 'Choose a strategy to achieve initial/initial_sfx files (start: start from interpolated files and then propagate the FG between cycles, restart: start from a previous run and then propagate the FG between cycles, cold_start: always start from interpolated files).', 'enum': ['start', 'restart', 'cold_start'], 'type': 'string'}, 'n_addcalculatedfields': {'default': 1, 'description': 'Number of AddCalculatedFields tasks', 'type': 'integer'}, 'n_creategrib': {'default': 1, 'description': 'Number of CreateGrib tasks', 'type': 'integer'}, 'n_io_merge': {'default': 1, 'description': 'Number of IOmerge tasks', 'type': 'integer'}, 'split_mars': {'default': False, 'description': 'Do marsprep only for one step before c903', 'type': 'boolean'}, 'suite_definition': {'default': 'TactusSuiteDefinition', 'description': 'Suite definition to play in EcFlow', 'type': 'string'}}, 'title': 'Suite control', 'type': 'object'}, 'system': {'description': 'TACTUS system specific path settings', 'properties': {'archive': {'default': '@ARCHIVE_ROOT@/@ARCHIVE_TIMESTAMP@/@MEMBER_STR@', 'description': 'Absolute path of date specific archive', 'type': 'string'}, 'archive_timestamp': {'default': '@YYYY@/@MM@/@DD@/@HH@/', 'description': 'Relative path of date specific archive', 'type': 'string'}, 'bdclimdir': {'description': 'Location of host model climate files used for e927/c903', 'type': 'string'}, 'bddir': {'description': 'Location of input boundaries from e.g. MARS', 'type': 'string'}, 'bddir_sfx': {'description': 'Location of input intial file for surfex', 'type': 'string'}, 'bdfile_sfx_template': {'description': 'Name template for input initial surfex file for prep', 'type': 'string'}, 'bdfile_sst_template': {'description': 'Name template for input SST & SIC files of the host model', 'type': 'string'}, 'bdfile_template': {'description': 'Name template for input boundary files', 'type': 'string'}, 'bindir': {'description': 'Location of binaries', 'type': 'string'}, 'c903_input_definition': {'default': '@CYCLE@/c903.json', 'description': 'Path to the c903 json config file', 'type': 'string'}, 'climdir': {'description': 'Location of generated/used climate files (PGD/monthly files)', 'type': 'string'}, 'e923_input_definition': {'default': '@CYCLE@/e923.json', 'description': 'Path to the e923 json config file', 'type': 'string'}, 'e927_input_definition': {'default': '@CYCLE@/e927.json', 'description': 'Path to the e927 json config file', 'type': 'string'}, 'forecast_dir_link': {'default': '../Forecast', 'description': 'Name of the softlink pointing to the Forecast task work directory', 'type': 'string'}, 'forecast_input_definition': {'default': '@CYCLE@/forecast.json', 'description': 'Path to the forecast json config file', 'type': 'string'}, 'fullpos_config_file': {'description': 'Path to the fullpos yml config file', 'type': 'string'}, 'global_sfcdir': {'description': 'Path to SFC files needed for c903, which are not in MARS', 'type': 'string'}, 'intp_bddir': {'default': '@WRK@', 'description': 'Location of boundaries on the current domain, i.e. the output directory of e927', 'type': 'string'}, 'intp_bddir_sfx': {'default': '@ARCHIVE@', 'description': 'Location of the inital surfex file for the current domain, i.e. the output directory of Prep', 'type': 'string'}, 'logs': {'description': 'Location of collected logfiles', 'type': 'string'}, 'namelists': {'default': '@TACTUS_HOME@/data/namelists/@CYCLE@', 'description': 'Path to static namelists', 'type': 'string'}, 'prev_case': {'default': '@CASE@', 'description': 'Name of the previous experiment', 'type': 'string'}, 'sfx_input_definition': {'default': '@CYCLE@/sfx.json', 'description': 'Path to the surfex json config file', 'type': 'string'}, 'sstfile_template': {'description': 'Name template for SST & SIC input boundary files interpolated to the model grid (currently hard-coded in model code)', 'type': 'string'}, 'sys_name': {'default': 'tactus', 'description': 'System/package name (typically matching PACKAGE_NAME) used when resolving paths.', 'type': 'string'}, 'wrk': {'default': '@CASEDIR@/@YYYY@@MM@@DD@_@HH@@mm@', 'description': 'Location of date specific working directory', 'type': 'string'}}, 'title': 'System', 'type': 'object'}, 'troika': {'description': 'TACTUS system specific path settings', 'properties': {'config_file': {'default': '@TACTUS_HOME@/data/config_files/troika.yml', 'description': 'Troika config file', 'type': 'string'}, 'troika': {'default': 'troika', 'description': 'Troika command, allows to specify troika with full path', 'type': 'string'}}, 'title': 'Troika', 'type': 'object'}}, 'description': "Model for validation of `tactus`'s main config file.", 'properties': {'boundaries': {'$ref': '#/definitions/boundaries'}, 'clean_old_data': {'$ref': '#/definitions/clean_old_data'}, 'collectlogs': {'$ref': '#/definitions/collectlogs'}, 'compile': {'$ref': '#/definitions/compile'}, 'fullpos': {'$ref': '#/definitions/fullpos'}, 'general': {'$ref': '#/definitions/general'}, 'gribmodify': {'$ref': '#/definitions/gribmodify'}, 'include': {'$ref': '#/definitions/include'}, 'perturbations': {'$ref': '#/definitions/perturbations'}, 'pgd': {'$ref': '#/definitions/pgd'}, 'suite_control': {'$ref': '#/definitions/suite_control'}, 'system': {'$ref': '#/definitions/system'}, 'troika': {'$ref': '#/definitions/troika'}}, 'required': ['general'], 'title': 'Config', 'type': 'object'}

MAIN_CONFIG_JSON_SCHEMA_PATH = PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files/config_file_schemas/main_config_schema.json')

PACKAGE_CONFIG_PATH = PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files/config.toml')

PACKAGE_INCLUDE_DIR = PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files/include')

SCHEMAS_DIRECTORY = PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files/config_file_schemas')

class ConfigPaths[source]

Bases: object

Support multiple path search.

CONFIG_DATA_SEARCHPATHS = [PosixPath('/home/runner/work/tactus/tactus/tactus/data')]

SCHEMAS_SEARCHPATHS = [PosixPath('/home/runner/work/tactus/tactus/tactus/data/config_files/config_file_schemas')]

erroneous_paths = []

erroneous_schema_paths = []

static path_from_subpath(subpath) → Path[source]

Interface to find full path given any subpath, by searching ‘searchpaths’.

Parameters:: subpath (str) – Subpath to search for
Returns:: Full path to target
Return type:: (Path)
Raises:: RuntimeRerror – Various errors

static print(config_file=None, host=None)[source]

Prints the available config directories.

Displays the main config search paths as defined by list_paths in addition to the actual search paths in the config file used.

exception ConflictingValidationSchemasError[source]

Bases: Exception

Error to be raised when more than one schema is defined for a config section.

class JsonSchema(*args, **kwargs)[source]

Bases: BaseMapping

Class to use for JSON schemas. Provides a validate method to validate data.

get_markdown_doc()[source]: Return human-readable doc for the schema in markdown format.

validate(data)[source]: Return a copy of data validated against the stored JSON schema.

class ParsedConfig(*_, **__)[source]

Bases: BasicConfig

Object that holds parsed configs validated against a json_schema.

property data: Return the underlying data stored by the instance.

expand_macros(expand_all=False, protect_time=False)[source]

Expand macros in config recursively.

Parameters:

expand_all (boolean) – Flag to expand all macros
protect_time (boolean) – Flag to control expansion of time variables

Returns:

Parsed configuration

Return type:

config (ParsedConfig)

classmethod from_file(path, include_dir=None, **kwargs)[source]: Do as in BasicConfig. If None, include_dir will become path.parent.

property include_dir: Return the search dir used sections in the raw config’s include section.

property json_schema: Return the instance’s JSON schema.

evaluate_dynamic_dates(config_data)[source]: Replace dynamic dates like ‘yesterday’ with actual timestamps.

tactus.csc_actions module

Handle tstep selection.

class SelectTstep[source]

Bases: object

Class for the tstep selection.

tactus.custom_validators module

Define custom validators for Pydantic models.

Exports:: validate_durations: Validate the format of a duration string against ISO8601. import_from_string: Validate if a string can be imported and return the imported object.

import_from_string(value: str) → ModuleType | Type[source]

Validate if a string can be imported and return the imported object.

Only supports importing objects specified with full module path, e.g. module.submodule.object.

Parameters:

value – The string to import.

Returns:

The imported object.

Raises:

ImportError – If the module or object cannot be imported.
TypeError – If the input is not a string.

tactus.datetime_utils module

Implement helper routines to deal with dates and times.

class DatetimeConstants(*_args, **_kwargs)[source]

Bases: QuasiConstant

Datetime-related constants.

DEFAULT_SHIFT = datetime.timedelta(0)

ISO_8601_TIME_DURATION_REGEX = '^P(?!$)(\\d+Y)?(\\d+M)?(\\d+W)?(\\d+D)?(T(?=\\d+[HMS])(\\d+H)?(\\d+M)?(\\d+S)?)?$'

as_datetime(obj)[source]: Convert obj to string, parse into datetime and add UTC timezone iff naive.

as_julian(yyyymmdd)[source]

Convert a date in YYYYMMDD format to a Julian Day Number.

Parameters:: yyyymmdd (int or str) – Date in YYYYMMDD format.
Returns:: Julian Day Number corresponding to the given date at midnight.
Return type:: int

as_timedelta(obj) → timedelta[source]: Convert obj to string and parse as duration.

check_syntax(output_settings: Tuple[str] | List[str], length: int)[source]

Check syntax of output_settings.

Parameters:

output_settings (Union[Tuple[str], List[str]]) – Specifies the output steps
length (integer) – length to check on

Raises:

SystemExit – General system handler

cycle_offset(basetime, bdcycle, bdcycle_start=datetime.timedelta(0), bdshift=datetime.timedelta(0))[source]

Calculcate offset from a reference time.

Parameters:

basetime (datetime) – Reference time
bdcycle (timedelta) – Interval between cycles
bdcycle_start (timedelta) – Time of day when bdcycle starts
bdshift (timedelta) – shift of boundary usage

Returns:

a timdelta object of the offset

Return type:

timedelta

dt2str(dt)[source]

Convert timdelta object to file name suitable string.

Parameters:

dt (timedelta object) – duration

Returns:

string representation of duration: suitable for FA files

Return type:

duration (str)

evaluate_date(date: str, reference_date=None) → str[source]

Parses an ISO 8601 datetime and/or duration and returns the computed datetime.

If the input is a datetime (e.g., “2025-03-19T00:00:00Z”), it returns it as-is.
If the input includes a duration (e.g., “2025-03-19T00:00:00Z/-P1D”), it applies the duration to the datetime.
If the input is only a duration (e.g., “-P1D”), it applies it to today’s midnight (UTC).

Parameters:

date (str) – An ISO 8601 datetime, duration, or both.
reference_date (str) – An ISO 8601 datetime, duration, or both.

Returns:

The computed datetime in ISO format.

Return type:

str

expand_output_settings(output_settings: str | Tuple[str] | List[str], forecast_range: str) → List[List[timedelta]][source]

Expand the output_settings coming from config.

Parameters:

output_settings (Union[str, Tuple[str], List[str]]) – Specifies the output steps
forecast_range (str) – Forecast range in duration syntax

Returns:

List of output subsections.: Can be empty in case of empty output_settings

Return type:

sections (List[List[timedelta]])

Raises:

RuntimeError – Handle erroneous time increment

get_decadal_list(dt_start, dt_end) → list[source]: Return a list of dates for which decadal pgd files have to be created.

get_decade(dt) → str[source]: Return the decade given a datetime object.

get_month_list(start, end) → list[source]: Get list of months between two given dates (input as string).

oi2dt_list(output_settings: str | Tuple[str] | List[str], forecast_range: str) → List[timedelta][source]

Build list of output occurences.

Parameters:

output_settings (Union[str, Tuple[str], List[str]]) – Specifies the output steps
forecast_range (str) – Forecast range in duration syntax

Returns:

Sorted list of output occurences

Return type:

dt (List[timedelta])

tactus.derived_variables module

Derive runtime variables.

check_fullpos_namelist(config, nlgen)[source]

Find existing fullpos select files or generate them.

Parameters:

config (tactus.ParsedConfig) – Configuration
nlgen (dict) – master forecast namelist

Returns:

Possibly updated forecast namelist

Return type:

nlgen (dict)

derived_variables(config, processor_layout=None)[source]

Derive some variables required in the namelists.

Parameters:

config (tactus.ParsedConfig) – Configuration
processor_layout (ProcessorLayout, optional) – Processor layout object

Returns:

Derived config update

Return type:

update (dict)

Raises:

NotImplementedError – For configurations checking

set_times(config)[source]

Set basetime/validtime if not present.

Parameters:: config (.config_parser.ParsedConfig) – Parsed config file contents.
Raises:: ValueError – If start > end
Returns:: Dict of corrected basetime/validtime
Return type:: update (dict)

tactus.domain_utils module

Utility for domain handling.

get_domain(config) → Dict[str, Any][source]

Read and return domain data.

Parameters:: config (tactus.ParsedConfig) – Configuration from which we get the domain data
Returns:: Dictionary containing the domain

tactus.experiment module

Experiment tools.

class EPSExp(config: ParsedConfig)[source]

Bases: Exp

Setup EPS experiment from config file.

setup_exp(host=None)[source]

Generate EPS member settings and update config file.

Only deviations of member settings from general EPS settings are added to the config.

class Exp(config: ParsedConfig, merged_config: dict | None)[source]

Bases: object

Experiment class.

class ExpFromFiles(config: ParsedConfig, exp_dependencies, mod_files: List[Path], host=None, merged_config=None)[source]

Bases: Exp

Generate Exp object from existing files. Use config files from a setup.

static deep_update(source, overrides)[source]

Update a nested dictionary or similar mapping.

Modify source in place.

Parameters:

source (dict) – Source
overrides (dict) – Updates

Returns:

Updated dictionary

Return type:

dict

static setup_files(output_file, case=None)[source]

Set up the files for an experiment.

Parameters:

output_file (str) – Output file
case (str, optional) – Experiment name. Defaults to None.

Returns:

Experiment dependencies from setup.

Return type:

exp_dependencies(dict)

static toml_dump(to_dump, fname)[source]

Dump toml to file.

Using tomlkit to preserve stucture

Parameters:

to_dump (dict) – Data to save
fname (str) – Filename

case_setup(config: ParsedConfig, output_file, mod_files: List[Path], case=None, host=None, expand_config=False)[source]

Do experiment setup.

Parameters:

config (.config_parser.ParsedConfig) – Parsed config file contents.
output_file (str) – Output config file.
mod_files (list) – Modifications. Defaults to None.
case (str, optional) – Case identifier. Defaults to None.
host (str, optional) – host name. Defaults to None.
expand_config (boolean, optional) – Flag for expanding macros in config

Returns:

Output config file.

Return type:

output_file (str)

get_git_info()[source]: Get git information.

tactus.formatters module

Define format validators used in the JSON schema validation.

duration_format_validator(duration: str) → bool[source]

Validate the format of a duration string against ISO8601.

Parameters:: duration – The duration string to validate.
Returns:: True if the duration string is valid, False otherwise.

duration_slice_format_validator(duration_slice: str) → bool[source]

Validate the format of a duration slice string against ISO8601.

Each part of the slice is validated using duration_format_validator.

Parameters:: duration_slice – The duration slice string to validate.
Returns:: True if the duration slice string is valid, False otherwise.

tactus.fullpos module

Fullpos namelist generation.

class Fullpos(domain, fpdir='.', fpfiles=None, fpdict=None, rules=None)[source]

Bases: object

Fullpos namelist generator based on (yaml) dicts.

check_non_instant_fields(selection, time_selection)[source]

Search for non instant fields.

Parameters:

selection (dict) – Dict with fullpos settings to be examined
time_selection (str) – Which selection time rules to check

Raises:

RuntimeError – Non instant fields found

construct()[source]

Construct the fullpos namelists.

Returns:: namfpc part selection (dict): xxtddddhhmm part
Return type:: namfpc_out (dict)
Raises:: InvalidSelectionCombination – Invalid combination in selection

expand(v, levtype, levels, domain, i)[source]

Expand fullpos namelists to levels and domains.

Parameters:

v (str) – parameter list
levtype (str) – type of vertical level in fullpos syntax
levels (list) – list of levels
domain (str) – domain name
i (int) – parameter conuter

Returns:

expaned names i (int): parameter counter

Return type:

d (dict)

load(fpdir, fpfiles)[source]

Load fullpos yaml file.

Parameters:

fpdir (str) – Path do fullpos settings
fpfiles (list) – List of yml files to read

Returns:

fullpos settings

Return type:

nldict (dict)

replace_rules(vin)[source]

Replace patterns from the rules dict.

Parameters:: vin (str|list) – Input values
Returns:: possibly replaced strings
Return type:: v (str|list)
Raises:: RuntimeError – Invalid type

update_selection(additions_list=None, additions_dict=None)[source]

Add choices to the selection section.

Parameters:

additions_list (list) – Additional selection to be read from files
additions_dict (dict) – Additional selection as a dictionary

exception InvalidSelectionCombinationError[source]

Bases: ValueError

Custom exception.

flatten_list(li)[source]: Recursively flatten a list of lists (of lists).

tactus.general_utils module

General utils for use throughout the package.

expand_dict_key_slice(dict_: Dict[int | str, Any], indices: List[int]) → Dict[int, Any][source]

Expand key slices of a Dict.

Handles slices in the form of “start:stop:step”, expands them to individual keys, and assigns the original value to all individual keys. Keys are converted to integers.

Any of the start, stop and step can be ommited. If start is ommited, it is set to the minimum value of indices. If stop is ommited, it is set to the maximum value of indices. If step is ommited, it is set to 1.

Parameters:

dict – The dict, which keys shall be expanded.
indices – The indices to respect when expanding, i.e. if expanded index is not in indices, it will not be added to the new dict.

Returns:

New dict with expanded keys.

Return type:

dict

expand_string_slice(string: int | str, indices: List[int]) → Generator[int | List[int], None, None][source]

Expand a string slice into a list of integers returned as generator.

Parameters:

string (int | str) – The string to expand
indices (List[int]) – Indices to respect, i.e. for max/min bounds

Yields:

Generator[int | List[int], None, None] –

The expanded string returned as: a generator.

Raises:

ValueError – If string, that is not a slice string, cannot be converted to int

get_empty_nested_defaultdict()[source]: Return an empty nested (recursive) defaultdict object.

merge_dicts(dict1: dict, dict2: dict, overwrite: bool = False, remove_none: bool = False) → dict[source]

Merge two dictionaries with values from dict2 taking precedence.

If values are lists, they are concatenated.

Parameters:

dict1 (dict) – Reference dict
dict2 (dict) – Update dict
overwrite (bool) – Whether to overwrite values in dict1 with values from dict2 if the keys are the same, but the types of the values are not lists or dicts.
remove_none (bool) – Whether to delete value in dict1 when the value in dict2 is None

Returns:

Merged dict

Return type:

(dict)

Raises:

RuntimeError – Invalid type

recursive_delete_keys(mapping: Dict[str, Any], keys_dict: Dict[str, bool])[source]

Recursively delete keys from a mapping based on a dictionary of keys to delete.

Keys are deleted in-place on the mapping object.

Parameters:

mapping – The mapping to delete keys from.
keys_dict – The dictionary of keys to delete.

recursive_dict_deviation(base_dict: dict, deviating_dict: dict) → dict[source]

Calculate the (recursive) difference between two dicts.

Parameters:

base_dict – The base dictionary to calculate deviation from.
deviating_dict – The dict to calculate the deviation of w.r.t to the base_dict

Returns:

The deviation as a dictionary.

recursive_freeze(obj)[source]

Transform recursively a dict into a immutable frozendict.

Parameters:: obj – the input dict
Returns:: The converted frozendict

recursive_substitute(value, platform, pos: List[str] | None = None)[source]

Recursively substitute variables in a nested dictionary.

Substitution is mainly done on value-level, but full configuration subtrees can be copied using the magic keys COPY (copies only all the values at the specified key) and COPYALL (copies all the values at the specified key, nestedly).

Parameters:

value – Value to substitute macros
platform – The platform used to substitute values
pos – current path-like position in the config tree (as used for COPYALL)

Returns:

subsituted value

recursive_unfreeze(obj, return_type=<class 'dict'>)[source]

Transform recursively a frozendict into a type that is mutable (e.g. dict).

Parameters:

obj – the input dict
return_type – the type to return (default = dict)

Returns:

The frozendict converted to return_type

value_from_any_generator(any_: Any | Sequence[Any] | Mapping[int, Any], indices: List[int], default_value: str | None = None) → Generator[str, None, None][source]

Yield values from any type.

Parameters:

any – The input object to yield values from.
indices – The indices to retrieve from the value.
default_value – The default value to use if an index is not found in Mapping.

Yields:

The value from the input object.

value_from_mapping_generator(mappable: Mapping[int, Any], keys: List[int], default_value: Any) → Generator[Any, None, None][source]

Yield values from a dictionary according to keys.

Parameters:

mappable – The mappable to yield values from
keys – The keys for which to retrieve corresponding values from the dictionary.
default_value – The default value to use if a key is not found.

Yields:

The value corresponding to the key.

value_from_sequence_generator(sequence: Sequence[Any]) → Generator[Any, None, None][source]

Yield alternately one of the values from a sequence of values.

The order of the yielded values is determined by the order of the sequence.

Parameters:: sequence – The sequence to yield values from.
Yields:: One of the values from sequence in alternate order.

tactus.geo_utils module

Utilities for simple geographic tasks.

class Projection(proj4str)[source]

Bases: object

Projection class.

check_key(key: str, config: dict) → bool[source]

Check if key is in config.

Parameters:

key (str) – Key to check
config (dict) – Configuration

Returns:

True if key is in config

Return type:

bool

Raises:

ValueError – If key is not in config

get_domain_properties(domain_spec: dict) → dict[source]

Get domain properties.

Parameters:: domain_spec (dict) – Domain specification
Returns:: Domain properties
Return type:: dict

class Projstring[source]

Bases: object

Proj4 string class.

get_projstring(lon0=0.0, lat0=-90.0) → str[source]

Get proj4 string.

Parameters:

lon0 (float, optional) – Central longitude. Defaults to 0.0.
lat0 (float, optional) – Central latitude. Defaults to -90.0.

Returns:

Proj4 string

Return type:

str

tactus.host_actions module

Handle host detection.

exception AmbigiousHostError[source]

Bases: ValueError

Custom exception.

exception HostNotFoundError[source]

Bases: ValueError

Custom exception.

class SelectHost[source]

Bases: object

Class for the host selection.

class TactusHost(known_hosts=None, known_hosts_file=None)[source]

Bases: object

TactusHost object.

detect_tactus_host(use_default=True)[source]

Detect tactus host by matching various properties.

First check self.tactus_host as set by os.getenv(“TACTUS_HOST”), second use the defined hosts in known_hosts.yml. If no matches are found return the first host defined in known_hosts.yml

Parameters:: use_default (boolean, optional) – Flag to return default host if host not found
Returns:: mapped hostname
Return type:: tactus_host (str)
Raises:: RuntimeError – Ambiguous matches

set_tactus_home(config, tactus_home=None)[source]

Set tactus_home in various ways.

Parameters:

config (.config_parser.ParsedConfig) – Parsed config file contents.
tactus_home (str) – Externally set tactus_home

Returns:

tactus_home

tactus.initial_conditions module

Initial_conditions.

class InitialConditions(config)[source]

Bases: object

FirstGuess.

check_if_found()[source]: Check if files are present.

find_initial_files()[source]: Find initial file.

success()[source]: Report of success.

tactus.logs module

Logging-related classes, functions and definitions.

class InterceptHandler(level=0)[source]

Bases: Handler

Add logging handler to augment python stdlib logging.

Logs which would otherwise go to stdlib logging are redirected through loguru.

emit(record: LogRecord) → None[source]

Emit a record.

Parameters:: record (builtin_logging.LogRecord) – Output record

class LogDefaults(*_args, **_kwargs)[source]

Bases: QuasiConstant

Defaults used for the logging system.

DIRECTORY = PosixPath('/home/runner/.logs/tactus')

LEVEL = 'INFO'

RETENTION_TIME = '1 week'

SINKS = {'console': <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>}

class LogFormatter(datetime: str = '<green>{time:YYYY-MM-DD HH:mm:ss}</green>', level: str = '<level>{level: <8}</level>', code_location: str = '<cyan>@{name}</cyan>:<cyan>{function}</cyan> <cyan><{file.path}</cyan>:<cyan>{line}>:</cyan>', message: str = '<level>{message}</level>')[source]

Bases: object

Helper class to setup logging without poluting the module’s main scope.

code_location: str = '<cyan>@{name}</cyan>:<cyan>{function}</cyan> <cyan><{file.path}</cyan>:<cyan>{line}>:</cyan>'

datetime: str = '<green>{time:YYYY-MM-DD HH:mm:ss}</green>'

format_string(loglevel: str)[source]: Return the appropriate fmt string according to log level and fmt opts.

level: str = '<level>{level: <8}</level>'

message: str = '<level>{message}</level>'

class LoggerHandlers(default_level: str = 'INFO', **sinks)[source]

Bases: Sequence

Helper class to configure logger handlers when using loguru.logger.configure.

add(name, sink, **configs)[source]: Add handler to instance.

log_elapsed_time(package_name='tactus', version='1.2.0', **kwargs)[source]: Return a decorator that logs beginning, exit and elapsed time of function.

tactus.mars_utils module

Utility for marsprep.

class BaseRequest(*, class_: str, data_type: str, expver: str, levtype: str, date: str, time: str, steps: str, param: str, target: str)[source]

Bases: object

Represents a structured data request with configurable parameters.

This class facilitates the creation of data requests by encapsulating request attributes and providing methods for modifying and enhancing the request structure. The request data is stored in a dictionary accessible via the request attribute.

Parameters:

class (str) – The class of the request, e.g., “D1”.
data_type (str) – The type of data requested (e.g., analysis, forecast).
expver (str) – Experiment version identifier.
levtype (str) – Level type (e.g., surface, pressure levels).
date (str) – The date range for the request in YYYYMMDD format.
time (str) – The time range for the request in HHMM format.
steps (str) – Forecast steps or time intervals.
param (str) – Parameter codes for the requested data.
target (str) – Target file or identifier for the output.
request (dict) – A dictionary representation of the request, initialized automatically upon object creation.

add_database_options()[source]: Add database options.

add_levelist(levelist: str)[source]

Set multilevel if needed.

Parameters:: levelist (string) – list of levels
Returns:: None

class_: str

data_type: str

date: str

expver: str

levtype: str

param: str

replace(**kwargs)[source]: Return new instance with updated values.

request: dict

steps: str

target: str

time: str

update_request(upd: dict)[source]

Add or replace keys in a mars request.

Parameters:: upd (dict) – a dict with keys and values

add_additional_data_to_all(tag: str, steps: List[str], members_dict: Dict[str, List[int]], additional_data: Dict[str, bytes])[source]

Add additional common data to all files defined by tag, step and member.

The additional_data dict is expected to have the following structure:

{
    "data_key1": b"Data to add to all files",
    "data_key2": b"Data to add to all files",
    ...
}

Parameters:

tag (str) – Name of tag
steps (List[int]) – Steps used to construct filename.
members_dict (Dict[str, List[int]]) – Dictionary with members used to construct filename.
additional_data (Dict[str, bytes]) – Dictionary containing the data to add.

Raises:

FileNotFoundError – If a file does not exist when trying to append data to it.

add_additional_file_specific_data(additional_data: Dict[str, bytes])[source]

Add additional file specific data.

The additional_data dict is expected to have the following structure, where the “common_data” key is optional:

{
    "common_data": b"Common data to add to all files",
    "file1": b"Data to add to file1",
    "file2": b"Data to add to file2",
    ...
}

If the “common_data” key is present, the data will be added to a file before the file-specific data is added.

Parameters:: additional_data (Dict[str, bytes]) – Dictionary containing the data to add (value) to a given file (key).
Raises:: FileNotFoundError – If a file does not exist when trying to append data to it.

check_data_available(basetime, mars)[source]

Check if there is data for basetime for choosen expver.

Parameters:

basetime (str) – basetime
mars (dict) – mars config section

Raises:

ValueError – if the basetime falls outside the available range

compile_target(tag: str, member_type: str, members: int | List[int], step: str | int = '[STEP]') → str[source]

Compiles a target name given member_type, members and (optinally) step.

Parameters:

tag (str) – the tag used for this target
member_type (str) – member_type for member, control_member or perturbed_members
members (int or List[int]) – the id of the member(s)
step (str or int) – the id of the step(s)

Returns:

Compiled target name

fix_snow_layer(tag: str, steps: List[int], members_dict: Dict[str, List[int]])[source]: Fix snow layer for snow albedo.

get_and_remove_data(file_name: str) → bytes[source]

Read in and subsequently remove binary data.

Parameters:: file_name (str) – The name of the file containing the data to be read.
Returns:: The binary data read from the file.
Return type:: bytes

get_domain_data(config)[source]

Read and return domain data.

Parameters:: config (tactus.ParsedConfig) – Configuration from which we get the domain data
Returns:: String containing the domain info for MARS

get_mars_keys(source, key_filter='-w shortName:s=z')[source]: Get the basic MARS settings from a static file, to correct a request.

get_steplist(bd_offset, fc_range, bdint)[source]

Get the list of steps for Mars request.

Parameters:

bd_offset (timedelta) – first boundary time
fc_range (timedelta) – forecast range
bdint (timedelta) – frequency of boundary files

Returns:

list of steps

Return type:

steps (List[int])

get_steps_and_members_to_retrieve(steps: List[int], path: Path, tag: str, members: List[int], platform: Platform = None, basetime=None, validtime=None, use_lockfile=False) → Tuple[List[int], List[int], Dict[str, List[int | None]], Dict[int, List[int]], Dict[int, List[int]]][source]

Check which mars file already exist and returns steps and members which missing.

Parameters:

steps (List[int]) – list of steps
path (pathlib.Path) – path to global files
tag (string) – name of tag to check
members (List[int]) – members to retrieve
platform (Platform, optional) – Platform to process macro’s in tag
basetime (optional) – Base time used in platform.substitute. Defaults to None.
validtime (optional) – Valid time used in platform.substitute. Defaults to None.
use_lockfile (optional) – use a lockfile for each file fetched? Defaults to False.

Returns:

steps to retrieve waitfor_steps (List[int]): steps processed elsewhere members_dict (Dict[str, List[int | None]]): members to retrieve missing_member_steps (Dict[int, List[int]]): dict with missing steps per member waitfor_member_steps (Dict[int, List[int]]): dict with steps to wait for

Return type:

steps (List[int])

get_value_from_dict(dict_, key_orig)[source]

Check value according to key.

If a string returns the value itself
If key is a date search for the most suitable match in value
Else return the value matching the key.

Parameters:

dict (str, BaseConfig object) – Values to select
key_orig (str) – key for value checking

Returns:

Found value

Return type:

value (str)

Raises:

ValueError – Exception

mars_selection(selection: str, config: ParsedConfig) → dict[source]

Copy default settings if requested.

Parameters:

selection (str) – The selection to use.
config (tactus.ParsedConfig) – Configuration object

Returns:

mars config section

Return type:

mars (dict)

mars_write_method(mars_version: int) → str[source]

Gets the mars write_method depending on mars version.

Parameters:: mars_version (int) – version of mars client
Returns:: retrieve or read
Return type:: write_method

move_files(tag: str, steps: List[int], members_dict: Dict[str, List[int]], target_dir: Path)[source]

Move files to the final location.

Parameters:

tag (str) – Name of tag
steps (list) – steps to process the files
members_dict (Dict[str, List[int]]) – dict with members to procces the files
target_dir (Path) – target directory to move the files to

wait4_file(file: str)[source]

Wait for a single file to appear in final location.

Parameters:: file (str) – Path to file
Raises:: SystemExit – if lockfile disappears before file is created

waitfor_files(tag: str, steps: List[int], members_dict: Dict[str, List[int]], target_dir: Path)[source]

Wait for files to appear in final location.

Parameters:

tag (str) – Name of tag
steps (list) – steps to wait for (processed elsewhere)
members_dict (Dict[str, List[int]]) – dict with members
target_dir (Path) – target directory where files should appear

write_compute_mars_req(name: str, formula: str, fieldset: str = '', target: str = '', omode: str = 'w')[source]

Write a COMPUTE request for MARS.

Parameters:

name (string) – name of request
formula (string) – formula for computation.
fieldset (string) – fieldset for computation.
target (string) – target file
omode (string) – file open mode (w: create, a: append), default=”a”

write_retrieve_mars_req(req, name: str, method: str, omode: str = 'w')[source]

Write a RETRIEVE/READ request for MARS.

Parameters:

req (BaseRequest) – namelist object to write
name (string) – request file name
method (string) – selected method, retrieve or read
omode (string) – file open mode (w: create, a: append), default=”w”

write_write_mars_req(name: str, fieldset: str, target: str, omode: str = 'w')[source]

Write a WRITE request for MARS.

Parameters:

name (string) – name of request
fieldset (string) – fieldset for computation.
target (string) – target file
omode (string) – file open mode (w: create, a: append), default=”a”

tactus.namelist module

Namelist handling for MASTERODB w/SURFEX.

exception InvalidNamelistKindError[source]

Bases: ValueError

Custom exception.

exception InvalidNamelistTargetError[source]

Bases: ValueError

Custom exception.

class NamelistComparator(config)[source]

Bases: object

Helper class for namelist generation and integration.

compare_dicts(dbase, dcomp, action)[source]

Compare two dictionaries, recursively if needed.

The dict must only consist of dict, list, str, float, int, bool

Parameters:

dbase – base dict
dcomp – comparison dict
action (str) – one of ‘intersection’, ‘diff’ or ‘union’

Returns:

result dict from the specified comparison action

Return type:

dout

Raises:

SystemExit # noqa – DAR401

compare_lists(libase, licomp, sib, sic, sio, action)[source]

Compare two lists, recursively if needed.

The list must only consist of dict, list, str, float, int, bool

Parameters:

libase – base list
licomp – comparison list
sib – list of start indices for libase
sic – list of start indices for licomp
sio – list of start indices for liout (modified)
action (str) – one of ‘intersection’, ‘diff’ or ‘union’

Returns:

result list from the specified comparison action

Return type:

liout

Raises:

SystemExit # noqa – DAR401

class NamelistConverter[source]

Bases: object

Helper class to convert namelists between cycles, based on thenamelisttool.

static apply_tnt_directives_to_ftn_namelist(tnt_directive_filename, input_ftn)[source]

Apply the tnt directives to a fotran namelist using tnt.

Parameters:

tnt_directive_filename – the tnt directive filename
input_ftn – the namelist fortran file

Raises:

SystemExit – when conversion failed

static apply_tnt_directives_to_namelist_dict(tnt_directive_filename, namelist_dict)[source]

Apply the tnt directives to a namelist as dictionary.

Parameters:

tnt_directive_filename – the tnt directive filename
namelist_dict – the namelist dictionary

Returns:

the converted namelist dictionary

Return type:

new_namelist

Raises:

SystemExit – when conversion failed

static convert_ftn(input_ftn, output_ftn, from_cycle, to_cycle)[source]

Convert a namelist in fortran file between two cycles.

Parameters:

input_ftn – the input fortran filename
output_ftn – the output fortran filename
from_cycle – the input cycle
to_cycle – the target cycle

static convert_yml(input_yml, output_yml, from_cycle, to_cycle)[source]

Convert a namelist in yml file between two cycles.

Parameters:

input_yml – the input yaml filename
output_yml – the output yaml filename
from_cycle – the input cycle
to_cycle – the target cycle

Raises:

SystemExit – when conversion failed

static get_known_cycles()[source]: Return the cycles handled by the converter.

static get_tnt_files_list(from_cycle, to_cycle)[source]: Return the list of tnt directive files required for the conversion.

static get_to_next_version_tnt_filenames()[source]: Return the tnt file names between get_known_cycles().

class NamelistGenerator(config, kind, substitute=True)[source]

Bases: object

Fortran namelist generator based on hierarchical merging of (yaml) dicts.

assemble_namelist(target)[source]

Generate the namelists for ‘target’.

Parameters:: target (str) – task to generate namelists for
Returns:: Assembled namelist
Return type:: nlres (f90nml.Namelist)

fn_config(arg, default=None)[source]: Resolve namelist function cfg.

fn_stepfreq(arg)[source]: Resolve namelist function stepfreq.

fn_steplist(time_intervals)[source]: Resolve namelist function steplist.

fn_tstep(arg)[source]: Resolve namelist function tstep.

fn_vsigqsat_by_gridsize(arg1, arg2)[source]

Resolve namelist function vsigqsat_by_gridsize.

Parameters:

arg1 – VSIGQSAT reference value
arg2 – Reference gridsize

Returns:

Scaled VSIGQSAT

Return type:

result

generate_namelist(target, output_file)[source]

Generate the namelists for ‘target’.

Parameters:

target (str) – task to generate namelists for
output_file – where to write the result (fort.4 or EXSEG1.nam typically)

load(target)[source]

Generate the namelists for ‘target’.

Parameters:: target (str) – task to generate namelists for
Returns:: Assembled namelist
Return type:: nlres (dict)
Raises:: InvalidNamelistTargetError # noqa – DAR401

load_user_namelist()[source]

Read user provided namelist.

Returns:: Logics if namelist is found or not nldict : Namelist as dictionary cndict : Rules for dictionary
Return type:: found (boolean)

resolve_macros(cn)[source]: Resolve all macros in a nested dict (both keys and values!).

update(nldict, cndict_tag)[source]

Update with additional namelist dict.

Parameters:

nldict (dict) – additional namelist dict
cndict_tag – name to be used for recognition

update_from_config(target)[source]

Update with additional namelist dict from config.

Parameters:: target (str) – task to generate namelists for

uppercase_keys(d)[source]

Convert the keys in a dict to uppercase.

Parameters:: d (dict) – Dict to be parsed
Returns:: The parsed dict
Return type:: d (dict)

write_namelist(nml, output_file)[source]

Write namelist using f90nml.

Parameters:

nml (f90nml.Namelist) – namelist to write
output_file (str) – namelist file name

class NamelistIntegrator(config)[source]

Bases: object

Helper class to read fortran namelists and store as yaml, in a more compact form.

Reduces duplication if several namelists have similar settings.

static dict2yml(nmldict, ymlfile, ordered_sections=None)[source]: Write dict as yaml file.

ftn2dict(ftnfile)[source]: Read fortran namelist file with f90nml and return as plain dict.

static yml2dict(ymlfile)[source]: Read yaml namelist file and return as dict.

find_value(s)[source]: Purpose: un-quote (list of) numbers and booleans.

flatten_cn(li)[source]: Recursively flatten a list of lists (of lists).

list_set_at_index(li, ix, val)[source]: Handle ‘sparse’ lists (that may contain null values).

represent_ordereddict(dumper, data)[source]: YAML representer, simplifies working with namelists read by f90nml.

to_dict(x)[source]: Recursively modify OrderedDict etc. to normal dict (required for OmegaConf).

write_namelist(nml, output_file)[source]

Write namelist using f90nml.

Parameters:

nml (f90nml.Namelist) – namelist to write
output_file (str) – namelist file name

tactus.os_utils module

Utilities for simple tasks on OS level.

class FileLock(filepath: str, timeout: int = 600, check_interval: int = 10, delete_existing: bool = False)[source]

Bases: object

Context manager for file locking using lockfiles.

class Search[source]

Bases: object

Search class.

static find_files(directory, prefix='', postfix='', pattern='', recursive=True, onlyfiles=True, fullpath=False, olderthan=None, inorder=False) → list[source]

Find files in a directory.

Parameters:

directory (str) – Directory to search in.
prefix (str, optional) – Only find files with this prefix. Defaults to “”.
postfix (str, optional) – Only find files with the postfix. Defaults to “”.
pattern (str, optional) – Only find files with matching pattern. Defaults to “”.
recursive (bool, optional) – Go into directories recursively. Defaults to True.
onlyfiles (bool, optional) – Show only files. Defaults to True.
fullpath (bool, optional) – Give full path. Defaults to False. If recursive=True, fullpath is given automatically.
olderthan (int, optional) – Match only files older than X seconds from now. Defaults to None.
inorder (bool, optional) – Return sorted list of filenames. Defaults to False.

Returns:

List containing file names that matches criterias

Return type:

list

Examples

>>> files = find_files(
                '/foo/', prefix="", postfix="", recursive=False,
                onlyfiles=True, fullpath=True, olderthan=86400*100
            )

filepath_iterator(paths, filename_pattern='*')[source]

Return iterator of paths to files given a list of file or directory paths.

Given a path or list of paths, yield Path objects corresponding to them. If a path points to a directory, then the directory is searched recursively and the paths to the files found in this process will be yielded.

Parameters:

paths (Union[pathlib.Path, List[pathlib.Path], str, List[str]]) – A single path or a collection of paths.
filename_pattern (str, optional) – Pattern used in the recursive glob in order to select the names of the files to be yielded. Defaults to “*”.

Yields:

pathlib.Path – Path to located files.

join_files(input_files: List[str], output_filepath: str)[source]

Joins multiple files into a single file.

Parameters:

input_files (List[str]) – List of files to be joined/concatenated
output_filepath (str) – Output file

list_files_join(folder, f_pattern)[source]

Read and return file names based on given pattern.

Parameters:

folder – path with file location
f_pattern – glob pattern

Returns:

list of files that should be joined

ping(host)[source]

Ping host.

Parameters:: host (str) – Host to ping
Returns:: True if host responded
Return type:: (boolean)

remove_empty_dirs(src, dry_run=False)[source]

Remove directories.

Recursively and permanently removes the specified directory, and all of its empty subdirectories.

Parameters:

src (str or Path) – Top search directory
dry_run (boolean) – Flag for execution of cleaning or not

Returns:

True if any files found

Return type:

found_files (boolean)

remove_ifexists(file, etime=1.7976931348623157e+308)[source]: Utility function to be used for lockfiles.

resolve_path_relative_to_package(path: Path, ignore_errors: bool = False) → Path[source]

Resolve path relative to any sys.path entry.

If the path exists as is, return it. If not, derive a relative path by stripping known sys.path prefixes, then search every sys.path entry for that relative path. Raises an error if more than one candidate is found to avoid silent ambiguity.

Parameters:

path (Path) – Path to resolve.
ignore_errors (bool, optional) – Option to ignore errors. Defaults to False.

Returns:

Original path (if exists locally), or resolved path relative to: a sys.path entry.

Return type:

Path

Raises:

FileNotFoundError – If it was impossible to determine path relative to package.
FileNotFoundError – If file does not exist locally or in the package directory.

strip_off_mount_path(path: str | Path) → Path[source]

Strip off the mount path from a given path.

Assumptions:

the path contains the user name as a directory.
the parent of the user directory is of the format “<new-dir-name>” or “%_<new-dir-name>_*”, where “*” contains no underscore(s), “%” might contain underscore(s) and where the <new-dir-name> will be used as the new parent directory name relative to the user directory.

Parameters:: path (Union[str, Path]) – Path to strip off the mount path from.
Returns:: Path with the mount path stripped off.
Return type:: path
Raises:: ValueError – If the parent of the user directory only contains 1 underscore.

Example

>>> strip_off_mount_path("/etc/ecmwf/nfs/dh1_home_b/$USER/tactus/tactus")
Path("/home/$USER/tactus/tactus")
>>> strip_off_mount_path("/etc/ecmwf/nfs/dh1_10_perm_b/$USER/tactus")
Path("/perm/$USER/tactus")

tactusmakedirs(path: str | Path, unixgroup='', exist_ok=True, def_dir_mode=493)[source]

Create directories and change unix group as required.

For a given path the top directory that does not yet exist is searched for, created and unix group is set, if required. Permissions are set such that all subdirectories and new files inherit the unix group.

Parameters:

path (str | Path) – directory path that should be created if it doesn’t already exist.
unixgroup (str, optional) – unix group the newly created dirs should belong to.
exist_ok (boolean, optional) – Define whether directories may already exist or whether an error should be raised.
def_dir_mode (int, optional) – Default directory persmission mode. Defaults to 0o755

Raises:

OSError – If cannot create the directory.

tactus.plugin module

Plug-in functionality.

class TactusPlugin(name: str, path: Path)[source]

Bases: object

Tactus plugin.

class TactusPluginFromConfigFile(config_file)[source]

Bases: TactusPlugin

Tactus plugin.

static get_plugin_config(config_file)[source]: Get the registry config.

class TactusPluginRegistry(config=None)[source]

Bases: object

Registry of plugins.

get_registry_config()[source]: Get the registry config.

load_plugin(plugin: TactusPlugin)[source]

Load plugin.

Parameters:: plugin (TactusPlugin) – tactus plugin

load_plugins()[source]: Load all registered plugins.

plugin_exists(plugin: TactusPlugin)[source]

Check if plugin exists.

Parameters:: plugin (TactusPlugin) – tactus plugin
Returns:: True if already exists in registry.
Return type:: bool

register_plugin(plugin: TactusPlugin)[source]

Register plugin.

Parameters:: plugin (TactusPlugin) – tactus plugin

save_registry(config_file)[source]

Save registry config.

Parameters:: config_file (str) – Filename

tactus_plugin()[source]: Base tactus plugin.

class TactusPluginRegistryFromConfig(config)[source]

Bases: TactusPluginRegistry

Create a registry from a tactus config file.

class TactusPluginRegistryFromFile(config_file)[source]

Bases: TactusPluginRegistry

Registry file of plugins.

tactus.reference_checker module

Compare results agains references.

class CheckDefinition(taskname, rulename, label_suffix, method, inpath_pattern, files_pattern, references_pattern, results_dir_pattern, generate_dir_pattern, tool='')[source]

Bases: object

A definition containing the required information to perform a check.

create_items(platform: Platform)[source]

Transform the patterns in the definition into a list of items with filepath.

Parameters:: platform – platform class providing the substitution capabilities.

static create_list_of_check_definitions(config, taskname, label_suffix, rules_active, check, generate) → list[CheckDefinition][source]

Create the list of items to be checked.

Parameters:

config (tactus.ParsedConfig) – Configuration
taskname – the name of the task
label_suffix – the suffix for the label
rules_active – list of rules that are active
check – boolean indicating if the check should be performed
generate – boolean indicating if the reference generation should be performed

Returns:

list of CheckDefinition

class CheckItem(test_file, reference_file, generate_file, result_file, result='N/A')[source]

Bases: object

An item to be checked, describing the test performed and the result.

class CheckSummary(fileformat, filename)[source]

Bases: object

Generic class for summary files generation containing comparisons results.

static create_summary_list(config) → list[CheckSummary][source]

Create the list of summary_list from the configuration.

Parameters:: config (tactus.ParsedConfig) – Configuration
Returns:: list of CheckSummary

delete()[source]: Delete file if exists with a logger information message.

init_full_path(platform)[source]

Init self.full path using platform substitute.

Parameters:: platform – Platform

class CheckSummaryAnalysis(check)[source]

Bases: object

Manage the final section of a summary, containing global tests results.

increment_error_count()[source]: Increment error_count.

increment_generated_count()[source]: Increment generated_count.

increment_missing_count()[source]: Increment missing_count.

increment_success_count()[source]: Increment success_count.

message()[source]: Retern a summary message.

success()[source]: Return true iff all the tests are successful.

total_count()[source]: Retern total number of files tested and missing.

class CheckSummaryJson(fileformat, filename)[source]

Bases: CheckSummary

Class to generate summary File in txt format.

append(check_definitions: list[CheckDefinition])[source]

Write the summary file to disk in the json format.

Parameters:: check_definitions – the input list of check definitions
Raises:: FileNotFoundError – First call CheckSummaryJson.create to generate

compute_and_append_analysis(check)[source]: Perform an analysis of the json summary and append it at the end.

create(platform)[source]

Create a summary file with header.

Parameters:: platform – the platform

class CheckSummaryTxt(fileformat, filename)[source]

Bases: CheckSummary

Class to generate summary File in txt format.

append(check_definitions: list[CheckDefinition])[source]

Write check_definitions to summary file to disk in the txt format.

Parameters:: check_definitions – the input list of check definitions
Raises:: FileNotFoundError – First call CheckSummaryTxt.create to generate

compute_and_append_analysis(check)[source]: Perform an analysis of the txt summary and append it at the end.

create(platform)[source]

Create a summary file with header.

Parameters:: platform – the platform

class NormsChecker(which, tolerance, mode)[source]

Bases: ReferenceChecker

Compare the norms in Node files against a reference.

compare(test_log, reference_log, out_file) → str[source]

Compare the norms against a reference.

Parameters:

test_log – name of the lof to compare
reference_log – name of the reference log file
out_file – name of the file produced by the comparion

Returns:

str giving the result of the comparison

class ReferenceCheckManager(config, taskname, label_suffix, rules_active, check, generate, create_summary, analyze_summary, suppress_exception)[source]

Bases: object

A class managing the checks against reference.

analyze_summaries()[source]: Analyze the summaries.

append_results_to_summaries()[source]: Write the summary_list to disk in the correct output format.

check_references()[source]: Check the result files against the reference files.

static create_reference_check_manager(config, taskname) → ReferenceCheckManager[source]

Static method to create a ReferenceCheckManager.

Parameters:

config – configuration dictionary
taskname – the name of the task

Returns:

A ReferenceCheckManager object or None

create_summaries_with_header(platform)[source]: Create summary_list file on disk with the correct output format.

execute(fmanager: FileManager)[source]

Do reference against reference checking.

Parameters:: fmanager – a file manager

generate_references(fmanager: FileManager)[source]

Generate the reference files into the reference folder.

Already existing files will be overwitten :param fmanager: a file manager

prepare(platform: Platform)[source]

Prepare for comparing against references.

Parameters:: platform – platform class providing the substitution capabilities.

class ReferenceChecker(tool)[source]

Bases: object

Base class for comparison against a reference.

compare(test_file, reference_file) → str[source]

Compare a file to a reference file.

Parameters:

test_file – name of the file to compare
reference_file – name of the reference file

Returns:

str giving the result of the comparison

static create_reference_checker(method, config) → ReferenceChecker[source]

Instanciate a ReferenceChecker following the method and the configuration.

Parameters:

method – str defining the method
config (tactus.ParsedConfig) – Configuration

Returns:

A ReferenceChecker

prepare(platform: Platform)[source]

Prepare the binary path by substituting the pattern.

Parameters:: platform – platform class providing the substitution capabilities.

class XToolChecker(binary_pattern, file_format, tolerance, args_template=None)[source]

Bases: ReferenceChecker

Compare fields against reference using xtool.

compare(test_file, reference_file, out_file) → str[source]

Compare the fields against a reference using xtool.

Parameters:

test_file – name of the lof to compare
reference_file – name of the reference log file
out_file – name of the file produced by the comparison

Returns:

str giving the result of the comparison

Raises:

Exception – Any exception occuring during xtool that is no CalledProcessError

prepare(platform: Platform)[source]

Prepare the binary path by substituting the pattern.

Parameters:: platform – platform class providing the substitution capabilities.

tactus.scheduler module

Scheduler module.

class EcflowClient(server, task)[source]

Bases: object

An ecflow client.

Encapsulate communication with the ecflow server. This will automatically call the child command init()/complete(), for job start/finish. It will also handle exceptions and signals, by calling the abort child command. ONLY one instance of this class, should be used. Otherwise zombies will be created.

static at_time()[source]

Generate time stamp.

Returns:: Time stamp.
Return type:: str

signal_handler(signum, extra=None)[source]

Signal handler.

Parameters:

signum (_type_) – _description_
extra (_type_, optional) – _description_. Defaults to None.

class EcflowLogServer(config)[source]

Bases: object

Ecflow log server.

class EcflowServer(config, start_command=None)[source]

Bases: Server

Ecflow server.

begin_suite(suite_name)[source]

Begin the suite.

Parameters:: suite_name (str) – Nam eof the suite.

force_aborted(task)[source]

Force the task aborted.

Parameters:: task (scheduler.EcflowTask) – Task to force aborted.

force_complete(task)[source]

Force the task complete.

Parameters:: task (scheduler.EcflowTask) – Task to force complete.

get_all_tasks(node)[source]: Recursively yield all Task nodes under a Suite or Family node.

get_config_of_suite(suite)[source]: Get cinfig file of the suite.

get_ecf_vars(suite)[source]

Get some ecf_vars from a ecflow suite.

Parameters:: suite (Ecflow suite) – suite object.
Returns:: List of Paths
Return type:: ecf_vars (list)

get_suites_from_server(ignore, complete=False)[source]

Get all suites from ecflow server.

Parameters:

ignore (list) – List of suites which should be ignore.
complete (boolean) – True if suite should be complete. Defaults: Fasle.

Returns:

List of ecflow Suite objects on server.

Return type:

list

remove_suites(suite_list, check_if_complete=True)[source]

Remove suites selected from a list.

Parameters:

suite_list (list) – Suite names.
check_if_complete (boolean) – True if suite has to be complete.

replace(suite_name, def_file)[source]

Replace the suite name from def_file.

Parameters:

suite_name (str) – Suite name.
def_file (str) – Definition file.

Raises:

RuntimeError – If suite cannot be replaced.

start_server()[source]

Start the server.

Raises:: RuntimeError – Server is not running or Could not restart server.

suite_finish_time(suite, force_delete_time, last_task_name=None)[source]

Get time when suite finished.

Parameters:

suite (ecflow.Suite) – suite object
force_delete_time (datetime) – return this time, if files don’t exists.
last_task_name (str) – name of the last task in the suite

Returns:

timestamp when suite finished

Return type:

float

suite_is_complete(suite)[source]

Returns the true if a suite is complete.

Parameters:: suite (suite object) – Suite
Returns:: Suite has complete status
Return type:: suite_is_complte (boolean)

suspend(task_or_path)[source]

Suspend a task, family, or suite.

Parameters:: task_or_path (EcflowTask | str) – Task object or full ecFlow path

class EcflowTask(ecf_name, ecf_tryno, ecf_pass, ecf_rid, ecf_timeout=20)[source]

Bases: object

Ecflow scheduler task.

class Server(config)[source]

Bases: ABC

Base server/scheduler class.

abstractmethod begin_suite(suite_name)[source]

Begin the suite in a server specific way.

Parameters:: suite_name (str) – Name of the suite
Raises:: NotImplementedError – Must be implemented by the child server object.

abstractmethod replace(suite_name, def_file)[source]

Create or change the suite definition.

Parameters:

suite_name (str) – Name of the suite.
def_file (str) – Name of the definition file.

Raises:

NotImplementedError – Must be implemented by the child server object.

replace_node(node_path, def_file)[source]

Start the suite.

Replace a node.

Parameters:

node_path (str) – Path of the node
def_file (str) – Name of the definition file.

abstractmethod start_server()[source]

Start the server.

Raises:: NotImplementedError – Must be implemented by the child server object.

start_suite(suite_name, def_file, begin=True)[source]

Start the suite.

All the servers have these methods implemented and can start the server in a server specific way.

Parameters:

suite_name (str) – Name of the suite
def_file (str) – Name of the definition file.
begin (bool, optional) – If the suite should begin. Defaults to True.

tactus.submission module

Module to handle submissions.

class NoSchedulerSubmission(task_settings)[source]

Bases: object

Create and submit job without a scheduler.

submit(task: str, config: ParsedConfig, template_job: str, task_job: Path, output: Path, member: int | None = None, troika: str | None = 'troika', create_only: bool | None = False)[source]

Submit task.

Parameters:

task (str) – Task name
config (tactus.ParsedConfig) – Config
template_job (str) – Task template job file
task_job (Path) – Task job file
output (Path) – Output file
member (int, optional) – Member number for which to submit job. Defaults to None.
troika (str, optional) – troika binary. Defaults to “troika”.
create_only – (bool, optional): Only create the job, do not submit it.

Raises:

RuntimeError – Submission failure.

class ProcessorLayout(kwargs: dict)[source]

Bases: object

Set processor information.

get_proc_dict()[source]: Generate a processor dict.

get_wrapper()[source]: Get and potentially parse the wrapper.

class TaskSettings(config)[source]

Bases: object

Set the task specific setttings.

get_settings(task)[source]

Get the settings.

Parameters:: task (_type_) – _description_
Returns:: _description_
Return type:: _type_

get_task_settings(task, key=None, variables=None, ecf_micro='%')[source]

Get task settings.

Parameters:

task (_type_) – _description_
key (_type_, optional) – _description_. Defaults to None.
variables (_type_, optional) – _description_. Defaults to None.
ecf_micro (str, optional) – _description_. Defaults to “%”.

Returns:

_description_

Return type:

_type_

parse_job(task: str, config: ParsedConfig, input_template_job: str, task_job: Path, member: int | None = None, variables=None, ecf_micro: str = '%', scheduler: str = 'ecflow')[source]

Read default job and change interpretor.

Parameters:

task (str) – Task name
config (tactus.config) – The configuration
input_template_job (str) – Input container template.
task_job (Path) – Task container
member (int, optional) – Member number for which to parse job. Defaults to None.
variables (_type_, optional) – _description_. Defaults to None.
ecf_micro (str, optional) – _description_.
scheduler (str, optional) – Scheduler. Defaults to ecflow.

Raises:

RuntimeError – In case of missing module env file

parse_submission_defs(task)[source]

Parse the submssion definitions.

Parameters:: task (str) – The name of the task
Returns:: Parsed settings
Return type:: dict
Raises:: RuntimeError – Undefined submit type

recursive_items(dictionary)[source]

Recursive loop of dict.

Parameters:: dictionary (_type_) – _description_
Yields:: _type_ – _description_

update_task_setting(dic, upd)[source]

Update task settings dictionary.

Parameters:

dic (dict) – Dictionary to update
upd (dict) – Values to update

Returns:

updated dictionary

Return type:

dict

tactus.test_runner module

Test runner functionality for running integration test cases.

class TestCases(args)[source]

Bases: object

Class to orchestrate the tests.

create(cases=None)[source]

Create the modif files and populate self.cmds for the given cases.

Parameters:: cases (list, optional) – Cases to process; defaults to self.selection

execute(args)[source]

Execute test cases.

Parameters:: args – Command line arguments

get_binaries()[source]: Get the correct binaries.

get_tactus_version()[source]: Get tactus version info.

get_tag(definitions)[source]

Get and validate tag.

Parameters:: definitions (dict) – Configuration
Raises:: ValueError – If tag has leading digits

list()[source]: List configurations.

resolve_selection(definitions)[source]

Resolve the selections.

Parameters:: definitions (dict) – Configuration
Returns:: List of selected configurations
Return type:: selection (list)

start()[source]: Start the run.

update_binary_paths()[source]: Update the correct binaries in the internal config object.

update_hostnames(hostnames)[source]

Update host and domain name.

Parameters:: hostnames (dict) – Dict of host cases with properties

run_test(args, config=None)[source]

Entry point for the tactus test command.

Parameters:

args – Parsed command line arguments
config – Unused; the test command manages its own config loading

tactus.toolbox module

Toolbox handling e.g. input/output.

exception ArchiveError[source]

Bases: Exception

Error raised when there are problems archiving data.

class ArchiveProvider(config, pattern, fetch=True)[source]

Bases: Provider

Data from ECFS.

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool

class ECFS(config, pattern, fetch=True)[source]

Bases: ArchiveProvider

Data from ECFS.

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool

class FDB(config, pattern, fetch=True)[source]

Bases: ArchiveProvider

Dummy FDB class.

check_expver_restrictions(expver)[source]

Check if user is allowed to archive this expver.

Parameters:: expver (str) – FDB experiment identifier
Raises:: RuntimeError – If user is not allowed to archive for this expver

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Raises:: RuntimeError – If expver not set
Returns:: True if success
Return type:: bool

class FileManager(config)[source]

Bases: object

FileManager class.

Default tactus provider.

Platform specific.

create_list(basetime, forecast_range, input_template, output_settings)[source]

Create list of files to process.

Parameters:

basetime (datetime.datetime) – Base time,
forecast_range (datetime.datetime) – forecast range,
input_template (str) – Input template,
output_settings (str) – Output settings

Returns:

dict of validates and grib fiels

Return type:

dict

get_input(target, destination, basetime=None, validtime=None, check_archive=False, provider_id='symlink')[source]

Set input data to tactus.

Parameters:

target (str) – Input file pattern
destination (str) – Destination file pattern
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.
check_archive (bool, optional) – Also check archive. Defaults to False.
provider_id (str, optional) – Provider ID. Defaults to “symlink”.

Returns:

provider, resource

Return type:

tuple

Raises:

ProviderError – “No provider found for {target}”
NotImplementedError – “Checking archive not implemented yet”

get_output(target, destination, basetime=None, validtime=None, archive=False, provider_id='move')[source]

Set output data from tactus.

Parameters:

target (str) – Input file pattern
destination (str) – Destination file pattern
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.
archive (bool, optional) – Also archive data. Defaults to False.
provider_id (str, optional) – Provider ID. Defaults to “move”.

Returns:

provider, aprovider, resource

Return type:

tuple

Raises:

ArchiveError – Could not archive data
NotImplementedError – Archive = True is not implemented

input(target, destination, basetime=None, validtime=None, check_archive=False, provider_id='symlink')[source]

Set input data to tactus.

Parameters:

target (str) – Input file pattern
destination (str) – Destination file pattern
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.
check_archive (bool, optional) – Also check archive. Defaults to False.
provider_id (str, optional) – Provider ID. Defaults to “symlink”.

input_data_iterator(input_data_definition, basetime=None, validtime=None, provider_id='symlink')[source]

Handle input data spec dict.

Loop through the defined data types and fetch them.

Parameters:

input_data_definition (dict) – Input data spec
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.
provider_id (str) – Provider id. Defaults to “symlink”

Raises:

RuntimeError – “Invalid data handle type”

output(target, destination, basetime=None, validtime=None, archive=False, provider_id='move')[source]

Set output data from tactus.

Parameters:

target (str) – Input file pattern
destination (str) – Destination file pattern
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.
archive (bool, optional) – Also archive data. Defaults to False.
provider_id (str, optional) – Provider ID. Defaults to “move”.

set_resources_from_dict(res_dict)[source]

Set resources from dict.

Parameters:: res_dict (_type_) – _description_
Raises:: ValueError – If the passed file type is neither ‘input’ nor ‘output’.

class LocalFileOnDisk(config, pattern, basetime=None, validtime=None)[source]

Bases: Resource

Local file on disk.

class LocalFileSystemCopy(config, pattern, fetch=True)[source]

Bases: Provider

Local file system copy.

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool

class LocalFileSystemMove(config, pattern, fetch=False)[source]

Bases: Provider

Local file system move.

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool

class LocalFileSystemSymlink(config, pattern, fetch=True)[source]

Bases: Provider

Local file system.

create_resource(resource)[source]

Symlink the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool

class Platform(config)[source]

Bases: object

Platform.

evaluate(command_string: str, object_: str | object) → Any[source]

Evaluate command string, by applying corresponding command of object.

Parameters:

command_string (str) – Command string to evaluate
object (Union[str, object]) – Object to apply command from (if command is function of object). If str, the object is assumed to be a module. If a class, the command is assumed to be a method of the class.

Returns:

Return original command string if it is not a function call,: otherwise return the result of the function call.

Return type:

any

Raises:

ModuleNotFoundError – If module object_ not found
AttributeError – If module/class object_ has no attribute named func
TypeError – If object is not a class or a string
TypeError – If the command to evaluate is not a function

expand_macros(macros)[source]

Check and expand macros.

Parameters:: macros (dict) – Input macros
Returns:: Stored macros
Return type:: out_macros (dict)

fill_each_macro(macro_config)[source]: Fill each of the macros.

fill_macros()[source]: Fill the macros.

get_gen_macros()[source]

Get the environment macros.

Returns:: Environment macros to be used.
Return type:: dict

get_os_macros()[source]

Get the environment macros.

Returns:: Environment macros to be used.
Return type:: dict

get_platform()[source]

Get the platform.

Returns:: Platform specifc values.
Return type:: dict

get_platform_value(role, alt=None)[source]

Get the path.

Parameters:

role (str) – Type of variable to substitute
alt (str) – Alternative return value

Returns:

Value from platform.[role]

Return type:

str

get_provider(provider_id, target, fetch=True)[source]

Get the needed provider.

Parameters:

provider_id (str) – The intent of the provider.
target (Resource) – The target.
fetch (boolean) – Fetch the file or store it. Default to True.

Returns:

Provider

Return type:

Provider

Raises:

NotImplementedError – If provider not defined.

get_system_macros()[source]

Get the macros.

Returns:: Macros to define.
Return type:: dict

get_system_value(role)[source]

Get the system value.

Parameters:: role (str) – Type of variable to substitute
Returns:: Value from system.[role]
Return type:: str

get_value(setting, alt=None)[source]

Get the config value with substition.

Parameters:

setting (str) – Type of variable to substitute
alt (str) – Alternative return value

Returns:

Value from config with substituted variables

Return type:

str

resolve_macros(config_dict, keyval=None)[source]: Resolve all macros in a nested dict (both keys and values!).

store_macro(key, val)[source]

Check and store a macro.

Parameters:

key (str) – Macro key
val (str) – Macro value

Raises:

RuntimeError – If macro already exists

sub_str_dict(input_dict, basetime=None, validtime=None)[source]

Substitute strings in dictionary.

Parameters:

input_dict (dict) – Dict to be parsed
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.

Returns:

Updated dict

Return type:

d (dict)

sub_value(pattern, key, value, micro='@', ci=True)[source]

Substitute the value case-insensitively.

Parameters:

pattern (str) – Input string
key (str) – Key to replace
value (str) – Value to replace
micro (str, optional) – Micro character. Defaults to “@”.
ci (bool, optional) – Case insensitive. Defaults to True.

Returns:

Replaces string

Return type:

str

substitute(pattern, basetime=None, validtime=None, bd_index=None, keyval=None)[source]

Substitute pattern.

Parameters:

pattern (str) – _description_
basetime (datetime.datetime, optional) – Base time. Defaults to None.
validtime (datetime.datetime, optional) – Valid time. Defaults to None.
bd_index (int, optional) – Boundary file index. Defaults to None
keyval (str) – Key associated with pattern

Returns:

Substituted string.

Return type:

str

Raises:

RuntimeError – In case of erroneous macro

substitute_datetime(pattern, datetime, suffix='')[source]

Substitute datetime related properties.

Parameters:

pattern (str) – _description_
datetime (DateTime object) – datetime to treat
suffix (str) – Add on to key

Returns:

Substituted string.

Return type:

str

substitute_duration(pattern, duration, prefix='')[source]

Substitute duration related properties.

Parameters:

pattern (str) – _description_
duration (duration object) – duration to treat
prefix (str) – Add before key

Returns:

Substituted string.

Return type:

str

substitute_tstep(pattern)[source]

Substitute tstep.

Parameters:: pattern (str) – _description_
Returns:: Substituted string.
Return type:: str
Raises:: RuntimeError – In case of erroneous macro

class Provider(config, identifier, fetch=True)[source]

Bases: object

Base provider class.

create_missing_dir(target)[source]

Create a directory if missing.

Parameters:: target (str) – Name of target

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – The resource to be created
Raises:: NotImplementedError – Should be implemented

exception ProviderError[source]

Bases: Exception

Error raised when there are provider-related problems.

class Resource(_config, identifier)[source]

Bases: object

Resource container.

class S3(config, pattern, fetch=True)[source]

Bases: ArchiveProvider

Transfer data with S3.

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool
Raises:: RuntimeError – If resource is not created

class SCP(config, pattern, fetch=True)[source]

Bases: ArchiveProvider

Transfer data with SCP.

create_resource(resource)[source]

Create the resource.

Parameters:: resource (Resource) – Resource.
Returns:: True if success
Return type:: bool
Raises:: RuntimeError – If directory is not created

compute_georef(domain_config)[source]: Computes georef from domain_config.