Source code for recipe_system.cal_service.localdb

# Defines the LocalDB class for calibration returns. This is a high-level
# interface to the local calibration manager, and provides an API for
# modifying the database on disk.

from os import path, makedirs

from .caldb import CalDB, CalReturn
from .calrequestlib import get_cal_requests, generate_md5_digest

    from . import localmanager
    localmanager_available = True
except ImportError as e:
    localmanager_available = False
    import_error = str(e)

DEFAULT_DB_NAME = "cal_manager.db"

[docs] class LocalDB(CalDB): """ The class handling a calibration database stored on disk, via the LocalManager class. In addition to the methods required to interface with DRAGONS data reduction pipelines, other methods are used to provide a full API and effect actions of the "caldb" script. An attempt to create an instance of this class without the LocalManager being importable will result in an error. Attributes ---------- dbfile : str name of the file on disk holding the database _calmgr : LocalManager instance the local calibration manager that will handle the requests """ def __init__(self, dbfile, name=None, valid_caltypes=None, procmode=None, get_cal=True, store_cal=True, log=None, force_init=False): if not localmanager_available: raise ValueError(f"Cannot initialize local database {name} as" "localmanager could not be imported.\n" f"{import_error}") if name is None: # Do this first so "~" is in the name name = dbfile dbfile = path.expanduser(dbfile) if path.isdir(dbfile) or dbfile.endswith(path.sep): dbfile = path.join(dbfile, DEFAULT_DB_NAME) name = path.join(name, DEFAULT_DB_NAME) super().__init__(name=name, get_cal=get_cal, store_cal=store_cal, log=log, valid_caltypes=valid_caltypes, procmode=procmode) self.dbfile = dbfile self._calmgr = localmanager.LocalManager(dbfile) if not path.exists(dbfile) and force_init: self.log.stdinfo(f"Local database file {dbfile} does not exist. " "Initializing.") if not path.exists(path.dirname(dbfile)): makedirs(path.dirname(dbfile)) self.init() def _get_calibrations(self, adinputs, caltype=None, procmode=None, howmany=1): self.log.debug(f"Querying {} for {caltype}") cal_requests = get_cal_requests(adinputs, caltype, procmode=procmode, is_local=True) cals = [] for rq in cal_requests: local_cals = self._calmgr.calibration_search(rq, howmany=howmany) if not local_cals[0]: cals.append(None) continue good_cals = [] for calurl, calmd5 in zip(*local_cals): calfile = calurl[7:] # strip "file://" cached_md5 = generate_md5_digest(calfile) if calmd5 == cached_md5: self.log.debug(f"{rq.filename}: retrieved {calfile}") good_cals.append(calfile) else: self.log.warning(f"md5 checksum of {calfile} does not match." " Not returning this calibration") # Append list if >1 requested, else just the filename string if good_cals: cals.append(good_cals if howmany != 1 else good_cals[0]) else: cals.append(None) return CalReturn([None if cal is None else (cal, for cal in cals]) def _store_calibration(self, cal, caltype=None): """Store the calibration. The LocalDB is not interested in science""" if self.store_cal: if caltype is None or "science" not in caltype: if not path.exists(cal): raise OSError(f"File {cal} does not exist.") if caltype is not None: self.log.stdinfo(f"{}: Storing {cal} as {caltype}") self._calmgr.ingest_file(cal) else: self.log.stdinfo(f"{}: NOT storing {cal} as {caltype}") # The following methods provide an API to modify the database, by # initializing it, removing a named calibration, and listing the files # it contains.
[docs] def init(self, wipe=False): """ Initialize a calibration database. Callers will usually only want to do this once. But if called again, it will wipe the old database. Parameters ---------- wipe : <bool>, optional If the database exists and this parameter is `True`, the file will be removed and recreated before initializing Raises ------ IOError If the file exists and there a system error when trying to remove it (eg. lack of permissions). LocalManagerError If the file exists and `wipe` was `False` """ return self._calmgr.init_database(wipe=wipe)
[docs] def add_cal(self, calfile): self._store_calibration(calfile)
[docs] def add_directory(self, path, walk=False): """ Ingest one or more files from a given directory, optionally searching all subdirectories. This is not used by primitives in the DRAGONS data reduction pipelines. Parameters ---------- path : str directory containing files to be ingested walk : bool add files from all subdirectories? """ self._calmgr.ingest_directory(path, walk=walk, log=None)
[docs] def remove_cal(self, calfile): """ Removes a calibration file from the database. Note that only the filename is relevant. All duplicate copies in the database will be removed. Parameters ---------- calfile : <str> Path to the file. It can be either absolute or relative """ return self._calmgr.remove_file(path.basename(calfile))
[docs] def list_files(self): """ List all files in the local calibration database. This is not used by primitives in the DRAGONS data reduction pipelines. Returns ------- LocalManager.list_files: <generator>. (See class docstring for example of how to use this generator.) Raises ------ LocalManagerError Raised when unable to read database. """ return self._calmgr.list_files()