"""RUN LOG HANDLING.
This module defines methods for creating a run log.
:Author: Samuel Farrens <samuel.farrens@cea.fr>
"""
import numpy as np
from shapepipe.pipeline.shared import split_module_run
[docs]class RunLog(object):
"""Run Log Class.
This class manages the run log for ShapePipe.
Parameters
----------
run_log_file : str
Name of the run log file
module_list : list
List of modules to be run
current_run : str
Name of the current run
"""
def __init__(self, run_log_file, module_list, current_run):
self.run_log_file = run_log_file
self._module_list = ','.join(module_list)
self.current_run = current_run
self._write()
self._runs = get_list(run_log_file)
[docs] def _write(self):
"""Write.
Write current run to the run log.
"""
with open(self.run_log_file, 'a') as run_log:
run_log.write(f'{self.current_run} {self._module_list}\n')
[docs] def get_run(self, search_string):
"""Get Run.
Get a specific run that matches the input search string.
Parameters
----------
search_string : str
Pattern to match to runs
Returns
-------
str
The run matching the search string
Raises
------
RuntimeError
If no runs found that match the search string
RuntimeError
If more than one run is found matches the search string
"""
runs = [run for run in self._runs if search_string in run]
if len(runs) < 1:
raise RuntimeError(
f'No runs found matching search string \'{search_string}\'.'
)
elif len(runs) > 1:
raise RuntimeError(
'More than one run found matching search string '
+ f'\'{search_string}\''
)
return runs[0].split(' ')[0]
[docs]def get_list(run_log_file):
"""Get List.
Get a list of all runs in the run log.
Parameters
----------
run_log_file : str
Run log file name
Returns
-------
list
Run log file entries
"""
with open(run_log_file, 'r') as run_log:
lines = run_log.readlines()
runs = [line.rstrip() for line in lines]
return runs
[docs]def get_all(runs, module):
"""Get All.
Get all previous pipeline runs of a given model.
Parameters
----------
runs : list
Log file entries consisting of directory and module(s)
module : str
Module name
Raises
------
RuntimeError
If no previous runs are found
Returns
-------
list
All run paths for a given module
"""
module_base, _ = split_module_run(module)
all_runs = [
run for run in runs
if module_base in run.split()[1].split(',')
]
if len(all_runs) == 0:
raise RuntimeError(
f'No previous run of module \'{module_base}\' found'
)
all_runs = all_runs[::-1]
return all_runs
[docs]def get_last(runs, module):
"""Get Last.
Get the last run of the pipeline for a given module.
Parameters
----------
runs : list
Log file entries consisting of directory and module(s)
module : str
Module name
Returns
-------
str
The last run for a given module
"""
all_runs = get_all(runs, module)
last_run = all_runs[0]
return last_run.split(' ')[0]
[docs]def get_last_dir(run_log_file, module):
"""Get Last Dir.
Return directory path corresponding to last run of given module.
Parameters
----------
run_log_file : str
Run log file name
module : str
Module name
Returns
-------
str
Directory name of last module run
"""
runs = get_list(run_log_file)
all_runs = get_all(runs, module)
last_run = all_runs[0].split(' ')[0]
last_dir = f'{last_run}/{module}/output'
return last_dir