From 195aa85759b5c5d6ebd59e5731e86d4c09651564 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 13 Nov 2024 16:54:57 +1100 Subject: [PATCH 1/7] Converted webinterface/settings.py --- webinterface/settings.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/webinterface/settings.py b/webinterface/settings.py index 6336dd43b..94b496ca6 100644 --- a/webinterface/settings.py +++ b/webinterface/settings.py @@ -1,4 +1,4 @@ -import os +from Pathlib import Path import environ # Load the Django congig from the .env file @@ -6,8 +6,7 @@ environ.Env.read_env() -# Build paths inside the project like this: os.path.join(BASE_DIR, ...) -BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +BASE_DIR = Pathlib(__file__).resolve().parent[1] # Quick-start development settings - unsuitable for production @@ -215,8 +214,8 @@ STATIC_URL = env('STATIC_URL', cast=str, default='/static/') if BASE_URL: STATIC_URL = '/' + BASE_URL.strip('/') + '/' + STATIC_URL.strip('/') + '/' -STATICFILES_DIRS = env('STATICFILES_DIRS', cast=list, default=[os.path.join(BASE_DIR, 'static')]) -STATIC_ROOT = env('STATIC_ROOT', cast=str, default=os.path.join(BASE_DIR, 'staticfiles')) +STATICFILES_DIRS = env('STATICFILES_DIRS', cast=list, default=[BASE_DIR/'static']) +STATIC_ROOT = env('STATIC_ROOT', cast=str, default=BASE_DIR/'staticfiles')) STATICFILES_STORAGE = 'whitenoise.storage.CompressedManifestStaticFilesStorage' # Logging @@ -278,14 +277,14 @@ # PIPELINE settings # project default folder -PIPELINE_WORKING_DIR = env('PIPELINE_WORKING_DIR', cast=str, default=os.path.join(BASE_DIR, 'pipeline-runs')) +PIPELINE_WORKING_DIR = env('PIPELINE_WORKING_DIR', cast=str, default=BASE_DIR/'pipeline-runs') if '/' not in PIPELINE_WORKING_DIR: - PIPELINE_WORKING_DIR = os.path.join(BASE_DIR, PIPELINE_WORKING_DIR) + PIPELINE_WORKING_DIR = BASE_DIR/PIPELINE_WORKING_DIR # raw image data folder (containing FITS files, selavy, etc) -RAW_IMAGE_DIR = env('RAW_IMAGE_DIR', cast=str, default=os.path.join(BASE_DIR, 'raw-images')) +RAW_IMAGE_DIR = env('RAW_IMAGE_DIR', cast=str, default=BASE_DIR/'raw-images') if '/' not in RAW_IMAGE_DIR: - RAW_IMAGE_DIR = os.path.join(BASE_DIR, RAW_IMAGE_DIR) + RAW_IMAGE_DIR = BASE_DIR/RAW_IMAGE_DIR # extra user-supplied data folder # HOME_DATA_DIR is relative to HOME_DATA_ROOT if HOME_DATA_ROOT is not None From d095377a7b58568440616c48cc1f0ec97d3cc6cd Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 13 Nov 2024 16:56:40 +1100 Subject: [PATCH 2/7] Converted vast_pipeline/management/helpers.py --- vast_pipeline/management/helpers.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/vast_pipeline/management/helpers.py b/vast_pipeline/management/helpers.py index c2bbca310..11d64953a 100644 --- a/vast_pipeline/management/helpers.py +++ b/vast_pipeline/management/helpers.py @@ -2,7 +2,7 @@ Helper functions for the commands. """ -import os +from pathlib import Path import logging from typing import Tuple @@ -26,10 +26,11 @@ def get_p_run_name(name: str, return_folder: bool = False) -> Tuple[str, str]: The run directory (if `return_folder` is set to `True`). """ if '/' in name: - folder = os.path.realpath(name) - run_name = os.path.basename(folder) + folder = Path(name).resolve() + run_name = folder.parent[0] return (run_name, folder) if return_folder else run_name - folder = os.path.join(os.path.realpath(sett.PIPELINE_WORKING_DIR), name) + working_dir = Path(sett.PIPELINE_WORKING_DIR).resolve() + folder = working_dir/name return (name, folder) if return_folder else name From 6a3a156ae73150e88fab1436b85d3a79c021c444 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 13 Nov 2024 16:59:10 +1100 Subject: [PATCH 3/7] Converted vast_pipeline/management/commands/initpiperun.py --- vast_pipeline/management/commands/initpiperun.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/vast_pipeline/management/commands/initpiperun.py b/vast_pipeline/management/commands/initpiperun.py index 7097ff96c..7ab0f2fe3 100644 --- a/vast_pipeline/management/commands/initpiperun.py +++ b/vast_pipeline/management/commands/initpiperun.py @@ -4,7 +4,7 @@ Usage: ./manage.py initpiperun pipeline_run_name """ -import os +from Pathlib import Path import logging from typing import Any, Dict, Optional @@ -53,14 +53,14 @@ def initialise_run( raise PipelineInitError(msg) # create the pipeline run folder - run_path = os.path.join(sett.PIPELINE_WORKING_DIR, run_name) + run_path = Path(PIPELINE_WORKING_DIR)/run_name - if os.path.exists(run_path): + if run_path.exists(): msg = 'pipeline run path already present!' raise PipelineInitError(msg) else: logger.info('creating pipeline run folder') - os.mkdir(run_path) + run_path.mkdir() # copy default config into the pipeline run folder logger.info('copying default config in pipeline run folder') @@ -71,7 +71,7 @@ def initialise_run( 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: + with open(run_path/'config.yaml', 'w') as fp: fp.write(template_str) # create entry in db From 6740247975ad8479382489dd35d38e7c19ee28b5 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 13 Nov 2024 17:00:58 +1100 Subject: [PATCH 4/7] Fixed import ordering --- vast_pipeline/management/commands/initpiperun.py | 3 ++- vast_pipeline/management/helpers.py | 2 +- webinterface/settings.py | 3 ++- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/vast_pipeline/management/commands/initpiperun.py b/vast_pipeline/management/commands/initpiperun.py index 7ab0f2fe3..1b4347018 100644 --- a/vast_pipeline/management/commands/initpiperun.py +++ b/vast_pipeline/management/commands/initpiperun.py @@ -4,9 +4,10 @@ Usage: ./manage.py initpiperun pipeline_run_name """ -from Pathlib import Path import logging + from typing import Any, Dict, Optional +from Pathlib import Path from django.core.management.base import BaseCommand, CommandError from django.conf import settings as sett diff --git a/vast_pipeline/management/helpers.py b/vast_pipeline/management/helpers.py index 11d64953a..ffc1c8277 100644 --- a/vast_pipeline/management/helpers.py +++ b/vast_pipeline/management/helpers.py @@ -2,11 +2,11 @@ Helper functions for the commands. """ -from pathlib import Path import logging from typing import Tuple from django.conf import settings as sett +from pathlib import Path logger = logging.getLogger(__name__) diff --git a/webinterface/settings.py b/webinterface/settings.py index 94b496ca6..6a0a466dc 100644 --- a/webinterface/settings.py +++ b/webinterface/settings.py @@ -1,6 +1,7 @@ -from Pathlib import Path import environ +from Pathlib import Path + # Load the Django congig from the .env file env = environ.Env() environ.Env.read_env() From ac10034ced90f703a1698d864d4233c2680853ef Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 13 Nov 2024 17:33:02 +1100 Subject: [PATCH 5/7] Converted vast_pipeline/management/commands/runpipeline.py and also renamed pipeline path variable --- .../management/commands/runpipeline.py | 82 +++++++++---------- 1 file changed, 40 insertions(+), 42 deletions(-) diff --git a/vast_pipeline/management/commands/runpipeline.py b/vast_pipeline/management/commands/runpipeline.py index 42918afec..e3c5ae780 100644 --- a/vast_pipeline/management/commands/runpipeline.py +++ b/vast_pipeline/management/commands/runpipeline.py @@ -4,15 +4,16 @@ Usage: ./manage.py runpipeline pipeline_run_name """ -import os import glob import shutil import logging import traceback import warnings +from Pathlib import Path + from argparse import ArgumentParser -from typing import Optional +from typing import Optional, Union from django.db import transaction from django.contrib.auth.models import User from django.core.management.base import BaseCommand, CommandError @@ -37,7 +38,7 @@ def run_pipe( - name: str, path_name: Optional[str] = None, + name: str, run_path: Optional[Union[str,Path]] = None, run_dj_obj: Optional[Run] = None, cli: bool = True, debug: bool = False, user: Optional[User] = None, full_rerun: bool = False, prev_ui_status: str = 'END' @@ -48,7 +49,7 @@ def run_pipe( Args: name: The name of the pipeline run (p_run.name). - path_name: + run_path: The path of the directory of the pipeline run (p_run.path), defaults to None. run_dj_obj: @@ -75,7 +76,10 @@ def run_pipe( PipelineConfigError: Raised if an error is found in the pipeline config. ''' - path = run_dj_obj.path if run_dj_obj else path_name + path = run_dj_obj.path if run_dj_obj else run_path + if type(path) is not Path: + path = Path(path) + # set up logging for running pipeline from UI if not cli: # set up the logger for the UI job @@ -83,7 +87,7 @@ def run_pipe( if debug: root_logger.setLevel(logging.DEBUG) f_handler = logging.FileHandler( - os.path.join(path, timeStamped('log.txt')), + path/timeStamped('log.txt'), mode='w' ) f_handler.setFormatter(root_logger.handlers[0].formatter) @@ -91,7 +95,7 @@ def run_pipe( pipeline = Pipeline( name=run_dj_obj.name if run_dj_obj else name, - config_path=os.path.join(path, 'config.yaml'), + config_path=path/'config.yaml', validate_config=False, # delay validation ) @@ -100,6 +104,8 @@ def run_pipe( pipeline.name, pipeline.config["run"]["path"], ) + + p_run_path = Path(p_run.path) # backup the last successful outputs. # if the run is being run again and the last status is END then the @@ -108,9 +114,9 @@ def run_pipe( # with the config file below that causes an error. if flag_exist: if cli and p_run.status == 'END': - backup_parquets(p_run.path) + backup_parquets(p_run_path) elif not cli and prev_ui_status == 'END': - backup_parquets(p_run.path) + backup_parquets(p_run_path) # validate run configuration try: @@ -145,17 +151,17 @@ def run_pipe( if not flag_exist: # check for and remove any present .parquet (and .arrow) files parquets = ( - glob.glob(os.path.join(p_run.path, "*.parquet")) + p_run_path.glob("*.parquet") # TODO Remove arrow when arrow files are no longer needed. - + glob.glob(os.path.join(p_run.path, "*.arrow")) - + glob.glob(os.path.join(p_run.path, "*.bak")) + + p_run_path.glob("*.arrow") + + p_run_path.glob("*.bak") ) for parquet in parquets: - os.remove(parquet) + parquet.rm() # copy across config file at the start logger.debug("Copying temp config file.") - create_temp_config_file(p_run.path) + create_temp_config_file(p_run_path) else: # Check if the status is already running or queued. Exit if this is @@ -170,16 +176,14 @@ def run_pipe( # copy across config file at the start logger.debug("Copying temp config file.") - create_temp_config_file(p_run.path) + create_temp_config_file(p_run_path) # Check if there is a previous run config and back up if so - if os.path.isfile( - os.path.join(p_run.path, 'config_prev.yaml') - ): + if (p_run_path / 'config_prev.yaml').is_file(): prev_config_exists = True shutil.copy( - os.path.join(p_run.path, 'config_prev.yaml'), - os.path.join(p_run.path, 'config.yaml.bak') + p_run_path/'config_prev.yaml'), + p_run_path/'config.yaml.bak' ) logger.debug(f'config_prev.yaml exists: {prev_config_exists}') @@ -206,9 +210,8 @@ def run_pipe( if initial_run is False: parquets = ( - glob.glob(os.path.join(p_run.path, "*.parquet")) - # TODO Remove arrow when arrow files are no longer needed. - + glob.glob(os.path.join(p_run.path, "*.arrow")) + p_run_path.glob("*.parquet") + + p_run_path.glob("*.arrow") ) if full_rerun: @@ -220,20 +223,20 @@ def run_pipe( logger.info( 'Cleaning up forced measurements before re-process data' ) - remove_forced_meas(p_run.path) + remove_forced_meas(p_run_path) for parquet in parquets: - os.remove(parquet) + parquet.rm() # remove bak files - bak_files = glob.glob(os.path.join(p_run.path, "*.bak")) + bak_files = p_run_path.glob("*.bak") if bak_files: for bf in bak_files: - os.remove(bf) + bf.rm() # remove previous config if it exists if prev_config_exists: - os.remove(os.path.join(p_run.path, 'config_prev.yaml')) + (p_run_path/'config_prev.yaml').rm() # reset epoch_based flag with transaction.atomic(): @@ -251,7 +254,7 @@ def run_pipe( " that a new or complete re-run should be performed" " instead. Performing no actions. Exiting." ) - os.remove(os.path.join(p_run.path, 'config_temp.yaml')) + (p_run_path/'config_temp.yaml').rm() pipeline.set_status(p_run, 'END') return True @@ -262,7 +265,7 @@ def run_pipe( " previous run. A complete re-run is required if" " changing to epoch based mode or vice versa." ) - os.remove(os.path.join(p_run.path, 'config_temp.yaml')) + (p_run_path/'config_temp.yaml').rm() pipeline.set_status(p_run, 'END') return True @@ -272,8 +275,7 @@ def run_pipe( 'images', 'associations', 'sources', 'relations', 'measurement_pairs' ]: - pipeline.previous_parquets[i] = os.path.join( - p_run.path, f'{i}.parquet.bak') + pipeline.previous_parquets[i] = p_run_path/f'{i}.parquet.bak' except Exception as e: logger.error('Unexpected error occurred in pre-run steps!') pipeline.set_status(p_run, 'ERR') @@ -353,9 +355,9 @@ def run_pipe( # copy across config file now that it is successful logger.debug("Copying and cleaning temp config file.") shutil.copyfile( - os.path.join(p_run.path, 'config_temp.yaml'), - os.path.join(p_run.path, 'config_prev.yaml')) - os.remove(os.path.join(p_run.path, 'config_temp.yaml')) + p_run_path/'config_temp.yaml', + p_run_path/'config_prev.yaml')) + (p_run_path/'config_temp.yaml').rm() # set the pipeline status as completed pipeline.set_status(p_run, 'END') @@ -420,7 +422,7 @@ def handle(self, *args, **options) -> None: # configure logging root_logger = logging.getLogger('') f_handler = logging.FileHandler( - os.path.join(run_folder, timeStamped('log.txt')), + run_folder/timeStamped('log.txt')), mode='w' ) f_handler.setFormatter(root_logger.handlers[0].formatter) @@ -432,18 +434,14 @@ def handle(self, *args, **options) -> None: # set the traceback on options['traceback'] = True - # p_run_name = p_run_path - # remove ending / if present - if p_run_name[-1] == '/': - p_run_name = p_run_name[:-1] # grab only the name from the path - p_run_name = p_run_name.split(os.path.sep)[-1] + p_run_name = p_run_name.name debug_flag = True if options['verbosity'] > 1 else False _ = run_pipe( p_run_name, - path_name=run_folder, + run_path=run_folder, debug=debug_flag, full_rerun=options["full_rerun"], ) From 1e9a6b907d94e38c13731877996ff72c9f901370 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Thu, 14 Nov 2024 14:33:06 +1100 Subject: [PATCH 6/7] Converted clearpiperun --- vast_pipeline/management/commands/clearpiperun.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/vast_pipeline/management/commands/clearpiperun.py b/vast_pipeline/management/commands/clearpiperun.py index 2aa14c2c7..49af57abb 100644 --- a/vast_pipeline/management/commands/clearpiperun.py +++ b/vast_pipeline/management/commands/clearpiperun.py @@ -1,8 +1,8 @@ -import os import logging import shutil from argparse import ArgumentParser +from pathlib import Path from glob import glob from django.core.management.base import BaseCommand, CommandError from django.db import transaction @@ -107,6 +107,7 @@ def handle(self, *args, **options) -> None: except Run.DoesNotExist: raise CommandError(f'Pipeline run {p_run_name} does not exist') + p_run_path = Path(p_run.path) logger.info("Deleting pipeline run '%s' from database", p_run_name) with transaction.atomic(): p_run.status = 'DEL' @@ -118,7 +119,7 @@ def handle(self, *args, **options) -> None: logger.info("Time to delete run from database: %.2f sec", t) # remove forced measurements in db if presents - forced_parquets = remove_forced_meas(p_run.path) + forced_parquets = remove_forced_meas(p_run_path) t = timer.reset() logger.info("Time to delete forced measurements: %.2f sec", t) @@ -126,8 +127,8 @@ def handle(self, *args, **options) -> None: if not options['keep_parquet'] and not options['remove_all']: logger.info('Deleting pipeline "%s" parquets', p_run_name) parquets = ( - glob(os.path.join(p_run.path, '*.parquet')) - + glob(os.path.join(p_run.path, '*.arrow')) + p_run_path.glob('*.parquet') + + p_run_path.glob('*.arrow') ) for parquet in parquets: try: @@ -143,7 +144,7 @@ def handle(self, *args, **options) -> None: if options['remove_all']: logger.info('Deleting pipeline folder') try: - shutil.rmtree(p_run.path) + shutil.rmtree(p_run_path) except Exception as e: self.stdout.write(self.style.WARNING( f'Issues in removing run folder: {e}' From 03855bf6dd8569f12b7492ec360f9d106acf9afc Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Thu, 14 Nov 2024 14:55:01 +1100 Subject: [PATCH 7/7] Converted restorepiperun and fixed helper docs --- .../management/commands/restorepiperun.py | 53 +++++++++---------- vast_pipeline/management/helpers.py | 2 +- 2 files changed, 26 insertions(+), 29 deletions(-) diff --git a/vast_pipeline/management/commands/restorepiperun.py b/vast_pipeline/management/commands/restorepiperun.py index 9cb3a36a2..fefef12a4 100644 --- a/vast_pipeline/management/commands/restorepiperun.py +++ b/vast_pipeline/management/commands/restorepiperun.py @@ -1,10 +1,10 @@ -import os import logging import shutil import numpy as np import pandas as pd from argparse import ArgumentParser +from pathlib import Path from glob import glob from django.db.models import Q from django.db import transaction @@ -52,7 +52,7 @@ def restore_pipe(p_run: Run, bak_files: Dict[str, str], prev_config: PipelineCon Args: p_run (Run): The run model object. - bak_files (Dict[str, str]): + bak_files (Dict[str, Path]): Dictionary containing the paths to the .bak files. prev_config (PipelineConfig): Back up run configuration. @@ -60,13 +60,15 @@ def restore_pipe(p_run: Run, bak_files: Dict[str, str], prev_config: PipelineCon Returns: None """ + p_run_path = Path(p_run.path) + # check images match img_f_list = prev_config["inputs"]["image"] if isinstance(img_f_list, dict): img_f_list = [ item for sublist in img_f_list.values() for item in sublist ] - img_f_list = [os.path.basename(i) for i in img_f_list] + img_f_list = [Path(i).name for i in img_f_list] prev_images = pd.read_parquet( bak_files['images'], columns=['id', 'name', 'measurements_path'] @@ -82,9 +84,7 @@ def restore_pipe(p_run: Run, bak_files: Dict[str, str], prev_config: PipelineCon # check forced measurements monitor = prev_config["source_monitoring"]["monitor"] if monitor: - forced_parquets = glob(os.path.join( - p_run.path, 'forced_*.parquet.bak' - )) + forced_parquets = p_run_path.glob('forced_*.parquet.bak') if not forced_parquets: raise CommandError( @@ -175,9 +175,7 @@ def restore_pipe(p_run: Run, bak_files: Dict[str, str], prev_config: PipelineCon logger.debug('(type, #deleted): %s', detail_del) if monitor: - current_forced_parquets = glob(os.path.join( - p_run.path, 'forced_*.parquet' - )) + current_forced_parquets = p_run_path.glob('forced_*.parquet') current_forced_meas = pd.concat( [pd.read_parquet( @@ -264,20 +262,20 @@ def restore_pipe(p_run: Run, bak_files: Dict[str, str], prev_config: PipelineCon for i in bak_files: bak_file = bak_files[i] if i == 'config': - actual_file = bak_file.replace('.yaml.bak', '_prev.yaml') + actual_file = str(bak_file).replace('.yaml.bak', '_prev.yaml') else: - actual_file = bak_file.replace('.bak', '') + actual_file = str(bak_file).replace('.bak', '') shutil.copy(bak_file, actual_file) - os.remove(bak_file) + bak_file/unlink() if monitor: for i in current_forced_parquets: - os.remove(i) + i.unlink() for i in forced_parquets: - new_file = i.replace('.bak', '') + new_file = str(i).replace('.bak', '') shutil.copy(i, new_file) - os.remove(i) + i.unlink() class Command(BaseCommand): @@ -338,7 +336,7 @@ def handle(self, *args, **options) -> None: # configure logging root_logger = logging.getLogger('') f_handler = logging.FileHandler( - os.path.join(run_folder, timeStamped('restore_log.txt')), + run_folder/timeStamped('restore_log.txt'), mode='w' ) f_handler.setFormatter(root_logger.handlers[0].formatter) @@ -361,28 +359,27 @@ def handle(self, *args, **options) -> None: " Unable to run restore." ) - path = p_run.path + path = Path(p_run.path) pipeline = Pipeline( name=p_run_name, - config_path=os.path.join(path, 'config.yaml') + config_path=path/'config.yaml' ) try: # update pipeline run status to restoring prev_status = p_run.status pipeline.set_status(p_run, 'RES') - prev_config_file = os.path.join(p_run.path, 'config.yaml.bak') + prev_config_file = path/'config.yaml.bak' - if os.path.isfile(prev_config_file): + if prev_config_file.isfile(): + new_prev_config_file = Path(str(prev_config_file).replace('.yaml.bak', '.bak.yaml')) shutil.copy( prev_config_file, - prev_config_file.replace('.yaml.bak', '.bak.yaml') - ) - prev_config_file = prev_config_file.replace( - '.yaml.bak', '.bak.yaml' + new_prev_config_file ) + prev_config_file = new_prev_config_file prev_config = PipelineConfig.from_file(prev_config_file) - os.remove(prev_config_file) + prev_config_file.unlink() else: raise CommandError( 'Previous config file does not exist.' @@ -395,11 +392,11 @@ def handle(self, *args, **options) -> None: 'relations', 'skyregions', 'sources', 'config' ]: if i == 'config': - f_name = os.path.join(p_run.path, f'{i}.yaml.bak') + f_name = path/f'{i}.yaml.bak' else: - f_name = os.path.join(p_run.path, f'{i}.parquet.bak') + f_name = path/f'{i}.parquet.bak' - if os.path.isfile(f_name): + if f_name.isfile(): bak_files[i] = f_name elif ( i != "measurement_pairs" diff --git a/vast_pipeline/management/helpers.py b/vast_pipeline/management/helpers.py index ffc1c8277..ae857a680 100644 --- a/vast_pipeline/management/helpers.py +++ b/vast_pipeline/management/helpers.py @@ -12,7 +12,7 @@ logger = logging.getLogger(__name__) -def get_p_run_name(name: str, return_folder: bool = False) -> Tuple[str, str]: +def get_p_run_name(name: str, return_folder: bool = False) -> Tuple[str, Path]: """ Determines the name of the pipeline run. Can also return the output folder if selected.