Initialises a pipeline run and creates the relevant directories.
Usage: ./manage.py initpiperun pipeline_run_name
Command
Bases: BaseCommand
This script initialise the Pipeline Run folder and related config for the pipeline.
Source code in vast_pipeline/management/commands/initpiperun.py
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134 | class Command(BaseCommand):
"""
This script initialise the Pipeline Run folder and related config
for the pipeline.
"""
help = (
'Create the pipeline run folder structure to run a pipeline '
'instance'
)
def add_arguments(self, parser) -> None:
"""
Enables arguments for the command.
Args:
parser (ArgumentParser): The parser object of the command.
Returns:
None
"""
# positional arguments
parser.add_argument(
'runname',
type=str,
help='Name of the pipeline run.'
)
def handle(self, *args, **options) -> None:
"""
Handle function of the command.
Args:
*args: Variable length argument list.
**options: Variable length options.
Returns:
None
"""
# configure logging
if options['verbosity'] > 1:
# set root logger to use the DEBUG level
root_logger = logging.getLogger('')
root_logger.setLevel(logging.DEBUG)
# set the traceback on
options['traceback'] = True
try:
_ = initialise_run(options['runname'])
except Exception as e:
raise CommandError(e)
logger.info((
'pipeline run initialisation successful! Please modify the '
'"config.yaml"'
))
|
add_arguments(parser)
Enables arguments for the command.
Parameters:
| Name | Type | Description | Default |
parser | ArgumentParser | The parser object of the command. | required |
Returns:
Source code in vast_pipeline/management/commands/initpiperun.py
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105 | def add_arguments(self, parser) -> None:
"""
Enables arguments for the command.
Args:
parser (ArgumentParser): The parser object of the command.
Returns:
None
"""
# positional arguments
parser.add_argument(
'runname',
type=str,
help='Name of the pipeline run.'
)
|
handle(*args, **options)
Handle function of the command.
Parameters:
| Name | Type | Description | Default |
*args | | Variable length argument list. | () |
**options | | | {} |
Returns:
Source code in vast_pipeline/management/commands/initpiperun.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134 | def handle(self, *args, **options) -> None:
"""
Handle function of the command.
Args:
*args: Variable length argument list.
**options: Variable length options.
Returns:
None
"""
# configure logging
if options['verbosity'] > 1:
# set root logger to use the DEBUG level
root_logger = logging.getLogger('')
root_logger.setLevel(logging.DEBUG)
# set the traceback on
options['traceback'] = True
try:
_ = initialise_run(options['runname'])
except Exception as e:
raise CommandError(e)
logger.info((
'pipeline run initialisation successful! Please modify the '
'"config.yaml"'
))
|
initialise_run(run_name, run_description=None, user=None, config=None)
Initialise a pipeline run.
Parameters:
| Name | Type | Description | Default |
run_name | str | A unique name for the run. | required |
run_description | Optional[str] | Description for the run, only used if initialised with the web UI. Defaults to None. | None |
user | Optional[User] | User that created the run, only used if initialised with the web UI. Defaults to None. | None |
config | Optional[Dict[str, Any]] | Dictionary of configuration values to pass to the run config template, only used if initialised with the web UI. Defaults to None. | None |
Raises:
Returns:
| Type | Description |
Run | The initialised pipeline Run Django model object. |
Source code in vast_pipeline/management/commands/initpiperun.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77 | def initialise_run(
run_name: str,
run_description: Optional[str] = None,
user: Optional[User] = None,
config: Optional[Dict[str, Any]] = None,
) -> Run:
"""Initialise a pipeline run.
Args:
run_name (str): A unique name for the run.
run_description (Optional[str], optional): Description for the run, only used if
initialised with the web UI. Defaults to None.
user (Optional[User], optional): User that created the run, only used if
initialised with the web UI. Defaults to None.
config (Optional[Dict[str, Any]], optional): Dictionary of configuration values
to pass to the run config template, only used if initialised with the web UI.
Defaults to None.
Raises:
PipelineInitError: `run_name` was not unique.
PipelineInitError: A directory named `run_name` already exists.
Returns:
The initialised pipeline Run Django model object.
"""
# check for duplicated run name
p_run = Run.objects.filter(name__exact=run_name)
if p_run:
msg = 'Pipeline run name already used. Change name'
raise PipelineInitError(msg)
# create the pipeline run folder
run_path = os.path.join(sett.PIPELINE_WORKING_DIR, run_name)
if os.path.exists(run_path):
msg = 'pipeline run path already present!'
raise PipelineInitError(msg)
else:
logger.info('creating pipeline run folder')
os.mkdir(run_path)
# copy default config into the pipeline run folder
logger.info('copying default config in pipeline run folder')
template_kwargs = config if config else sett.PIPE_RUN_CONFIG_DEFAULTS
template_str = make_config_template(
PipelineConfig.TEMPLATE_PATH, run_path=run_path, **template_kwargs
)
with open(os.path.join(run_path, 'config.yaml'), 'w') as fp:
fp.write(template_str)
# create entry in db
p_run, _ = get_create_p_run(run_name, run_path, run_description, user)
return p_run
|