Merge remote-tracking branch 'upstream/develop' into develop

Author: martinkilbinger

cs_util/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-
 """cs_util PACKAGE.
 
 Provide a basic description of what your package contains.

cs_util/canfar.py

@@ -0,0 +1,182 @@
+"""CANFAR TOOLS.
+
+This module defines methods for managing CANFAR specific actions.
+
+:Author: Samuel Farrens <samuel.farrens@cea.fr>
+         Martin Kilbinger <martin.kilbinger@cea.fr>
+
+"""
+
+import os
+import sys
+from contextlib import redirect_stdout
+from io import StringIO
+
+import vos.commands as vosc
+
+
+class vosError(Exception):
+    """VOS Error.
+
+    Generic error that is raised by the vosHandler.
+
+    """
+
+    pass
+
+
+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
+
+    @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')
+
+    @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)
+
+    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__}'
+            )
+
+
+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.')
+
+
+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
+    -------
+    list
+        file or directory at path, type is str
+
+    """
+    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')

cs_util/cat.py

@@ -4,7 +4,7 @@
 
 :Description: This script contains methods to read and write galaxy catalogues.
 
-:Author: Martin Kilbinger
+:Author: Martin Kilbinger <martin.kilbinger@cea.fr>
 
 """
 

cs_util/cfis.py

@@ -0,0 +1,112 @@
+"""CFIS.
+
+:Name: cfis.py
+
+:Description: This script contains CFIS-specific methods.
+
+:Author: Martin Kilbinger <martin.kilbinger@cea.fr>
+
+"""
+
+
+import re
+import numpy as np
+from astropy import units
+from astropy import coordinates as coords
+
+
+class Cfis(object):
+    """Cfis
+
+    Class for CFIS image properties.
+
+    """
+    size = {'tile' : 0.5 * units.deg}
+
+
+def get_tile_number(tile_name):
+    """Get Tile Number.
+
+    Return tile number of given image tile name.
+
+    Parameters
+    ----------
+    tile_name : str
+        tile name
+
+    Raises
+    ------
+    ValueError
+        if tile name does not match expected pipeline numbering scheme
+
+    Returns
+    -------
+    tuple
+        tile number for x and tile number for y
+
+    """
+    m = re.search(r'(\d{3})[\.-](\d{3})', tile_name)
+    if m is None or len(m.groups()) != 2:
+        raise ValueError(
+            f'Image name \'{tile_name}\' does not match tile name syntax'
+        )
+
+    nix = m.groups()[0]
+    niy = m.groups()[1]
+
+    return nix, niy
+
+
+def get_tile_coord_from_nixy(nix, niy):
+    """Get Tile Coord From Nixy.
+
+    Return coordinates corresponding to tile with number (nix,niy).
+
+    Parameters
+    ----------
+    nix : str or list
+        tile number(s) for x coordinate(s);
+    niy : str or list
+        tile number(s) for y coordinate(s)
+
+    See also
+    --------
+    get_tile_number_from_coord
+
+    Returns
+    -------
+    tuple
+        right ascension and declination
+
+    """
+    number_pattern = re.compile(r'\d{3}')
+
+    # Transform from str to int
+    if not np.isscalar(nix):
+        # Check input numbers
+        if not all(
+            [bool(number_pattern.fullmatch(number)) for number in nix + niy]
+        ):
+            raise ValueError('Input needs to be three-digit numbers')
+
+        # Transform to list
+        xi = np.array(nix).astype(int)
+        yi = np.array(niy).astype(int)
+    else:
+        if not all(
+            [bool(number_pattern.fullmatch(number)) for number in [nix, niy]]
+        ):
+            raise ValueError('Input needs to be three-digit numbers')
+
+        xi = int(nix)
+        yi = int(niy)
+
+    # Declination
+    d = yi / 2 - 90
+    dec = coords.Angle(d, unit='deg')
+
+    # Right ascension
+    r = xi / 2 / np.cos(dec.radian)
+    ra = coords.Angle(r, unit='deg')
+
+    return ra, dec