Source code for polaris.provenance

import os
import subprocess
import sys


[docs] def write(work_dir, tasks, config=None): """ Write a file with provenance, such as the git version, conda packages, command, and tasks, to the work directory Parameters ---------- work_dir : str The path to the work directory where the tasks will be set up tasks : dict A dictionary describing all of the tasks and their steps config : polaris.config.PolarisConfigParser Configuration options for this task, a combination of user configs and the defaults for the machine and component """ polaris_git_version = None if os.path.exists('.git'): try: args = ['git', 'describe', '--tags', '--dirty', '--always'] polaris_git_version = subprocess.check_output(args).decode('utf-8') polaris_git_version = polaris_git_version.strip('\n') except subprocess.CalledProcessError: pass if config is None: # this is a call to clean and we don't need to document the component # version component_git_version = None else: component_git_version = _get_component_git_version(config) try: args = ['conda', 'list'] conda_list = subprocess.check_output(args).decode('utf-8') except subprocess.CalledProcessError: conda_list = None calling_command = ' '.join(sys.argv) try: os.makedirs(work_dir) except OSError: pass provenance_path = f'{work_dir}/provenance' if os.path.exists(provenance_path): provenance_file = open(provenance_path, 'a') provenance_file.write('\n') else: provenance_file = open(provenance_path, 'w') provenance_file.write( '**************************************************' '*********************\n' ) if polaris_git_version is not None: provenance_file.write( f'polaris git version: {polaris_git_version}\n\n' ) if component_git_version is not None: provenance_file.write( f'component git version: {component_git_version}\n\n' ) provenance_file.write(f'command: {calling_command}\n\n') provenance_file.write('tasks:\n') for _, task in tasks.items(): prefix = ' ' lines = list() to_print = { 'path': task.path, 'name': task.name, 'component': task.component.name, 'subdir': task.subdir, } for key in to_print: key_string = f'{key}: '.ljust(15) lines.append(f'{prefix}{key_string}{to_print[key]}') lines.append(f'{prefix}steps:') for step in task.steps.values(): if step.name == step.subdir: lines.append(f'{prefix} - {step.name}') else: lines.append(f'{prefix} - {step.name}: {step.subdir}') lines.append('') print_string = '\n'.join(lines) provenance_file.write(f'{print_string}\n') if conda_list is not None: provenance_file.write('conda list:\n') provenance_file.write(f'{conda_list}\n') provenance_file.write( '**************************************************' '*********************\n' ) provenance_file.close()
def _get_component_git_version(config): component_path = config.get('paths', 'component_path') if not os.path.exists(component_path): return None cwd = os.getcwd() os.chdir(component_path) try: args = ['git', 'describe', '--tags', '--dirty', '--always'] component_git_version = subprocess.check_output(args).decode('utf-8') component_git_version = component_git_version.strip('\n') except subprocess.CalledProcessError: component_git_version = None os.chdir(cwd) return component_git_version