# Virus Data Analysis Tool: a data reduction GUI for HETDEX/VIRUS data
# Copyright (C) 2015, 2016 "The HETDEX collaboration"
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Database management for VDAT
"""
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import os
import peewee
from playhouse.shortcuts import model_to_dict
import six
from .base import VDATModels, VDATDBError
import vdat.utilities as vutil
[docs]class MergeTypeError(VDATDBError, TypeError):
'''Raised when merging table entries of different types'''
pass
[docs]class BoolField(peewee.Field):
"""sqlite doesn't play well with boolean, apparently. This field coerce
``True`` and ``False`` into strings ``"true"`` and ``"false"`` and
vice versa
"""
db_field = "text"
[docs] def db_value(self, value):
"""Convert the input bool into a string"""
if value:
return "true"
else:
return "false"
[docs] def python_value(self, value):
"""Convert the input string into a bool"""
if value == "true":
return True
else:
return False
# model for the vdat directories
[docs]class VDATDir(VDATModels):
"""Redux directories metadata"""
version = '1.0'
version_f = peewee.TextField(constraints=[peewee.Check('version_f == ' +
version), ])
# redux directory. Used for updating the path entry in the case the
# redux directory is moved somewhere else
redux_dir = peewee.TextField()
# night
night = peewee.TextField()
# type of the directory
type_ = peewee.TextField()
# name of the directory
name = peewee.TextField()
# full path to name
# TODO: figure out if this field can be built out of others
path = peewee.TextField()
# if True the directory has been created during symlinking, and cannot be
# removed
is_clone = BoolField()
# name of the shot the data are coming from
shot = peewee.TextField()
# time stamp of the directory
timestamp = peewee.DateTimeField()
# original types
original_type_ = peewee.TextField()
# object type
object_ = peewee.TextField()
# default reference zero directory
zero_dir = peewee.ForeignKeyField("self", related_name="target_zero_dir",
null=True)
# default reference calibration directory
cal_dir = peewee.ForeignKeyField("self", related_name="target_cal_dir",
null=True)
class Meta:
# index via path, name and type_ and make that they are unique
indexes = (
(('redux_dir', 'night', 'type_', 'name'), True),
)
order_by = ['night', 'type_', 'name']
[docs] def make_path(self):
"""Fill the ``path`` field joining ``redux_dir``, ``night``, ``type_`` and
``name``
"""
self.path = os.path.join(self.redux_dir, self.night, self.type_,
self.name)
[docs] def merge_entries(self, others, exclude=['id', 'zero_dir', 'cal_dir',
'ifus']):
"""Merge the self with ``other`` excluding ``exclude``.
The merging is done using :func:`vdat.utilities.merge_dicts`
Parameters
----------
other : (list of) :class:`VDATDir` instance
list of models to merge with
exclude : list of strings
list of field names to ignore when merging
Raises
------
MergeTypeError
if the types of ``others`` are not the same as the owner class
"""
if not hasattr(others, '__iter__'):
others = [others, ]
if not all(isinstance(i, type(self)) for i in others):
raise MergeTypeError('Some of the ``others`` are not of the type'
' {}'.format(type(self)))
new_dir = vutil.merge_dicts([self.data, ] + [o.data for o in others],
exclude=exclude)
for k, v in six.iteritems(new_dir):
setattr(self, k, v)
@property
def data_clean(self):
"""Expose the internal data of the model without the id and without
recursing the foreign keywords
Returns
-------
dict
dictionary of data
"""
return model_to_dict(self, recurse=False,
exclude=[getattr(self.__class__, i) for i in
['id', 'zero_dir', 'cal_dir']])
# model for the vdat directories
[docs]class VDATExposures(VDATModels):
"""Redux directories metadata"""
version = '1.0'
version_f = peewee.TextField(constraints=[peewee.Check('version_f == ' +
version), ])
# Path of the redux subdirectory (full path to name)
path = peewee.TextField()
# Name of the redux subdirectory
name = peewee.TextField()
# basename of one exposures, i.e. the 20151025T120000 part
basename = peewee.TextField()
# name of the original exposure directory
expname = peewee.TextField()
# exposure type
exptype = peewee.TextField()
# original types
original_type = peewee.TextField()
# object type
object_ = peewee.TextField()
