00001
00002
00003
00004
00005
00006
00007
00008
00009 """
00010 :py:class:`PSCalib.DCBase` - base class for the Detector Calibration (DC) project.
00011
00012 Usage::
00013
00014 # Import
00015 from PSCalib.DCBase import DCBase
00016
00017 o = DCBase()
00018
00019 # Dictionary of parameters
00020 # ========================
00021
00022 o.set_pars_dict(d) # set (dict) dictionary of pars.
00023 o.add_par(k,v) # add (k,v) par to the dictionary of pars.
00024 o.del_par(k) # delete par with key k.
00025 o.clear_pars() # delete all pars from the dictionary.
00026 d = o.pars_dict() # returns (dict) dictionary of pars.
00027 p = o.par(k) # returns par value for key k.
00028 t = o.pars_text() # returns (str) text of all pars.
00029
00030 # History records
00031 # ===============
00032
00033 o.set_history_dict(d) # set (dict) dictionary of history from specified dictionary
00034 o.add_history_record(rec, tsec=None) # add (str) record with (int) time[sec] to the history dictionary of (tsec:rec).
00035 # If tsec is None - current time is used as a key.
00036 o.del_history_record(tsec) # Delete one history record from the dictionary by its time tsec.
00037 o.clear_history() # Delete all history records from the dictionary.
00038 d = o.history_dict() # returns (dict) history dictionary associated with current object .
00039 r = o.history_record(tsec) # returns (str) history record for specified time tsec.
00040 t = o.history_text(tsfmt=None) # returns (str) all history records preceded by the time stamp as a text.
00041
00042 # Save and Load
00043 # =============
00044
00045 o.save_history_file(path='history.txt', verb=False) # save history in the text file
00046 o.load_history_file(path='history.txt', verb=False) # load history from the text file
00047
00048 o.save_base(grp) # save everything in hdf5 group
00049 o.load_base(name, grp) # load from hdf5 group
00050
00051 # Time convertors
00052 # ===============
00053
00054 t_str = o.tsec_to_tstr(tsec, tsfmt=None) # converts (float) time[sec] to the (str) time stamp
00055 t_sec = o.tstr_to_tsec(tstr, tsfmt=None) # converts (str) time stamp to (float) time[sec]
00056
00057 @see project modules
00058 * :py:class:`PSCalib.DCStore`
00059 * :py:class:`PSCalib.DCType`
00060 * :py:class:`PSCalib.DCRange`
00061 * :py:class:`PSCalib.DCVersion`
00062 * :py:class:`PSCalib.DCBase`
00063 * :py:class:`PSCalib.DCInterface`
00064 * :py:class:`PSCalib.DCUtils`
00065 * :py:class:`PSCalib.DCDetectorId`
00066 * :py:class:`PSCalib.DCConfigParameters`
00067 * :py:class:`PSCalib.DCFileName`
00068 * :py:class:`PSCalib.DCLogger`
00069 * :py:class:`PSCalib.DCMethods`
00070 * :py:class:`PSCalib.DCEmail`
00071
00072 This software was developed for the SIT project.
00073 If you use all or part of it, please give an appropriate acknowledgment.
00074
00075 @version $Id: DCBase.py 12879 2016-11-17 22:15:26Z dubrovin@SLAC.STANFORD.EDU $
00076
00077 @author Mikhail S. Dubrovin
00078 """
00079
00080
00081 __version__ = "$Revision: 12879 $"
00082
00083
00084 from time import time, sleep, localtime, gmtime, strftime, strptime, mktime
00085 from math import floor
00086 from PSCalib.DCLogger import log
00087 from PSCalib.DCUtils import get_subgroup, save_object_as_dset
00088
00089
00090
00091
00092 class DCBase(object) :
00093 """Base class for the Detector Calibration (DC) project.
00094
00095 Parameters
00096
00097 cmt : str - string of comment associated with derived class object.
00098 """
00099 _tsfmt = '%Y-%m-%dT%H:%M:%S'
00100 _offspace = ' '
00101
00102
00103 def __init__(self, cmt=None) :
00104 self._name = 'DCBase'
00105 self._dicpars = {}
00106 self._dichist = {}
00107 msg = 'In c-tor %s' % self._name
00108 log.debug(msg, self._name)
00109 self._grp_pars_name = '_parameters'
00110 self._grp_history_name = '_history'
00111 self._tsec_old = None
00112 self._lst_del_keys = []
00113
00114 if cmt is not None : self.add_history_record(cmt)
00115
00116
00117 def __del__(self) :
00118 self._dicpars.clear()
00119
00120
00121 def set_pars_dict(self, d) :
00122 self._dicpars.clear()
00123 for k,v in d.items() :
00124 self._dicpars[k] = v
00125
00126
00127 def add_par(self, k, v) :
00128 self._dicpars[k] = v
00129
00130
00131 def del_par(self, k) :
00132 if k in self._dicpars.keys() : del self._dicpars[k]
00133
00134
00135 def clear_pars(self) :
00136 self._dicpars.clear()
00137
00138
00139 def pars_dict(self) :
00140 return self._dicpars if len(self._dicpars)>0 else None
00141
00142
00143 def par(self, k ) :
00144 return self._dicpars.get(k, None)
00145
00146
00147 def pars_text(self) :
00148 return ', '.join(['(%s : %s)' % (str(k), str(v)) for k,v in self._dicpars.items()])
00149
00150
00151 def set_history_dict(self, d) :
00152 self._dichist.clear()
00153 for k,v in d.items() :
00154 self._dichist[k] = v
00155
00156
00157 def add_history_record(self, rec, tsec=None) :
00158 t_sec = tsec
00159 if t_sec is None :
00160 t_sec = time()
00161 if t_sec == self._tsec_old :
00162 t_sec += 0.001
00163 self._tsec_old = t_sec
00164 self._dichist[t_sec] = rec
00165
00166
00167
00168
00169
00170
00171 def del_history_record(self, k) :
00172 if k in self._dichist.keys() : del self._dichist[k]
00173
00174
00175 def clear_history(self) :
00176 self._dichist.clear()
00177
00178
00179 def history_dict(self) :
00180 return self._dichist
00181
00182
00183 def history_record(self, tsec) :
00184 return self._dichist.get(tsec)
00185
00186
00187 def history_text(self, tsfmt=None) :
00188 """Returns (str) history records preceded by the time stamp as a text"""
00189 fmt = self._tsfmt if tsfmt is None else tsfmt
00190 return '\n'.join(['%s %s' % (self.tsec_to_tstr(ts), str(rec)) for ts,rec in sorted(self._dichist.items())])
00191
00192
00193 def save_history_file(self, path='history.txt', verb=False) :
00194 """Save history in the text file"""
00195 f = open(path,'w')
00196 f.write(self.history_text())
00197 f.close()
00198 if verb :
00199
00200 log.debug('History records are saved in the file: %s' % path, self._name)
00201
00202
00203 def load_history_file(self, path='history.txt', verb=False) :
00204 """Load history from the text file"""
00205 f = open(path,'r')
00206 lines = f.readlines()
00207 f.close()
00208 for line in lines :
00209 tsstr, rec = line.rstrip('\n').split(' ',1)
00210
00211 self._dichist[self.tstr_to_tsec(tsstr)] = rec
00212 if verb :
00213
00214 log.debug('Read history records from the file: %s' % path, self._name)
00215
00216
00217 def _save_pars_dict(self, grp) :
00218 """Saves _dicpars in the h5py group"""
00219 if not self._dicpars : return
00220
00221
00222 grpdic = get_subgroup(grp, self._grp_pars_name)
00223 for k,v in self._dicpars.items() :
00224 ds = save_object_as_dset(grpdic, name=k, data=v)
00225
00226
00227 def _save_hystory_dict(self, grp) :
00228 """Saves _dichist in the h5py group"""
00229 if not self._dichist : return
00230
00231
00232 grpdic = get_subgroup(grp, self._grp_history_name)
00233 for k,v in self._dichist.items() :
00234
00235 tstamp = str('%.6f' % k)
00236
00237 ds = save_object_as_dset(grpdic, tstamp, data=v)
00238
00239
00240
00241 def save_base(self, grp) :
00242 self._save_pars_dict(grp)
00243 self._save_hystory_dict(grp)
00244
00245
00246 def group_name(self, grp) :
00247 return grp.name.rsplit('/')[-1]
00248
00249
00250 def is_base_group(self, name, grp) :
00251 return self.load_base(name, grp)
00252
00253
00254 def load_base(self, name, grp) :
00255 grpname = name
00256
00257 if grpname == self._grp_pars_name :
00258 self._load_pars_dict(grp)
00259 return True
00260 elif grpname == self._grp_history_name :
00261 self._load_hystory_dict(grp)
00262 return True
00263 return False
00264
00265
00266 def _load_pars_dict(self, grp) :
00267 log.debug('_load_pars_dict for group %s' % grp.name, self._name)
00268 self.clear_pars()
00269 for k,v in dict(grp).iteritems() :
00270 log.debug('par: %s = %s' % (k, str(v[0])), self._name)
00271 self.add_par(k, v[0])
00272
00273
00274 def _load_hystory_dict(self, grp) :
00275 log.debug('_load_hystory_dict for group %s' % grp.name, self._name)
00276 self.clear_history()
00277 for k,v in dict(grp).iteritems() :
00278 tsec = float(k)
00279 log.debug('t: %.6f rec: %s' % (tsec, v[0]), self._name)
00280 self.add_history_record(v[0], tsec)
00281
00282
00283 def tsec_to_tstr(self, tsec, tsfmt=None, addfsec=True) :
00284 """converts float tsec like 1471035078.908067 to the string 2016-08-12T13:51:18.908067"""
00285 fmt = self._tsfmt if tsfmt is None else tsfmt
00286 itsec = floor(tsec)
00287 strfsec = ('%.6f' % (tsec-itsec)).lstrip('0') if addfsec else ''
00288 return '%s%s' % (strftime(fmt, localtime(itsec)), strfsec)
00289
00290
00291 def tstr_to_tsec(self, tstr, tsfmt=None) :
00292 """converts string tstr like 2016-08-12T13:51:18.908067 to the float time in seconds 1471035078.908067"""
00293
00294 fmt = self._tsfmt if tsfmt is None else tsfmt
00295 ts, fsec = tstr.split('.')
00296 return mktime(strptime(ts, fmt)) + 1e-6*int(fsec)
00297
00298
00299
00300
00301 def print_base(self, offset=' ') :
00302 """Print content of dictionaries of parameters and history"""
00303 print '%s %s' % (offset, self._name)
00304
00305 if len(self._dicpars) : print '%s Parameters:' % offset
00306 for k,v in self._dicpars.items() :
00307 print ' %s par: %20s value: %s' % (offset, k, str(v))
00308
00309 if len(self._dichist) : print '%s History:' % offset
00310 for k,v in sorted(self._dichist.items()) :
00311
00312 print ' %s %s %s' % (offset, self.tsec_to_tstr(k, addfsec=False), str(v))
00313
00314
00315
00316 def make_record(self, action='', key='', cmt=False) :
00317 """Returns string record combined with comment.
00318
00319 Parameters
00320
00321 action : str - description of method action,
00322 key : str - key for hdf5 group or dataset name,
00323 cmt : str/None/False - additional comment or no-comment: False is used to turn off history record, None - no-comment.
00324 """
00325 if cmt is None or cmt is False : return '%s %s' % (action, key)
00326 return '%s %s: %s' % (action, key, cmt)
00327
00328
00329
00330 def test_pars() :
00331 o = DCBase()
00332 d = {1:'10', 2:'20', 3:'30'}
00333 o.set_pars_dict(d)
00334 print '\nTest pars: %s' % o.pars_text()
00335 o.del_par(2)
00336 print '\nAfter del_par(2): %s' % o.pars_text()
00337 print '\npar(3): %s' % o.par(3)
00338
00339
00340
00341 def test_history() :
00342 o = DCBase()
00343 o.add_history_record('rec 01')
00344 o.add_history_record('rec 02')
00345 o.add_history_record('rec 03')
00346 o.add_history_record('rec 04')
00347 o.add_history_record('rec 05')
00348 o.add_history_record('rec 06')
00349 print '\nTest history records:\n%s' % o.history_text()
00350 o.save_history_file('history-test.txt', verb=True)
00351 o.add_history_record('rec 07')
00352
00353 o.load_history_file('history-test.txt', verb=True)
00354 print '\nTest history records:\n%s' % o.history_text()
00355
00356
00357
00358 def test_time_converters() :
00359 o = DCBase()
00360 t_sec = time()
00361 t_str = o.tsec_to_tstr(t_sec, tsfmt=None)
00362 t_sec2 = o.tstr_to_tsec(t_str, tsfmt=None)
00363 print 'convert time %.6f to time stamp: %s' % (t_sec, t_str)
00364 print 'and back to time %.6f' % (t_sec2)
00365
00366
00367
00368 def test_make_record() :
00369 o = DCBase()
00370 print o.make_record(action='test make_record for cmt=False', key='keyword', cmt=False)
00371 print o.make_record(action='test make_record for cmt=None', key='keyword', cmt=None)
00372 print o.make_record(action='test make_record for cmt="my comment"', key='keyword', cmt="my comment")
00373
00374
00375
00376 if __name__ == "__main__" :
00377 import sys
00378 test_pars()
00379 test_history()
00380 test_time_converters()
00381 test_make_record()
00382 sys.exit('End of %s test.' % sys.argv[0])
00383
00384
