calibcant/util.py

   1 # Copyright (C) 2012-2013 W. Trevor King <wking@tremily.us>
   2 #
   3 # This file is part of calibcant.
   4 #
   5 # calibcant is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free Software
   7 # Foundation, either version 3 of the License, or (at your option) any later
   8 # version.
   9 #
  10 # calibcant is distributed in the hope that it will be useful, but WITHOUT ANY
  11 # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
  12 # A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License along with
  15 # calibcant.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 """Useful utilites not related to calibration.
  18
  19 Currently just a framework for consistently saving/loading calibration
  20 data.
  21 """
  22
  23 import h5py as _h5py
  24
  25 from h5config.storage.hdf5 import HDF5_Storage as _HDF5_Storage
  26 from h5config.storage.hdf5 import h5_create_group as _h5_create_group
  27
  28
  29 from . import LOG as _LOG
  30
  31
  32 class SaveSpec (object):
  33     def __init__(self, item=None, relpath='/', key=None, config=False,
  34                  array=False, units=None, deviation=None):
  35         self.item = item
  36         self.relpath = relpath
  37         self.key = key
  38         self.config = config
  39         self.array = array
  40         self.units = units
  41         self.deviation = deviation
  42
  43 def save(filename=None, group='/', specs=tuple()):
  44     f = None
  45     storage = _HDF5_Storage()
  46     try:
  47         if isinstance(group, str):
  48             f = _h5py.File(filename, 'a')
  49             group = _h5_create_group(f, group)
  50         for spec in specs:
  51             if spec.item is None:
  52                 continue
  53             cwg = _h5_create_group(group, spec.relpath)
  54             if spec.config:
  55                 storage.save(config=spec.item, group=cwg)
  56                 continue
  57             assert spec.units, spec.item
  58             for k in ['data', 'units', 'standard-deviation']:
  59                 try:
  60                     del cwg[k]
  61                 except KeyError:
  62                     pass
  63             cwg['data'] = spec.item
  64             cwg['units'] = spec.units
  65             if spec.deviation is not None:
  66                 cwg['standard-deviation'] = spec.deviation
  67     finally:
  68         if f:
  69             f.close()
  70
  71 def load(filename=None, group='/', specs=tuple()):
  72     data = {}
  73     f = None
  74     storage = _HDF5_Storage()
  75     try:
  76         if isinstance(group, str):
  77             f = _h5py.File(filename, 'a')
  78             group = _h5_create_group(f, group)
  79         for spec in specs:
  80             try:
  81                 cwg = group[spec.relpath]
  82             except KeyError:
  83                 continue
  84             d = data
  85             for n in spec.key[:-1]:
  86                 if n not in d:
  87                     d[n] = {}
  88                 d = d[n]
  89             if spec.config:
  90                 d[spec.key[-1]] = spec.config(storage=_HDF5_Storage(group=cwg))
  91                 d[spec.key[-1]].load()
  92                 continue
  93             assert spec.units, spec.key
  94             try:
  95                 if spec.array:
  96                     d[spec.key[-1]] = cwg['data'][...]
  97                 else:
  98                     d[spec.key[-1]] = float(cwg['data'][...])
  99             except KeyError:
 100                 continue
 101             except TypeError, e:
 102                 _LOG.warn('while loading {} from {}: {}'.format(
 103                         spec.key, cwg['data'], e))
 104                 raise
 105             if spec.key[-1] in d:
 106                 units_ = cwg['units'][...]
 107                 assert units_ == spec.units, (units_, spec.units)
 108                 if spec.deviation is not None:
 109                     try:
 110                         d[spec.deviation] = float(
 111                             cwg['standard-deviation'][...])
 112                     except KeyError:
 113                         pass
 114     finally:
 115         if f:
 116             f.close()
 117     return data