hpcflow package#

Subpackages#

Submodules#

hpcflow.actions module#

class hpcflow.actions.Action(commands, environments, input_file_generators=None, output_file_parsers=None, conditions=None)#

Bases: object

Parameters
classmethod from_spec(spec, all_envs, parameters, cmd_files)#

Parse an Action definition from a JSON-like dict.

Parameters

spec (dict with keys:) –

commands input_file_generators output_file_parsers environments: dict environment: str

Instead of environments (plural), this can be used to specify a single environment to be used for all resolved actions of this action.

get_commands_action_env()#
get_input_file_generator_action_env(input_file_generator)#
Parameters

input_file_generator (hpcflow.command_files.InputFileGenerator) –

get_output_file_parser_action_env(output_file_parser)#
Parameters

output_file_parser (hpcflow.command_files.OutputFileParser) –

get_parameter_dependence(parameter)#

Find if/where a given parameter is used by the action.

Parameters

parameter (hpcflow.parameters.SchemaParameter) –

get_resolved_action_env(relevant_scopes, input_file_generator=None, output_file_parser=None, commands=None)#
Parameters
resolve_actions()#
class hpcflow.actions.ActionCondition(path, condition=None)#

Bases: object

Class to represent a condition that must be met if an action is to be included.

Parameters

  • path (List[str]) –

  • condition (Optional[valida.conditions.ConditionLike]) –

Return type

None

condition: Optional[valida.conditions.ConditionLike] = None#
path: List[str]#
class hpcflow.actions.ActionEnvironment(environment: hpcflow.environment.Environment, scope: hpcflow.actions.ActionScope)#

Bases: object

Parameters
Return type

None

environment: hpcflow.environment.Environment#
classmethod from_spec(scope, env, all_envs)#
scope: hpcflow.actions.ActionScope#
class hpcflow.actions.ActionScope(typ, kwargs=<factory>)#

Bases: object

Class to represent the identification of a subset of task schema actions by a filtering process.

Parameters
Return type

None

classmethod input_file_generator(file=None)#
kwargs: Optional[Dict]#
classmethod main()#
classmethod output_file_parser(output=None)#
classmethod processing()#
typ: hpcflow.actions.ActionScopeType#
class hpcflow.actions.ActionScopeType(value)#

Bases: enum.Enum

An enumeration.

ALL = 0#
INPUT_FILE_GENERATOR = 3#
MAIN = 1#
OUTPUT_FILE_PARSER = 4#
PROCESSING = 2#
class hpcflow.actions.CommandsAction(environment, conditions, commands)#

Bases: hpcflow.actions.ResolvedAction

Represents an action without any associated input file generators and output parsers.

Parameters
Return type

None

commands: List[hpcflow.commands.Command]#
class hpcflow.actions.InputFileGeneratorAction(environment: hpcflow.environment.Environment, conditions: List[hpcflow.actions.ActionCondition], input_file_generator: hpcflow.command_files.InputFileGenerator)#

Bases: hpcflow.actions.ResolvedAction

Parameters
Return type

None

input_file_generator: hpcflow.command_files.InputFileGenerator#
class hpcflow.actions.OutputFileParserAction(environment: hpcflow.environment.Environment, conditions: List[hpcflow.actions.ActionCondition], output_file_parser: hpcflow.command_files.OutputFileParser)#

Bases: hpcflow.actions.ResolvedAction

Parameters
Return type

None

output_file_parser: hpcflow.command_files.OutputFileParser#
class hpcflow.actions.ResolvedAction(environment: hpcflow.environment.Environment, conditions: List[hpcflow.actions.ActionCondition])#

Bases: object

Parameters
Return type

None

conditions: List[hpcflow.actions.ActionCondition]#
environment: hpcflow.environment.Environment#

hpcflow.app_log module#

class hpcflow.app_log.AppLog(package_name, hpcflow_app_log=None)#

Bases: object

DEFAULT_LOG_FILE_PATH = 'app.log'#
DEFAULT_LOG_LEVEL_CONSOLE = 'WARNING'#
DEFAULT_LOG_LEVEL_FILE = 'INFO'#
add_file_logger(path, level=None, fmt=None)#
update_handlers(log_console_level=None, log_file_level=None, log_file_path=None, from_CLI=False)#

Modify logging configuration if non-None arguments are passed.

Parameters

from_CLI – If True, disallow future modifications of the specified parameters. This is to allow the CLI passed parameters to override those specified in the config file.

hpcflow.command_files module#

class hpcflow.command_files.FileNameExt(file_name: hpcflow.command_files.FileNameSpec)#

Bases: object

Parameters

file_name (hpcflow.command_files.FileNameSpec) –

Return type

None

file_name: hpcflow.command_files.FileNameSpec#
value(directory=None)#
class hpcflow.command_files.FileNameSpec(name, args=None, is_regex=False)#

Bases: object

classmethod from_spec(spec)#
value(directory=None)#
class hpcflow.command_files.FileNameStem(file_name: hpcflow.command_files.FileNameSpec)#

Bases: object

Parameters

file_name (hpcflow.command_files.FileNameSpec) –

Return type

None

file_name: hpcflow.command_files.FileNameSpec#
value(directory=None)#
class hpcflow.command_files.FileSpec(label: str, name: str)#

Bases: object

Parameters

  • label (str) –

  • name (str) –

Return type

None

classmethod from_spec(spec)#
label: str#
name: str#
value(directory=None)#
class hpcflow.command_files.InputFile(file, path=None, contents=None, extension='')#

Bases: hpcflow.command_files._FileContentsSpecifier

Parameters
class hpcflow.command_files.InputFileGenerator(input_file: hpcflow.command_files.FileSpec, inputs: List[hpcflow.parameters.Parameter], environment: hpcflow.environment.Environment = None)#

Bases: object

Parameters
Return type

None

environment: hpcflow.environment.Environment = None#
classmethod from_spec(label, info, parameters, cmd_files)#
input_file: hpcflow.command_files.FileSpec#
inputs: List[hpcflow.parameters.Parameter]#
class hpcflow.command_files.InputFileGeneratorSource(generator, path=None, contents=None, extension='')#

Bases: hpcflow.command_files._FileContentsSpecifier

Parameters
class hpcflow.command_files.OutputFileParser(output: hpcflow.parameters.Parameter, output_files: List[hpcflow.command_files.FileSpec], environment: hpcflow.environment.Environment = None, inputs: List[str] = None, options: Dict = None)#

Bases: object

Parameters
Return type

None

environment: hpcflow.environment.Environment = None#
classmethod from_spec(param_typ, info, parameters, cmd_files)#
inputs: List[str] = None#
options: Dict = None#
output: hpcflow.parameters.Parameter#
output_files: List[hpcflow.command_files.FileSpec]#
class hpcflow.command_files.OutputFileParserSource(parser, path=None, contents=None, extension='')#

Bases: hpcflow.command_files._FileContentsSpecifier

Parameters

hpcflow.commands module#

class hpcflow.commands.Command(command: str, arguments: List[Any] = None, stdout: str = None, stderr: str = None, stdin: str = None)#

Bases: object

Parameters

  • command (str) –

  • arguments (List[Any]) –

  • stdout (str) –

  • stderr (str) –

  • stdin (str) –

Return type

None

arguments: List[Any] = None#
command: str#
classmethod from_spec(spec)#
stderr: str = None#
stdin: str = None#
stdout: str = None#
class hpcflow.commands.CommandArgument(parts)#

Bases: object

Parameters

parts (List[Any]) –

Return type

None

parts#
Type

list of any of str, File, Parameter

parts: List[Any]#

hpcflow.config module#

class hpcflow.config.ConfigLoader(*args, **kwargs)#

Bases: object

Singleton class used to load the configuration from a YAML file.

_configurable_keys#
Type

list

_meta_data#
Type

dict

_configured_data#
Type

dict

_with_config#
Type

dict

_config_schemas#
Type

list of valida.schema.Schema

_cfg_file_dat_rt#
Type

dict

CONFIG_DIR_DEFAULT = '~/.hpcflow'#
CONFIG_DIR_ENV_VAR = 'HPCFLOW_CONFIG_DIR'#
DEFAULT_CONFIG = {'configs': {'default': {'config': {'environment_sources': [], 'log_file_path': 'logs/app.log', 'machine': 'fv-az77-645', 'task_schema_sources': [], 'telemetry': True}, 'invocation': {'environment_setup': None, 'match': {}}}}}#
SENTRY_DSN = 'https://2463b288fd1a40f4bada9f5ff53f6811@o1180430.ingest.sentry.io/6293231'#
SENTRY_ENV = 'main'#
SENTRY_TRACES_SAMPLE_RATE = 1.0#
append_value(name, value, is_json)#

Append a new value to an existing configuration list item.

Parameters

  • name – The name of a configurable configuration item to modify.

  • value – The value to append to the configuration item.

  • is_json – If True, value will first be JSON-decoded.

Returns

The new value that has been modified.

Return type

new_value

check_data_files()#

Check task schema and environment files specified in the config file exist/are reachable.

get(name, file_data=False)#

Get a configuration item.

Parameters

  • name (str) – The name of a configuration item (either a configurable key or a meta-data key)

  • file_data (bool) – If True, retrieve the value from the file data (_cfg_file_dat_rt) rather than the meta data (_meta_data) or configured data (_configured_data).

Return type

The value of the specified configuration item.

get_configurable()#

Get a list of all configurable keys.

pop_value(name, index)#

Pop (remove) an item from an existing configuration list item.

Parameters

  • name (str) – The name of a configurable configuration item to modify.

  • index (int) – The integer index (zero-based) of the item to remove.

Returns

The new value that has been modified.

Return type

new_value

prepend_value(name, value, is_json)#

Prepend a new value to an existing configuration list item.

Parameters

  • name – The name of a configurable configuration item to modify.

  • value – The value to prepend to the configuration item.

  • is_json – If True, value will first be JSON-decoded.

Returns

The new value that has been modified.

Return type

new_value

set_value(name, value, is_json)#

Set the value of a configuration item.

Parameters

  • name (str) – The name of a configurable configuration item to set.

  • value (Any) – The value to set the configuration item to.

  • is_json (bool) – If True, value will first be JSON-decoded.

Returns

The new value that has been set.

Return type

new_value

to_string(exclude=None, just_meta=False)#

Format the instance in a string, optionally exclude some keys.

Parameters

  • exclude (Optional[List]) – List of keys to exclude. Optional.

  • just_meta – If True, just return a str of the meta-data. This is useful to show during initialisation, in the case where the configuration is otherwise invalid.

unset_value(name)#

Unset (remove) the value of a configuration item.

Parameters

name (str) – The name of a configurable configuration item to unset.

Return type

True

class hpcflow.config.ConfigType#

Bases: type

Metaclass for the Config convenience class that allows access to the ConfigLoader singleton instance via the Config class.

to_string(exclude=None)#

hpcflow.core module#

class hpcflow.core.Task(schema, parameter_values)#

Bases: object

class hpcflow.core.TaskSchema(parameters)#

Bases: object

class hpcflow.core.Workflow(tasks)#

Bases: object

class hpcflow.core.WorkflowInteraction(value)#

Bases: enum.Enum

An enumeration.

CREATE = 0#
hpcflow.core.make_workflow()#

hpcflow.element module#

class hpcflow.element.Element(task: hpcflow.typing_stubs.Task, inputs: List[hpcflow.typing_stubs.Parameter], outputs: List[hpcflow.typing_stubs.Parameter])#

Bases: object

Parameters
Return type

None

inputs: List[hpcflow.typing_stubs.Parameter]#
outputs: List[hpcflow.typing_stubs.Parameter]#
task: hpcflow.typing_stubs.Task#
class hpcflow.element.ElementFilter(parameter_path: hpcflow.typing_stubs.ParameterPath, condition: valida.conditions.ConditionLike)#

Bases: object

Parameters
Return type

None

condition: valida.conditions.ConditionLike#
classmethod from_spec(spec)#
parameter_path: hpcflow.typing_stubs.ParameterPath#
class hpcflow.element.ElementGroup(name: str, where: Optional[hpcflow.element.ElementFilter] = None, group_by_distinct: Optional[hpcflow.typing_stubs.ParameterPath] = None)#

Bases: object

Parameters
Return type

None

group_by_distinct: Optional[hpcflow.typing_stubs.ParameterPath] = None#
name: str#
where: Optional[hpcflow.element.ElementFilter] = None#
class hpcflow.element.ElementRepeats(number: int, where: Optional[hpcflow.element.ElementFilter] = None)#

Bases: object

Parameters
Return type

None

number: int#
where: Optional[hpcflow.element.ElementFilter] = None#

hpcflow.environment module#

class hpcflow.environment.Environment(name: str, setup: Optional[Sequence] = None, executables: Optional[List[hpcflow.environment.Executable]] = <factory>)#

Bases: object

Parameters
Return type

None

executables: Optional[List[hpcflow.environment.Executable]]#
classmethod from_spec(spec)#
name: str#
setup: Optional[Sequence] = None#
class hpcflow.environment.Executable(label: str, instances: List[hpcflow.environment.ExecutableInstance] = <factory>, environment: Any = None)#

Bases: object

Parameters
Return type

None

environment: Any = None#
filter_instances(parallel_mode=None, num_cores=None)#
classmethod from_spec(spec)#
instances: List[hpcflow.environment.ExecutableInstance]#
label: str#
class hpcflow.environment.ExecutableInstance(parallel_mode: str, num_cores: Any, command: str)#

Bases: object

Parameters

  • parallel_mode (str) –

  • num_cores (Any) –

  • command (str) –

Return type

None

command: str#
classmethod from_spec(spec)#
num_cores: Any#
parallel_mode: str#
class hpcflow.environment.NumCores(start: int, stop: int, step: int = None)#

Bases: object

Parameters

  • start (int) –

  • stop (int) –

  • step (int) –

Return type

None

start: int#
step: int = None#
stop: int#

hpcflow.errors module#

exception hpcflow.errors.ConfigChangeFileUpdateError(name, err, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when the updating of the config YAML file fails.

exception hpcflow.errors.ConfigChangeInvalidError(name, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when trying to set an invalid key in the Config.

exception hpcflow.errors.ConfigChangeInvalidJSONError(name, json_str, err, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when attempting to set a config key using an invalid JSON string.

exception hpcflow.errors.ConfigChangePopIndexError(name, length, index, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when trying to pop an item from a config item with an invalid index.

exception hpcflow.errors.ConfigChangeTypeInvalidError(name, typ, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when trying to modify a config item using a list operation, when the config item is not a list.

exception hpcflow.errors.ConfigChangeValidationError(name, validation_err, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when a change to the configurable data would invalidate the config.

exception hpcflow.errors.ConfigError#

Bases: Exception

Raised when a valid configuration can not be associated with the current invocation.

exception hpcflow.errors.ConfigFileInvocationIncompatibleError(message='No config could be found that matches the current invocation.')#

Bases: hpcflow.errors.ConfigError

Raised when, given the run time information of the invocation, no compatible configuration can be found in the config file.

exception hpcflow.errors.ConfigFileValidationError#

Bases: hpcflow.errors.ConfigError

exception hpcflow.errors.ConfigUnknownItemError#

Bases: hpcflow.errors.ConfigError

exception hpcflow.errors.ConfigValidationError(message, meta_data=None)#

Bases: hpcflow.errors.ConfigError

Raised when the matching config data is invalid.

exception hpcflow.errors.DuplicateExecutableError#

Bases: ValueError

exception hpcflow.errors.EnvironmentSpecValidationError#

Bases: Exception

exception hpcflow.errors.InputSourceValidationError#

Bases: Exception

exception hpcflow.errors.InputValueDuplicateSequenceAddress#

Bases: ValueError

exception hpcflow.errors.InvalidIdentifier#

Bases: ValueError

exception hpcflow.errors.MissingActionEnvironment#

Bases: Exception

exception hpcflow.errors.MissingActionsError#

Bases: ValueError

exception hpcflow.errors.MissingCompatibleActionEnvironment#

Bases: Exception

exception hpcflow.errors.MissingEnvironmentFileError(file_name, err, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when an environment file specified in the config file does not exist.

exception hpcflow.errors.MissingInputs#

Bases: Exception

exception hpcflow.errors.MissingTaskSchemaFileError(file_name, err, message=None)#

Bases: hpcflow.errors.ConfigError

Raised when a task schema file specified in the config file does not exist.

exception hpcflow.errors.TaskSchemaSpecValidationError#

Bases: Exception

exception hpcflow.errors.TaskTemplateInvalidNesting#

Bases: ValueError

exception hpcflow.errors.TaskTemplateMultipleInputValues#

Bases: ValueError

exception hpcflow.errors.TaskTemplateMultipleSchemaObjectives#

Bases: ValueError

exception hpcflow.errors.TaskTemplateUnexpectedInput#

Bases: ValueError

exception hpcflow.errors.WorkflowSpecValidationError#

Bases: Exception

hpcflow.gui module#

hpcflow.hpcflow module#

class hpcflow.hpcflow.HPCFlowApp(name, version, description=None)#

Bases: object

Class for instantiating an HPCFlow application, which may provide, for instance, custom parameter logic, over the top of that provided by hpcflow.

Parameters

  • name (str) –

  • version (str) –

  • description (Optional[str]) –

Return type

None

description: Optional[str] = None#
make_CLI()#
name: str#
version: str#

hpcflow.loop module#

class hpcflow.loop.Loop(parameter: hpcflow.parameters.Parameter, stopping_criteria: hpcflow.loop.StoppingCriterion, maximum_iterations: int)#

Bases: object

Parameters
Return type

None

maximum_iterations: int#
parameter: hpcflow.parameters.Parameter#
stopping_criteria: hpcflow.loop.StoppingCriterion#
class hpcflow.loop.StoppingCriterion(parameter: hpcflow.parameters.Parameter, condition: valida.conditions.ConditionLike)#

Bases: object

Parameters
Return type

None

condition: valida.conditions.ConditionLike#
parameter: hpcflow.parameters.Parameter#

hpcflow.object_list module#

class hpcflow.object_list.DotAccessObjectList(*objects, access_attribute, descriptor)#

Bases: object

add_object(obj, index=- 1)#
class hpcflow.object_list.ExecutablesList(*execs)#

Bases: hpcflow.object_list.DotAccessObjectList

A list-like container for environment executables with dot-notation access by executable label.

class hpcflow.object_list.GroupList(*groups)#

Bases: hpcflow.object_list.DotAccessObjectList

A list-like container for the task schema group list with dot-notation access by group name.

class hpcflow.object_list.TaskList(*task_likes)#

Bases: hpcflow.object_list.DotAccessObjectList

A list-like container for a task-like list with dot-notation access by task unique-name.

hpcflow.object_list.index(obj_lst, obj)#

hpcflow.parameters module#

class hpcflow.parameters.AbstractInputValue#

Bases: object

Class to represent all sequence-able inputs to a task.

Return type

None

class hpcflow.parameters.BuiltinSchemaParameter#

Bases: object

Return type

None

class hpcflow.parameters.InputSource(source: str, where: Optional[hpcflow.element.ElementFilter] = None)#

Bases: object

Parameters
Return type

None

classmethod from_spec(spec)#
property imports_ref#
source: str#
property source_type#
property task_ref#
property task_source_type#
validate(schema_input, task_template, workflow_like)#

Check a supplied source is valid for a given workflow (template).

Parameters
where: Optional[hpcflow.element.ElementFilter] = None#
class hpcflow.parameters.InputValue(parameter: Union[hpcflow.parameters.Parameter, hpcflow.parameters.SchemaInput], path: Optional[Sequence[Union[str, int, float]]] = None, value: Optional[Any] = None)#

Bases: hpcflow.parameters.AbstractInputValue

Parameters
Return type

None

classmethod from_spec(spec, all_parameters)#
property is_sub_value#

True if the value is for a sub part of the parameter (i.e. if path is set). Sub-values are not added to the base parameter data, but are interpreted as single-value sequences.

parameter: Union[hpcflow.parameters.Parameter, hpcflow.parameters.SchemaInput]#
path: Optional[Sequence[Union[str, int, float]]] = None#
value: Optional[Any] = None#
class hpcflow.parameters.Parameter(typ: str, is_file: bool = False, sub_parameters: List[hpcflow.typing_stubs.SubParameter] = <factory>)#

Bases: object

Parameters
Return type

None

classmethod from_spec(spec)#
is_file: bool = False#
sub_parameters: List[hpcflow.typing_stubs.SubParameter]#
typ: str#
class hpcflow.parameters.ParameterPath(path: Sequence[Union[str, int, float]], task: Union[hpcflow.typing_stubs.TaskTemplate, hpcflow.typing_stubs.TaskSchema, NoneType] = None)#

Bases: object

Parameters
Return type

None

path: Sequence[Union[str, int, float]]#
task: Optional[Union[hpcflow.typing_stubs.TaskTemplate, hpcflow.typing_stubs.TaskSchema]] = None#
class hpcflow.parameters.ParameterPropagationMode(value)#

Bases: enum.Enum

An enumeration.

EXPLICIT = 1#
IMPLICIT = 0#
NEVER = 2#
class hpcflow.parameters.ResourceSpec#

Bases: hpcflow.parameters.AbstractInputValue

Return type

None

class hpcflow.parameters.SchemaInput(parameter, default_value=None, propagation_mode=ParameterPropagationMode.IMPLICIT, group=None, where=None)#

Bases: hpcflow.parameters.SchemaParameter

A Parameter as used within a particular schema, for which a default value may be applied.

Parameters
Return type

None

default_value: Optional[hpcflow.typing_stubs.InputValue] = None#
classmethod from_spec(spec, parameters)#
group: Optional[str] = None#
property input_or_output#
parameter: hpcflow.parameters.Parameter#
propagation_mode: hpcflow.parameters.ParameterPropagationMode = 0#
where: Optional[hpcflow.element.ElementFilter] = None#
class hpcflow.parameters.SchemaOutput(parameter, propagation_mode=ParameterPropagationMode.IMPLICIT)#

Bases: hpcflow.parameters.SchemaParameter

A Parameter as outputted from particular task.

Parameters
Return type

None

property input_or_output#
parameter: hpcflow.parameters.Parameter#
propagation_mode: hpcflow.parameters.ParameterPropagationMode = 0#
class hpcflow.parameters.SchemaParameter#

Bases: object

Return type

None

classmethod from_spec(spec, parameters)#
property name#
property typ#
class hpcflow.parameters.SubParameter(address: List[Union[int, float, str]], parameter: hpcflow.parameters.Parameter)#

Bases: object

Parameters
Return type

None

address: List[Union[int, float, str]]#
parameter: hpcflow.parameters.Parameter#
class hpcflow.parameters.ValuePerturbation(name: str, path: Optional[Sequence[Union[str, int, float]]] = None, multiplicative_factor: Union[int, float, numpy.number, NoneType] = 1, additive_factor: Union[int, float, numpy.number, NoneType] = 0)#

Bases: hpcflow.parameters.AbstractInputValue

Parameters

  • name (str) –

  • path (Optional[Sequence[Union[str, int, float]]]) –

  • multiplicative_factor (Optional[Union[int, float, numpy.number]]) –

  • additive_factor (Optional[Union[int, float, numpy.number]]) –

Return type

None

additive_factor: Optional[Union[int, float, numpy.number]] = 0#
classmethod from_spec(spec)#
multiplicative_factor: Optional[Union[int, float, numpy.number]] = 1#
name: str#
path: Optional[Sequence[Union[str, int, float]]] = None#
class hpcflow.parameters.ValueSequence(path: Sequence[Union[str, int, float]], values: List[Any], nesting_order: int)#

Bases: object

Parameters

  • path (Sequence[Union[str, int, float]]) –

  • values (List[Any]) –

  • nesting_order (int) –

Return type

None

check_address_exists(value)#

Check a given nested dict/list “address” resolves to somewhere within value.

classmethod from_linear_space(start, stop, num=50, address=None, **kwargs)#
classmethod from_range(start, stop, step=1, address=None)#
classmethod from_spec(spec)#
nesting_order: int#
path: Sequence[Union[str, int, float]]#
values: List[Any]#

hpcflow.runtime module#

class hpcflow.runtime.RunTimeInfo(name, version)#

Bases: hpcflow.utils.PrettyPrinter

Get useful run-time information, including the executable name used to invoke the CLI, in the case a PyInstaller-built executable was used.

sys_prefix#

From sys.prefix. If running in a virtual environment, this will point to the environment directory. If not running in a virtual environment, this will point to the Python installation root.

Type

str

sys_base_prefix#

From sys.base_prefix. This will be equal to sys_prefix (sys.prefix) if not running within a virtual environment. However, if running within a virtual environment, this will be the Python installation directory, and sys_prefix will be equal to the virtual environment directory.

Type

str

get_activate_env_command()#
get_deactivate_env_command()#

hpcflow.server module#

hpcflow.spec_parse module#

Module containing logic for parsing workflow spec files/strings.

hpcflow.spec_parse.get_environments()#
hpcflow.spec_parse.get_task_schemas_and_parameters()#
hpcflow.spec_parse.parse_YAML_spec_file(yaml_file)#

Generate a WorkflowTemplate from a YAML string.

hpcflow.spec_parse.parse_YAML_spec_str(yaml_str)#

Generate a WorkflowTemplate from a YAML string.

hpcflow.spec_schema module#

Module containing a Valida schema for the workflow spec file.

hpcflow.task module#

class hpcflow.task.Task(template, workflow, initial_index)#

Bases: object

Class to represent a Task as positioned within a Workflow or WorkflowTemplate.

Parameters
property element_indices#
property index#

Zero-based position within the workflow. Uses initial index if appending to the workflow is not complete.

property num_elements#
property template#
property unique_name#
property workflow#
class hpcflow.task.TaskTemplate(schemas, repeats=1, resources=None, inputs=None, input_files=None, input_file_generator_sources=None, output_file_parser_sources=None, perturbations=None, sequences=None, input_sources=None, nesting_order=None, groups=None)#

Bases: object

Parametrisation of an isolated task for which a subset of input values are given “locally”. The remaining input values are expected to be satisfied by other tasks/imports in the workflow.

Parameters
add_group(name, where, group_by_distinct)#
Parameters
property all_schema_input_types#

Get the set of all schema input types (over all specified schemas).

property all_schema_inputs: Tuple[hpcflow.parameters.SchemaInput]#
property all_schema_output_types#

Get the set of all schema output types (over all specified schemas).

property all_schema_outputs: Tuple[hpcflow.parameters.SchemaOutput]#
property defined_input_types#
classmethod from_spec(spec, all_schemas, all_parameters)#
get_input_multiplicities(missing_multiplicities=None)#

Get multiplicities for all inputs.

get_non_sub_parameter_input_values()#
get_sub_parameter_input_values()#
property groups#
property input_file_generator_sources#
property input_files#
property input_sources#
property inputs#
property name#
property nesting_order#
property non_universal_input_types#

Get input types for each schema that are non-universal.

property objective#
property output_file_parser_sources#
property perturbations#
property provides_parameters#
property repeats#
property resources#
property schemas#
property sequences#
property undefined_input_types#
property undefined_inputs#
property universal_input_types#

Get input types that are associated with all schemas

property unsourced_inputs#

Get schema input types for which no input sources are currently specified.

hpcflow.task_schema module#

class hpcflow.task_schema.TaskObjective(name: str)#

Bases: object

Parameters

name (str) –

Return type

None

name: str#
class hpcflow.task_schema.TaskSchema(objective: Union[hpcflow.task_schema.TaskObjective, str], actions: List[hpcflow.actions.Action], method: Optional[str] = None, implementation: Optional[str] = None, inputs: Optional[List[Union[hpcflow.parameters.Parameter, hpcflow.parameters.SchemaInput]]] = <factory>, outputs: Optional[List[Union[hpcflow.parameters.Parameter, hpcflow.parameters.SchemaOutput]]] = <factory>)#

Bases: object

Parameters
Return type

None

actions: List[hpcflow.actions.Action]#
classmethod from_spec(spec, all_parameters, environments, cmd_files)#
get_parameter_dependence(parameter)#

Find if/where a given parameter is used by the schema’s actions.

Parameters

parameter (hpcflow.parameters.SchemaParameter) –

implementation: Optional[str] = None#
property input_types#
inputs: Optional[List[Union[hpcflow.parameters.Parameter, hpcflow.parameters.SchemaInput]]]#
method: Optional[str] = None#
objective: Union[hpcflow.task_schema.TaskObjective, str]#
property output_types#
outputs: Optional[List[Union[hpcflow.parameters.Parameter, hpcflow.parameters.SchemaOutput]]]#
property provides_parameters#

hpcflow.typing_stubs module#

class hpcflow.typing_stubs.Element#

Bases: object

class hpcflow.typing_stubs.InputValue#

Bases: object

class hpcflow.typing_stubs.Parameter#

Bases: object

class hpcflow.typing_stubs.ParameterPath#

Bases: object

class hpcflow.typing_stubs.SubParameter#

Bases: object

class hpcflow.typing_stubs.Task#

Bases: object

class hpcflow.typing_stubs.TaskSchema#

Bases: object

class hpcflow.typing_stubs.TaskTemplate#

Bases: object

class hpcflow.typing_stubs.Workflow#

Bases: object

class hpcflow.typing_stubs.WorkflowTemplate#

Bases: object

hpcflow.utils module#

class hpcflow.utils.PrettyPrinter#

Bases: object

class hpcflow.utils.Singleton#

Bases: type

hpcflow.utils.check_valid_py_identifier(name)#

Check a string is (roughly) a valid Python variable identifier and if so return it in lower-case.

Notes

Will be used for:

  • task objective name

  • task method

  • task implementation

  • parameter type

  • parameter name

  • loop name

  • element group name

class hpcflow.utils.classproperty(f)#

Bases: object

hpcflow.utils.get_duplicate_items(lst)#

Get a list of all items in an iterable that appear more than once, assuming items are hashable.

Examples

>>> get_duplicate_items([1, 1, 2, 3])
[1]

>>> get_duplicate_items([1, 2, 3])
[]

>>> get_duplicate_items([1, 2, 3, 3, 3, 2])
[2, 3, 2]
hpcflow.utils.get_in_container(cont, path)#
hpcflow.utils.get_relative_path(path1, path2)#

Get relative path components between two paths.

Parameters

  • path1 (tuple of (str or int or float) of length N) –

  • path2 (tuple of (str or int or float) of length less than or equal to N) –

Returns

relative_path – The path components in path1 that are not in path2.

Return type

tuple of (str or int or float)

Raises

ValueError – If the two paths do not share a common ancestor of path components, or if path2 is longer than path1.

Notes

This function behaves like a simplified PurePath(*path1).relative_to(PurePath(*path2)) from the pathlib module, but where path components can include non-strings.

Examples

>>> get_relative_path(('A', 'B', 'C'), ('A',))
('B', 'C')

>>> get_relative_path(('A', 'B'), ('A', 'B'))
()
hpcflow.utils.get_time_stamp()#
hpcflow.utils.group_by_dict_key_values(lst, *keys)#

Group a list of dicts according to specified equivalent key-values.

Parameters

  • lst (list of dict) – The list of dicts to group together.

  • keys (tuple) – Dicts that have identical values for all of these keys will be grouped together into a sub-list.

Returns

grouped

Return type

list of list of dict

Examples

>>> group_by_dict_key_values([{'a': 1}, {'a': 2}, {'a': 1}], 'a')
[[{'a': 1}, {'a': 1}], [{'a': 2}]]
hpcflow.utils.make_workflow_id()#
hpcflow.utils.search_dir_files_by_regex(pattern, group=0, directory='.')#
hpcflow.utils.sentry_wrap(name, transaction_op=None, span_op=None)#
hpcflow.utils.set_in_container(cont, path, value)#

hpcflow.validation module#

hpcflow.validation.get_schema(filename)#

hpcflow.workflow module#

class hpcflow.workflow.Workflow(tasks: List[hpcflow.task.Task])#

Bases: object

Parameters

tasks (List[hpcflow.task.Task]) –

Return type

None

add_submission(filter)#
rename(new_name)#
tasks: List[hpcflow.task.Task]#
class hpcflow.workflow.WorkflowBlueprint(workflow_template)#

Bases: object

For pre-built workflows that are simpler to parametrise (e.g. fitting workflows).

Parameters

workflow_template (hpcflow.workflow.WorkflowTemplate) –

Return type

None

workflow_template: hpcflow.workflow.WorkflowTemplate#
class hpcflow.workflow.WorkflowTemplate(task_templates=None, loops=None)#

Bases: object

Parameters
add_task(task_template)#
Parameters

task_template (hpcflow.task.TaskTemplate) –

add_task_after(task)#
add_task_before(task)#
ensure_input_sources(new_task, new_index)#

Check valid input sources are specified for a new task to be added to the workflow in a given position. If none are specified, set them according to the default behaviour.

Parameters
classmethod from_spec(spec, all_schemas, all_parameters)#
get_input_value(task_index, element_index, parameter_path)#
get_input_values(task_index, parameter_path)#

Get the value of an input for each element in a task.

get_possible_input_sources(schema_input, new_task, new_index)#

Enumerate the possible sources for an input of a new task, given a proposed placement of that task.

Parameters
make_workflow(path)#
remove_task(task)#
static resolve_initial_elements(multi)#
Parameters

multi (list of dict) –

Each list item represents a sequence of values with keys:

multiplicity: int nesting_order: int address : str

Module contents#

hpcflow.set_config_dir(config_dir)#

previous

API reference

next

hpcflow.api package