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.api module

hpcflow.api.make_workflow(dir)

Make a new workflow, innit.

Parameters

dir (str) – Directory to make new workflow in.

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'
get_child_logger(name)
update_handlers(console_log_level=None, file_log_level=None, file_log_path=None)

Modify logging configuration if non-None arguments are passed.

hpcflow.cli module

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.Config

Bases: object

static get_config_file(config_dir)
static resolve_config_dir(config_dir=None)
static set_config(config_dir=None)

Load configuration from a YAML file.

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.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.MissingInputs

Bases: Exception

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)

Bases: object

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_environment_spec_schema()
hpcflow.spec_parse.get_environments()
hpcflow.spec_parse.get_task_schema_spec_schema()
hpcflow.spec_parse.get_task_schemas_and_parameters()
hpcflow.spec_parse.get_workflow_spec_schema()
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

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.set_in_container(cont, path, value)

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

previous

API reference

next

hpcflow.data package