From 07680d008799f3be47315ffe8fd67be692b64fbf Mon Sep 17 00:00:00 2001 From: "W. Trevor King" Date: Thu, 4 Nov 2010 09:58:37 -0400 Subject: [PATCH] Updated data_logger.py to more PEP-8-compatible code & deprecate in favor of h5py. --- README | 17 +- data_logger.py | 711 +++++++++++++++++++++++++++++++++---------------- 2 files changed, 494 insertions(+), 234 deletions(-) diff --git a/README b/README index daf8b85..389313e 100644 --- a/README +++ b/README @@ -9,6 +9,11 @@ General data is pickled using `cPickle`, and there are specialized methods for saving data in the form of binary strings or dicts of numpy arrays. +`data_logger` is basically a really wimpy version of h5py_. USE h5py +INSTEAD! + +.. _h5py: http://code.google.com/p/h5py/ + Installation ============ @@ -31,16 +36,8 @@ On Debian-esque systems, the installation procedure should be:: Gentoo ------ -You should probably use the `wtk overlay`_:: - - # emerge -av app-portage/layman dev-vcs/git - # layman --add wtk - # emerge -av sci-misc/data-logger - -.. _wtk overlay: http://www.physics.drexel.edu/~wking/unfolding-disasters/posts/Gentoo_overlay/ - -You could also follow the `pip` route, although `pip` isn't in the -Portage tree at the moment (`Gentoo bug 317791`_). Either:: +`pip` isn't in the Portage tree at the moment (`Gentoo bug 317791`_), +so either install it with `easy_install`:: # emerge -av dev-python/setuptools # easy_install pip diff --git a/data_logger.py b/data_logger.py index e65a869..293b4ef 100644 --- a/data_logger.py +++ b/data_logger.py @@ -1,7 +1,7 @@ #!/user/bin/python # # data_logger - classes for consistently logging data in an organized -# fasion. See the test functions for some usage examples +# fasion. See the doctests for some usage examples. # # Copyright (C) 2008-2010 William Trevor King # @@ -24,316 +24,579 @@ # write to Trevor King, Drexel University, Physics Dept., 3141 Chestnut St., # Philadelphia PA 19104, USA. +from __future__ import with_statement + import os, os.path import stat import cPickle as pickle import time import string + import numpy __version__ = "0.3" DEFAULT_PATH = "~/rsrch/data" -DEFAULT_PATH_REPLACE_STRING = "$DEFAULT$/" +DEFAULT_PATH_REPLACE_STRING = "${DEFAULT}/" -class Error (Exception) : +class Error (Exception): "Basic module error class" pass -class ErrorDirExists (Error) : +class ErrorDirExists (Error): "The specified directory already exists" + pass -def normalize_logdir(log_dir): - length = len(DEFAULT_PATH_REPLACE_STRING) - if log_dir[:length] == DEFAULT_PATH_REPLACE_STRING: - log_dir = os.path.join(DEFAULT_PATH, log_dir[length:]) - log_dir = os.path.expanduser(log_dir) - return log_dir +def normalize_log_dir(log_dir): + """Normalize a log directory. -class data_log : - """ - Data logging class. - Creates consistent, timestamped log files. + Expands the user symbol `~`, as well as + `DEFAULT_PATH_REPLACE_STRING`. + + Parameters + ---------- + log_dir : path + Raw `log_dir` passed into `.__init__()`. + + Returns + ------- + log_dir : path + Normalized version of the input `log_dir`. - Initialized with log_dir and log_name. - log_dir specifies the base data directory. - If it doesn't exist, log_dir is created. - - If log_dir begins with '$DEFAULT$/', that portion of the path is replaced - with the then-current contents of the DEFAULT_PATH module global. + Examples + -------- + >>> normalize_log_dir('~/.log') # doctest: +ELLIPSIS + '/.../.log' + >>> normalize_log_dir('${DEFAULT}/hi/there') # doctest: +ELLIPSIS + '/.../rsrch/data/hi/there' + """ + if log_dir.startswith(DEFAULT_PATH_REPLACE_STRING): + length = len(DEFAULT_PATH_REPLACE_STRING) + log_dir = os.path.join(DEFAULT_PATH, log_dir[length:]) + log_dir = os.path.expanduser(log_dir) + return log_dir - A subdir of log_dir is created (if necessary) named YYYYMMDD, - where YYYYMMDD is the current day in localtime. - If noclobber_logsubdir == True, this dir must not exist yet. - log_name specifies the base name for the created log files (in the log subdir). - The created log filenames are prefixed with a YYYYMMDDHHMMSS timestamp. - If the target filename already exists, the filename is postfixed with - '_N', where N is the lowest integer that doesn't clobber an existing file. +class DataLog (object): + """Create consistent, timestamped log files. - General data is saved to the log files with the write(obj) method. - By default, write() cPickles the object passed. - You can save in other formats by overriding write() + General data is saved to the log files with the `write(obj)` + method. By default, `write()` `cPickles` the object passed. You + can save in other formats by overriding `write()`. Binary data is can be saved directly to the log files with the - write_binary(binary_string) method. + `write_binary(binary_string)` method. All file names are stripped of possibly troublesome characters. + + Parameters + ---------- + log_dir : path + `log_dir` sets the base data directory. If it doesn't exist, + `log_dir` is created. + + If log_dir begins with '$DEFAULT$/', that portion of the path + is replaced with the then-current contents of the + `DEFAULT_PATH` module global. + + A subdir of log_dir is created (if necessary) named + `YYYYMMDD`, where `YYYYMMDD` is the current day in local time. + log_name : string + `log_name` specifies the base name for the created log files + (in the log subdir). The created log filenames are prefixed + with a `YYYYMMDDHHMMSS` timestamp. If the target filename + already exists, the filename is postfixed with `_N`, where + `N` is the lowest integer that doesn't clobber an existing + file. + noclobber_log_subdir : bool + `noclobber_log_subdir == True`, the `YYYMMDD` subdir of + `log_dir` must not exist yet. + timestamp : string + Overide default subdir `timestamp` (%Y%m%d). + + Examples + -------- + + >>> import shutil + >>> dl = DataLog('test_data_log', 'temperature', timestamp='20101103', + ... ) + >>> data = {'test':True, 'data':[1, 2, 3, 4]} + >>> files = [None]*10 + >>> for i in range(10): + ... files[i],ts = dl.write(data, timestamp='20101103235959') + >>> print '\\n'.join(files) + test_data_log/20101103/20101103235959_log + test_data_log/20101103/20101103235959_log_1 + test_data_log/20101103/20101103235959_log_2 + test_data_log/20101103/20101103235959_log_3 + test_data_log/20101103/20101103235959_log_4 + test_data_log/20101103/20101103235959_log_5 + test_data_log/20101103/20101103235959_log_6 + test_data_log/20101103/20101103235959_log_7 + test_data_log/20101103/20101103235959_log_8 + test_data_log/20101103/20101103235959_log_9 + >>> shutil.rmtree(dl._log_dir) """ - def __init__(self, log_dir=".", noclobber_logsubdir=False, - log_name="log", - timestamp=None) : + def __init__(self, log_dir=".", noclobber_log_subdir=False, + log_name="log", timestamp=None): + self._setup_character_translation() + self._log_name = self._clean_filename(log_name) # last check. + self._log_dir = self._create_log_dir(log_dir) # will not clobber. + self._subdir,self._timestamp = self._create_log_subdir( + self._log_dir, noclobber_log_subdir, timestamp) + + def _setup_character_translation(self): + """Setup `._delete_chars` and `._trans_table` for `._clean_filename()`. + """ # generate lists of not-allowed characters - unaltered_chars = "-._" + string.digits + string.letters + unaltered_chars = '-._' + string.digits + string.letters mapped_pairs = {' ':'_'} - allowed_chars = unaltered_chars + "".join(mapped_pairs.keys()) + allowed_chars = unaltered_chars + ''.join(mapped_pairs.keys()) all_chars = string.maketrans('','') - self.delete_chars = all_chars.translate(all_chars, allowed_chars) - trans_from = "".join(mapped_pairs.keys()) - trans_to = "".join(mapped_pairs.values()) # same order as keys, since no modifications to mapped_pairs were made in between the two calls - self.transtable = string.maketrans(trans_from, trans_to) - - self._log_name = self._clean_filename(log_name) # never checked after this... - self._log_dir = self._create_logdir(log_dir) # will not clobber. - subdir, timestamp = self._create_logsubdir(self._log_dir, - noclobber_logsubdir, - timestamp) - self.subdir = subdir - self.timestamp = timestamp - def _clean_filename(self, filename) : - """ - Currently only works on filenames, since it deletes '/'. - If you need it to work on full paths, use os.path.split(your_path)[1] - to strip of the filename portion... + self._delete_chars = all_chars.translate(all_chars, allowed_chars) + trans_from = ''.join(mapped_pairs.keys()) + trans_to = ''.join(mapped_pairs.values()) + # values in trans_to are in the same order as the keys in + # trans_from, since no modifications to mapped_pairs were made + # in between the two calls. + self._trans_table = string.maketrans(trans_from, trans_to) + + def _clean_filename(self, filename): + """Remove troublesome characters from filenames. + + This method only works on filenames, since it deletes '/'. If + you need it to work on full paths, use + `os.path.split(your_path)` and clean the portions separately. + + Parameters + ---------- + filename : string + + Examples + -------- + >>> import shutil + >>> dl = DataLog(log_dir="test_clean_filename") + >>> dl._clean_filename('hi there') + 'hi_there' + >>> dl._clean_filename('hello\\tthe/castle') + 'hellothecastle' + >>> shutil.rmtree(dl._log_dir) """ - cleanname = filename.translate(self.transtable, self.delete_chars) + cleanname = filename.translate(self._trans_table, self._delete_chars) return cleanname - def _create_logdir(self, log_dir) : - log_dir = normalize_logdir(log_dir) - if not os.path.exists(log_dir) : + + def _create_log_dir(self, log_dir): + """Create a clean base log dir (if necessary). + + Parameters + ---------- + log_dir : path + Raw `log_dir` passed into `.__init__()`. + + Returns + ------- + log_dir : path + Normalized version of the input `log_dir`. + + Examples + -------- + >>> import shutil + >>> dl = DataLog(log_dir='test_create_log_dir') + >>> shutil.rmtree(dl._log_dir) + """ + log_dir = normalize_log_dir(log_dir) + if not os.path.exists(log_dir): os.mkdir(log_dir, 0755) return log_dir - def _create_logsubdir(self, log_dir, noclobber_logsubdir, - timestamp=None) : - if timestamp == None : + + def _create_log_subdir(self, log_dir, noclobber_log_subdir=False, + timestamp=None): + """Create a clean log dir for logging. + + Parameters + ---------- + log_dir : path + Normalized version of the input `log_dir`. + noclobber_log_subdir : bool + `noclobber_log_subdir` passed into `.__init__()`. + timestamp : string + Overide default `timestamp` (%Y%m%d). + + Returns + ------- + subdir : path + Path to the timestamped subdir of `log_dir`. + timestamp : string + The timestamp used to generate `subdir`. + + Examples + -------- + >>> import os + >>> import shutil + >>> dl = DataLog(log_dir='test_create_log_subdir', + ... timestamp='20101103') + >>> os.listdir(dl._log_dir) + ['20101103'] + >>> dl._create_log_subdir(dl._log_dir, noclobber_log_subdir=True, + ... timestamp=dl._timestamp) + Traceback (most recent call last): + ... + ErrorDirExists: test_create_log_subdir/20101103 exists + >>> dl._create_log_subdir(dl._log_dir, noclobber_log_subdir=False, + ... timestamp=dl._timestamp) + ('test_create_log_subdir/20101103', '20101103') + >>> dl._create_log_subdir(dl._log_dir) # doctest: +ELLIPSIS + ('test_create_log_subdir/...', '...') + >>> shutil.rmtree(dl._log_dir) + """ + if timestamp == None: timestamp = time.strftime("%Y%m%d") # %H%M%S subdir = os.path.join(log_dir, timestamp) - if os.path.exists(subdir) : - if noclobber_logsubdir: + if os.path.exists(subdir): + if noclobber_log_subdir: raise ErrorDirExists, "%s exists" % subdir - else : + else: os.mkdir(subdir, 0755) return (subdir, timestamp) - def get_filename(self, timestamp=None) : - """ - Get a filename (using localtime if timestamp==None), - appending integers as necessary to avoid clobbering. - For use in write() routines. - Returns (filepath, timestamp) + + def _get_filename(self, timestamp=None): + """Get a filename for a new data log for `.write()`. + + Append integers as necessary to avoid clobbering. Note that + the appended integers are *not* thread-safe. You need to + actually create the file to reserve the name. + + Parameters + ---------- + log_dir : path + Normalized version of the input `log_dir`. + noclobber_log_subdir : bool + `noclobber_log_subdir` passed into `.__init__()`. + timestamp : string + Overide default `timestamp` (%Y%m%d%H%M%S). + + Returns + ------- + filepath : path + Path to the timestamped log file. + timestamp : string + The timestamp used to generate `subdir`. + + Examples + -------- + >>> import shutil + >>> dl = DataLog(log_dir='test_get_filename', + ... log_name='my-log', timestamp='20101103') + >>> f,t = dl._get_filename('20100103235959') + >>> f + 'test_get_filename/20101103/20100103235959_my-log' + >>> t + '20100103235959' + >>> open(f, 'w').write('dummy content') + >>> f,t = dl._get_filename('20100103235959') + >>> f + 'test_get_filename/20101103/20100103235959_my-log_1' + >>> t + '20100103235959' + >>> open(f, 'w').write('dummy content') + >>> f,t = dl._get_filename('20100103235959') + >>> f + 'test_get_filename/20101103/20100103235959_my-log_2' + >>> t + '20100103235959' + >>> dl._get_filename() # doctest: +ELLIPSIS + ('test_get_filename/20101103/..._my-log', '...') + >>> shutil.rmtree(dl._log_dir) """ - if timestamp == None : + if timestamp == None: timestamp = time.strftime("%Y%m%d%H%M%S") filename = "%s_%s" % (timestamp, self._log_name) - fullname = os.path.join(self.subdir, filename) + fullname = os.path.join(self._subdir, filename) filepath = fullname i = 1 - while os.path.exists(filepath) : + while os.path.exists(filepath): filepath = "%s_%d" % (fullname, i) i+=1 return (filepath, timestamp) - def write(self, obj, timestamp=None) : - """ - Save object to a timestamped file with pickle. - If timestamp == None, use the current localtime. - Returns (filepath, timestamp) + + def write(self, obj, timestamp=None): + """Save object to a timestamped file with `cPickle`. + + Parameters + ---------- + obj : object + Object to save. + timestamp : string + Passed on to `._get_filename()`. + + Returns + ------- + filepath : path + Path to the timestamped log file. + timestamp : string + The timestamp used to generate the log file. + + Examples + -------- + >>> import shutil + >>> dl = DataLog(log_dir='test_write', + ... log_name='my-log', timestamp='20101103') + >>> f,t = dl.write([1, 2, 3]) + >>> a = pickle.load(open(f, 'rb')) + >>> a + [1, 2, 3] + >>> shutil.rmtree(dl._log_dir) """ - filepath, timestamp = self.get_filename(timestamp) - fd = open(filepath, 'wb') - os.chmod(filepath, 0644) - pickle.dump(obj, fd) - fd.close() + filepath, timestamp = self._get_filename(timestamp) + with open(filepath, 'wb') as fd: + os.chmod(filepath, 0644) + pickle.dump(obj, fd) return (filepath, timestamp) - def write_binary(self, binary_string, timestamp=None) : - """ - Save binary_string to a timestamped file. - If timestamp == None, use the current localtime. - Returns (filepath, timestamp) + + def write_binary(self, binary_string, timestamp=None): + """Save a binary string to a timestamped file. + + Parameters + ---------- + binary_string : buffer + Binary string to save. + timestamp : string + Passed on to `._get_filename()`. + + Returns + ------- + filepath : path + Path to the timestamped log file. + timestamp : string + The timestamp used to generate the log file(s). + + Examples + -------- + >>> import shutil + >>> import numpy + >>> dl = DataLog(log_dir='test_write_binary', + ... log_name='my-log', timestamp='20101103') + >>> data = numpy.arange(5, dtype=numpy.uint16) + >>> filepath,ts = dl.write_binary(data.tostring()) + >>> data_in = numpy.fromfile(filepath, dtype=numpy.uint16, count=-1) + >>> data_in + array([0, 1, 2, 3, 4], dtype=uint16) + >>> (data == data_in).all() + True + >>> shutil.rmtree(dl._log_dir) """ - filepath, timestamp = self.get_filename(timestamp) + filepath, timestamp = self._get_filename(timestamp) # open a new file in readonly mode, don't clobber. fd = os.open(filepath, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0644) bytes_written = 0 bytes_remaining = len(binary_string) - while bytes_remaining > 0 : + while bytes_remaining > 0: bw = os.write(fd, binary_string[bytes_written:]) bytes_written += bw bytes_remaining -= bw os.close(fd) return (filepath, timestamp) - def _write_dict_of_arrays(self, d, base_filepath) : + + def _write_dict_of_arrays(self, d, base_filepath): + """Save dict of (string, numpy_array) pairs under `base_filepath`. + + Parameters + ---------- + d : dict + Dictionary to save. + base_filepath : path + Path for table of contents and from which per-pair paths + are constructed. + """ # open a new file in readonly mode, don't clobber. bfd = open(base_filepath, 'w', 0644) bfd.write("Contents (key : file-extension : format):\n") - for key in d.keys() : + for key in d.keys(): clean_key = self._clean_filename(key) bfd.write("%s : %s : %s\n" % (key, clean_key, str(d[key].dtype))) # write the keyed array to it's own file filepath = "%s_%s" % (base_filepath, clean_key) d[key].tofile(filepath) bfd.close() - def write_dict_of_arrays(self, d, timestamp=None) : - """ - Save dict of (string, numpy_array) pairs to timestamped files. - If timestamp == None, use the current localtime. - Returns (base_filepath, timestamp) + + def write_dict_of_arrays(self, d, timestamp=None): + """Save dict of (string, numpy_array) pairs to timestamped files. + + Parameters + ---------- + d : dict + Dictionary to save. + timestamp : string + Passed on to `._get_filename()`. + + Returns + ------- + filepath : path + Path to the timestamped log file. + timestamp : string + The timestamp used to generate the log file(s). + + Examples + -------- + >>> import os + >>> import shutil + >>> import numpy + >>> dl = DataLog(log_dir='test_write_dict_of_arrays', + ... log_name='my-log', timestamp='20101103') + >>> d = {'data1':numpy.arange(5, dtype=numpy.int16), + ... 'd\/at:$a 2':numpy.arange(3, dtype=numpy.float64)} + >>> filepath,ts = dl.write_dict_of_arrays( + ... d, timestamp='20101103235959') + >>> filepath + 'test_write_dict_of_arrays/20101103/20101103235959_my-log' + >>> print '\\n'.join(sorted(os.listdir(dl._subdir))) + 20101103235959_my-log + 20101103235959_my-log_data1 + 20101103235959_my-log_data_2 + >>> contents = open(filepath, 'r').read() + >>> print contents + Contents (key : file-extension : format): + data1 : data1 : int16 + d\/at:$a 2 : data_2 : float64 + + >>> data1_in = numpy.fromfile( + ... filepath+'_data1', dtype=numpy.int16, count=-1) + >>> data1_in + array([0, 1, 2, 3, 4], dtype=int16) + >>> data2_in = numpy.fromfile( + ... filepath+'_data_2', dtype=numpy.float64, count=-1) + >>> data2_in + array([ 0., 1., 2.]) + >>> shutil.rmtree(dl._log_dir) """ - base_filepath, timestamp = self.get_filename(timestamp) + base_filepath,timestamp = self._get_filename(timestamp) self._write_dict_of_arrays(d, base_filepath) return (base_filepath, timestamp) -class data_load : - """ - Loads data logged by data_log. + +class DataLoad (object): + """Load data logged by `DataLog`. """ - def read(self, file) : - """ - Load an object saved with data_log.write() - """ - return pickle.load(open(file, 'rb')) - def read_binary(self, file) : + def read(self, filename): + """Load an object saved with `DataLog.write()`. + + Parameters + ---------- + filename : path + `filename` returned by `DataLog.write()`. + + Returns + ------- + obj : object + The saved object. + + Examples + -------- + >>> import shutil + >>> dl = DataLog(log_dir='test_read', + ... log_name='my-log', timestamp='20101103') + >>> f,t = dl.write([1, 2, 3]) + >>> load = DataLoad() + >>> d = load.read(f) + >>> d + [1, 2, 3] + >>> shutil.rmtree(dl._log_dir) """ - Load an object saved with data_log.write_binary() - The file-name must not have been altered. + return pickle.load(open(filename, 'rb')) + + def read_binary(self, filename): + """Load an object saved with `DataLog.write_binary()`. + + Warning: this method *requires* `filename` to end with + `_float` and *assumes* that the file contains `numpy.float` + data. That is terrible. Use `h5py` instead of this module! + + Parameters + ---------- + filename : path + `filename` returned by `DataLog.write_binary()`. + + Returns + ------- + obj : object + The saved object. + + Examples + -------- + >>> import shutil + >>> import numpy + >>> dl = DataLog(log_dir='test_read_binary', + ... log_name='my-log_float', timestamp='20101103') + >>> f,t = dl.write_binary(numpy.array([1, 2, 3], dtype=numpy.float)) + >>> load = DataLoad() + >>> d = load.read_binary(f) + >>> d + array([ 1., 2., 3.]) + >>> shutil.rmtree(dl._log_dir) """ - type = file.split("_")[-1] - if type == "float" : + type_ = filename.split("_")[-1] + if type_ == "float": t = numpy.float - else : - raise Exception, "read_binary() not implemented for type %s" % (type) - return numpy.fromfile(file, dtype=t) - def read_dict_of_arrays(self, basefile) : - """ - Load an object saved with data_log.write_binary() - The file-names must not have been altered. + else: + raise Exception( + "read_binary() not implemented for type %s" % (type_)) + return numpy.fromfile(filename, dtype=t) + + def read_dict_of_arrays(self, basefile): + """Load an object saved with `DataLog.write_dict_of_arrays()`. + + The filenames must not have been altered. + + Parameters + ---------- + filename : path + `filename` returned by `DataLog.write_dict_of_arrays()`. + + Returns + ------- + obj : object + The saved object. + + Examples + -------- + >>> import pprint + >>> import shutil + >>> import numpy + >>> dl = DataLog(log_dir='test_read_dict_of_arrays', + ... log_name='my-log', timestamp='20101103') + >>> d = {'data1':numpy.arange(5, dtype=numpy.int16), + ... 'd\/at:$a 2':numpy.arange(3, dtype=numpy.float64)} + >>> f,t = dl.write_dict_of_arrays(d, timestamp='20101103235959') + >>> load = DataLoad() + >>> d = load.read_dict_of_arrays(f) + >>> pprint.pprint(d) + {'d\\\\/at:$a 2': array([ 0., 1., 2.]), + 'data1': array([0, 1, 2, 3, 4], dtype=int16)} + >>> shutil.rmtree(dl._log_dir) """ obj = {} i=0 realbasefile = os.path.realpath(basefile) - for line in file(realbasefile) : + for line in file(realbasefile): if i > 0 : # ignore first line ldata = line.split(' : ') name = ldata[0] fpath = "%s_%s" % (realbasefile, ldata[1]) - exec 'typ = numpy.%s' % ldata[2] - obj[name] = numpy.fromfile(fpath, dtype=typ) + type_ = getattr(numpy, ldata[2].strip()) + obj[name] = numpy.fromfile(fpath, dtype=type_) i += 1 return obj -_test_dir = "." - -def _check_data_logsubdir_clobber() : - log1 = data_log(_test_dir, noclobber_logsubdir=True) - try : - log2 = data_log(_test_dir, noclobber_logsubdir=True) - raise Error, "Didn't detect old log" - except ErrorDirExists : - pass # everything as it should be - os.rmdir(log1.subdir) - -def _check_data_log_filenames() : - data = {"Test":True, "Data":[1,2,3,4]} - log = data_log(_test_dir, noclobber_logsubdir=True) - files = [None]*10 - for i in range(10): - files[i], ts = log.write(data) - print "Contents of log directory (should be 10 identical logs)" - os.system('ls -l %s' % log.subdir) - for file in files : - os.remove( file ) - os.rmdir(log.subdir) - -def _check_data_log_pickle_integrity() : - data = {"Test":True, "Data":[1,2,3,4]} - # save the data - log = data_log(_test_dir, noclobber_logsubdir=True) - filepath, ts = log.write(data) - # read it back in - fd = open(filepath, 'rb') - data_in = pickle.load(fd) - fd.close() - # compare - if data != data_in : - print "Saved : ", data - print "Read back: ", data_in - raise Error, "Poorly pickled" - os.remove(filepath) - os.rmdir(log.subdir) - -def _check_data_log_binary_integrity() : - from numpy import zeros, uint16, fromfile - npts = 100 - data = zeros((npts,), dtype=uint16) - for i in range(npts) : - data[i] = i - # save the data - log = data_log(_test_dir, noclobber_logsubdir=True) - filepath, ts = log.write_binary(data.tostring()) - # read it back in - data_in = fromfile(filepath, dtype=uint16, count=-1) - # compare - if npts != len(data_in) : - raise Error, "Saved %d uint16s, read %d" % (npts, len(data_in)) - for i in range(npts) : - if data_in[i] != data[i] : - print "Disagreement in element %d" % i - print "Saved %d, read back %d" % (data[i], data_in[i]) - raise Error, "Poorly saved" - os.remove(filepath) - os.rmdir(log.subdir) - -def _check_data_loc_dict_of_arrays() : - from numpy import zeros, uint16, fromfile - npts = 100 - data1 = zeros((npts,), dtype=uint16) - for i in range(npts) : - data1[i] = i - data2 = zeros((npts,), dtype=uint16) - for i in range(npts) : - data2[i] = npts-i - data={"data1":data1, "d\/at:$a 2":data2} - # save the data - log = data_log(_test_dir, noclobber_logsubdir=True) - filepath, ts = log.write_dict_of_arrays(data) - # checking - print "Contents of log directory (should be 3 logs)" - os.system('ls -l %s' % log.subdir) - print "The table of contents file:" - os.system('cat %s' % (filepath)) - data1_in = fromfile(filepath+"_data1", dtype=uint16) - data2_in = fromfile(filepath+"_data_2", dtype=uint16) - for i in range(npts) : - if data1_in[i] != data1[i] : - print "Disagreement in element %d of data1" % i - print "Saved %d, read back %d" % (data1[i], data1_in[i]) - raise Error, "Poorly saved" - if data2_in[i] != data2[i] : - print "Disagreement in element %d of data2" % i - print "Saved %d, read back %d" % (data2[i], data2_in[i]) - raise Error, "Poorly saved" - os.remove(filepath) - os.remove(filepath+"_data1") - os.remove(filepath+"_data_2") - os.rmdir(log.subdir) - -def test() : - _check_data_logsubdir_clobber() - _check_data_log_filenames() - _check_data_log_pickle_integrity() - _check_data_log_binary_integrity() - _check_data_loc_dict_of_arrays() - -if __name__ == "__main__" : + +def test(): + import doctest + import sys + + result = doctest.testmod() + sys.exit(min(result.failed, 127)) + +if __name__ == "__main__": test() -- 2.26.2