Source code for shapepipe.utilities.canfar


This module defines methods for managing CANFAR specific actions.

:Author: Samuel Farrens <>
         Martin Kilbinger <>


import os
import sys
from contextlib import redirect_stdout
from io import StringIO

    import vos.commands as vosc
except ImportError:  # pragma: no cover
    import_fail = True
    import_fail = False

[docs]class vosError(Exception): """VOS Error. Generic error that is raised by the vosHandler. """ pass
[docs]class vosHandler: """VOS Handler. This class manages the use of VOS commands. Parameters ---------- command : str VOS command name """ def __init__(self, command): self._check_vos_install() self._avail_commands = tuple(vosc.__all__) self.command = command
[docs] @staticmethod def _check_vos_install(): """Check VOS Install. Check if VOS is correctly installed. Raises ------ ImportError if vos package cannot be imported """ if import_fail: raise ImportError( 'vos package not found, re-install ShapePipe ' + 'with \'./install_shapepipe --vos\'' )
@property def command(self): """Set Command. This method sets the VOS command property. Raises ------ ValueError if value is not valid vos command """ return self._command @command.setter def command(self, value): if value not in self._avail_commands: raise ValueError( f'vos command must be one of {self._avail_commands}' ) self._command = getattr(vosc, value)
[docs] def __call__(self, *args, **kwargs): """Call Method. This method allows class instances to be called as functions. Raises ------ vosError if error in vos command occurs """ try: self._command() except Exception: raise vosError( f'Error in VOs command: {self._command.__name__}' )
[docs]def download(source, target, verbose=False): """Download. Download file from vos. Parameters ---------- source : str source path on vos target : str target path verbose : bool, optional, default=False verbose output if True Returns ------- status : bool status, True/False or success/failure """ cmd = 'vcp' if not os.path.exists(target): sys.argv = [cmd, source, target] if verbose: print(f'Downloading file {source} to {target}...') vcp = vosHandler(cmd) vcp() if verbose: print('Download finished.') else: if verbose: print(f'Target file {target} exists, skipping download.')
[docs]def dir_list(path, verbose=False): """Set Directory List. List content of path on vos. Parameters ---------- path : str path on vos, starts with 'vos:cfis/...' verbose : bool, optional, default=False verbose output if True Raises ------ HTTPError, KeyError if error occurs during vos command Returns ------- vls_out : array of str file or directory at path """ cmd = 'vls' sys.argv = [cmd, path] vls = vosHandler(cmd) if verbose: print('Getting vos directory content from vls...') f = StringIO() try: with redirect_stdout(f): vls() except Exception: print('Error during vls command') raise vls_out = f.getvalue() return vls_out.split('\n')