Repository: NERSC/pytokio

Issue:
make tokio.timeseries independent of connectors
At present, `tokio.timeseries` imports `tokio.connectors.hdf5` to facilitate translating to/from HDF5 format.  This causes a fragile circular dependency between the two which causes problems when importing other connectors that depend on `tokio.connectors.common`.

There's no reason for the TimeSeries class to have dependencies on HDF5; that HDF5-related functionality should be in the HDF5 connector anyway.

Discussion:

Patch:
diff --git a/tokio/cli/archive_collectdes.py b/tokio/cli/archive_collectdes.py
index 505932c..7bd3c77 100644
--- a/tokio/cli/archive_collectdes.py
+++ b/tokio/cli/archive_collectdes.py
@@ -382,19 +382,19 @@ def pages_to_hdf5(pages, output_file, init_start, init_end, query_start, query_e
                                                      end=global_end,
                                                      timestep=timestep,
                                                      num_columns=num_columns,
-                                                     column_names=datasets[real_dataset_name].columns,
-                                                     hdf5_file=None)
+                                                     column_names=datasets[real_dataset_name].columns)
 
             # we can't update an average, so zero out all prior values that
             # we will be overwriting
             reset_timeseries(datasets[real_dataset_name], query_start, query_end)
         else:
-            timeseries = tokio.timeseries.TimeSeries(dataset_name=hdf5_dataset_name,
-                                                     start=init_start,
-                                                     end=init_end,
-                                                     timestep=timestep,
-                                                     num_columns=num_columns,
-                                                     hdf5_file=hdf5_file)
+            timeseries = hdf5_file.to_timeseries(dataset_name=hdf5_dataset_name)
+            if timeseries is None:
+                timeseries = tokio.timeseries.TimeSeries(dataset_name=hdf5_dataset_name,
+                                                         start=init_start,
+                                                         end=init_end,
+                                                         timestep=timestep,
+                                                         num_columns=num_columns)
         datasets[dataset_name] = timeseries
 
     # Process all pages retrieved (this is computationally expensive)
@@ -433,7 +433,7 @@ def pages_to_hdf5(pages, output_file, init_start, init_end, query_start, query_e
     _time0 = time.time()
     for dataset_name, dataset in datasets.items():
         if '/_' not in dataset_name:
-            dataset.commit_dataset(hdf5_file)
+            hdf5_file.commit_timeseries(dataset)
 
     if tokio.debug.DEBUG:
         print("Committed data to disk in %.4f seconds" % (time.time() - _time0))
diff --git a/tokio/cli/archive_lmtdb.py b/tokio/cli/archive_lmtdb.py
index 2136701..346b4c3 100644
--- a/tokio/cli/archive_lmtdb.py
+++ b/tokio/cli/archive_lmtdb.py
@@ -6,10 +6,10 @@ import sys
 import datetime
 import argparse
 import warnings
-import h5py
 import tokio.debug
 import tokio.timeseries
 import tokio.connectors.lmtdb
+import tokio.connectors.hdf5
 from tokio.common import isstr
 
 DATE_FMT = "%Y-%m-%dT%H:%M:%S"
@@ -156,7 +156,6 @@ class DatasetDict(dict):
                                                                  timestep=self.timestep,
                                                                  num_columns=len(columns),
                                                                  column_names=columns,
-                                                                 hdf5_file=None,
                                                                  sort_hex=self.sort_hex)
 
     def finalize(self):
@@ -455,17 +454,21 @@ def init_hdf5_file(datasets, init_start, init_end, hdf5_file):
                 warnings.warn("Dataset key %s is not in schema" % dataset_name)
             continue
         if hdf5_dataset_name not in hdf5_file:
-            new_dataset = tokio.timeseries.TimeSeries(dataset_name=hdf5_dataset_name,
-                                                      start=init_start,
-                                                      end=init_end,
-                                                      timestep=dataset.timestep,
-                                                      num_columns=dataset.dataset.shape[1],
-                                                      hdf5_file=hdf5_file)
-            new_dataset.commit_dataset(hdf5_file)
+            # attempt to convert dataset into a timeseries
+            timeseries = hdf5_file.to_timeseries(dataset_name=hdf5_dataset_name)
+
+            # if dataset -> timeseries failed, create and commit a new, empty timeseries
+            if timeseries is None:
+                timeseries = tokio.timeseries.TimeSeries(dataset_name=hdf5_dataset_name,
+                                                         start=init_start,
+                                                         end=init_end,
+                                                         timestep=dataset.timestep,
+                                                         num_columns=dataset.dataset.shape[1])
+                hdf5_file.commit_timeseries(timeseries=timeseries)
             print("Initialized %s in %s with size %s" % (
                 hdf5_dataset_name,
                 hdf5_file.name,
-                new_dataset.dataset.shape))
+                timeseries.dataset.shape))
 
 def archive_lmtdb(lmtdb, init_start, init_end, timestep, output_file, query_start, query_end):
     """
@@ -481,14 +484,14 @@ def archive_lmtdb(lmtdb, init_start, init_end, timestep, output_file, query_star
 
     datasets.finalize()
 
-    with h5py.File(output_file) as hdf5_file:
+    with tokio.connectors.hdf5.Hdf5(output_file) as hdf5_file:
         hdf5_file.attrs['version'] = SCHEMA_VERSION
 
         init_hdf5_file(datasets, init_start, init_end, hdf5_file)
 
         for dataset in datasets.values():
             print("Writing out %s" % dataset.dataset_name)
-            dataset.commit_dataset(hdf5_file)
+            hdf5_file.commit_timeseries(dataset)
 
     tokio.debug.debug_print("Wrote output to %s" % output_file)
 
diff --git a/tokio/connectors/hdf5.py b/tokio/connectors/hdf5.py
index b4a9cb4..2931ee4 100755
--- a/tokio/connectors/hdf5.py
+++ b/tokio/connectors/hdf5.py
@@ -7,12 +7,13 @@ derived datasets dynamically.
 """
 
 import math
+import time
 import datetime
 import warnings
 import h5py
 import numpy
 import pandas
-from tokio.common import isstr
+import tokio.common
 from tokio.connectors._hdf5 import (convert_counts_rates, #pylint: disable=unused-import
                                     map_dataset,
                                     demux_column,
@@ -641,7 +642,7 @@ class Hdf5(h5py.File):
             key (str): The standard name of a dataset to be accessed.
 
         Returns:
-            h5py.Dataset or numpy.ndarray: 
+            h5py.Dataset or numpy.ndarray:
 
               * h5py.Dataset if key is a literal dataset name
               * h5py.Dataset if key maps directly to a literal dataset name
@@ -689,7 +690,7 @@ class Hdf5(h5py.File):
             return key, None
 
         # Straight mapping between the key and a dataset
-        key = key.lstrip('/') if isstr(key) else key
+        key = key.lstrip('/') if tokio.common.isstr(key) else key
         if key in self.schema:
             hdf5_key = self.schema.get(key)
             if super(Hdf5, self).__contains__(hdf5_key):
@@ -725,8 +726,7 @@ class Hdf5(h5py.File):
                 version = self._version
             if isinstance(version, bytes):
                 return version.decode() # for python3
-            else:
-                return version
+            return version
 
     def set_version(self, version, dataset_name=None):
         """Set the version attribute from an HDF5 file dataset
@@ -742,12 +742,13 @@ class Hdf5(h5py.File):
         if dataset_name is None:
             self._version = version
             return self._version
-        else:
-            # resolve dataset name
-            dataset = self.__getitem__(dataset_name)
-            if dataset is None:
-                raise KeyError("Dataset %s does not exist" % dataset_name)
-            dataset.attrs["version"] = version
+
+        # resolve dataset name
+        dataset = self.__getitem__(dataset_name)
+        if dataset is None:
+            raise KeyError("Dataset %s does not exist" % dataset_name)
+        dataset.attrs["version"] = version
+        return version
 
     def get_columns(self, dataset_name):
         """Get the column names of a dataset
@@ -760,7 +761,7 @@ class Hdf5(h5py.File):
         """
         if self.get_version(dataset_name=dataset_name) is None:
             return self._get_columns_h5lmt(dataset_name)
-        
+
         return self.__getitem__(dataset_name).attrs[COLUMN_NAME_KEY].astype('U')
 
     def _get_columns_h5lmt(self, dataset_name):
@@ -949,7 +950,7 @@ class Hdf5(h5py.File):
 
             # if we have more indices than values, this is unrecoverable because
             # we don't know where in `values` data was written twice
-            if (num_indices > num_indices_expected):
+            if num_indices > num_indices_expected:
                 raise IndexError(warning_msg)
 
             # h5lmt format doesn't support distinguishing uninitialized elements
@@ -968,6 +969,178 @@ class Hdf5(h5py.File):
                                 index=indices,
                                 columns=columns)
 
+    def to_timeseries(self, dataset_name, light=False):
+        """Creates a TimeSeries representation of a dataset
+
+        Create a TimeSeries dataset object with the data from an existing HDF5
+        dataset.
+
+        Responsible for setting timeseries.dataset_name, timeseries.columns, timeseries.dataset,
+        timeseries.dataset_metadata, timeseries.group_metadata, timeseries.timestamp_key
+
+        Args:
+            dataset_name (str): Name of existing dataset in self to convert into
+                a TimeSeries object
+            light (bool): If True, don't actually load datasets into memory;
+                reference them directly into the HDF5 file
+
+        Returns:
+            tokio.timeseries.TimeSeries: The in-memory representation of the
+            given dataset.
+        """
+        timeseries = tokio.timeseries.TimeSeries()
+        timeseries.dataset_name = dataset_name
+
+        try:
+            dataset = self[dataset_name]
+        except KeyError:
+            # can't attach because dataset doesn't exist; pass this back to caller so it can init
+            return None
+
+        timeseries.dataset = dataset if light else dataset[:, :]
+
+        # load and decode version of dataset and file schema
+        timeseries.global_version = self['/'].attrs.get('version')
+        timeseries.version = self.get_version(dataset_name)
+        if isinstance(timeseries.version, bytes):
+            timeseries.version = timeseries.version.decode()
+
+        # copy columns into memory
+        columns = self.get_columns(dataset_name)
+        timeseries.set_columns(columns)
+
+        # copy metadata into memory
+        for key, value in dataset.attrs.items():
+            if isinstance(value, bytes):
+                timeseries.dataset_metadata[key] = value.decode()
+            else:
+                timeseries.dataset_metadata[key] = value
+        for key, value in dataset.parent.attrs.items():
+            if isinstance(value, bytes):
+                timeseries.group_metadata[key] = value.decode()
+            else:
+                timeseries.group_metadata[key] = value
+
+        timeseries.timestamp_key = get_timestamps_key(self, dataset_name)
+        timeseries.timestamps = self[timeseries.timestamp_key]
+        timeseries.timestamps = timeseries.timestamps if light else timeseries.timestamps[:]
+
+        timeseries.timestep = timeseries.timestamps[1] - timeseries.timestamps[0]
+        return timeseries
+
+    def commit_timeseries(self, timeseries, **kwargs):
+        """Writes contents of a TimeSeries object into a group
+
+        Args:
+            timeseries (tokio.timeseries.TimeSeries): the time series to save
+                as a dataset within self
+            kwargs (dict): Extra arguments to pass to self.create_dataset()
+        """
+        extra_dataset_args = {
+            'dtype': 'f8',
+            'chunks': True,
+            'compression': 'gzip',
+        }
+        extra_dataset_args.update(kwargs)
+
+        # Create the dataset in the HDF5 file (if necessary)
+        if timeseries.dataset_name in self:
+            dataset_hdf5 = self[timeseries.dataset_name]
+        else:
+            dataset_hdf5 = self.create_dataset(name=timeseries.dataset_name,
+                                               shape=timeseries.dataset.shape,
+                                               **extra_dataset_args)
+
+        # when timestamp_key has been left empty, use the default
+        timestamp_key = timeseries.timestamp_key
+        if timestamp_key is None:
+            timestamp_key = '/'.join(timeseries.dataset_name.split('/')[0:-1] \
+                                 + [DEFAULT_TIMESTAMP_DATASET])
+
+        # Create the timestamps in the HDF5 file (if necessary) and calculate
+        # where to insert our data into the HDF5's dataset
+        if timestamp_key not in self:
+            timestamps_hdf5 = self.create_dataset(name=timestamp_key,
+                                                  shape=timeseries.timestamps.shape,
+                                                  dtype='i8')
+            # Copy the in-memory timestamp dataset into the HDF5 file
+            timestamps_hdf5[:] = timeseries.timestamps[:]
+            t_start = 0
+            t_end = timeseries.timestamps.shape[0]
+            start_timestamp = timeseries.timestamps[0]
+            end_timestamp = timeseries.timestamps[-1] + timeseries.timestep
+        else:
+            existing_timestamps = self.get_timestamps(timeseries.dataset_name)
+            t_start, t_end = get_insert_indices(timeseries.timestamps, existing_timestamps)
+
+            if t_start < 0 \
+            or t_start > (len(existing_timestamps) - 2) \
+            or t_end < 1 \
+            or t_end > len(existing_timestamps):
+                raise IndexError("cannot commit dataset that is not a subset of existing data")
+
+            start_timestamp = existing_timestamps[0]
+            end_timestamp = existing_timestamps[-1] + timeseries.timestep
+
+        # Make sure that the start/end timestamps are consistent with the HDF5
+        # file's global time range
+        if 'start' not in self.attrs:
+            self.attrs['start'] = start_timestamp
+            self.attrs['end'] = end_timestamp
+        else:
+            if self.attrs['start'] != start_timestamp \
+            or self.attrs['end'] != end_timestamp:
+#               warnings.warn(
+                raise IndexError("Mismatched start or end values:  %d != %d or %d != %d" % (
+                    start_timestamp, self.attrs['start'],
+                    end_timestamp, self.attrs['end']))
+
+        # If we're updating an existing dataset, use its column names and ordering.
+        # Otherwise sort the columns before committing them.
+        if COLUMN_NAME_KEY in dataset_hdf5.attrs:
+            timeseries.rearrange_columns(timeseries.columns)
+        else:
+            timeseries.sort_columns()
+
+        # Copy the in-memory dataset into the HDF5 file
+        dataset_hdf5[t_start:t_end, :] = timeseries.dataset[:, :]
+
+        # Copy column names into metadata before committing metadata
+        timeseries.dataset_metadata[COLUMN_NAME_KEY] = timeseries.columns
+        timeseries.dataset_metadata['updated'] = int(time.mktime(datetime.datetime.now().timetuple()))
+
+        # If timeseries.version was never set, don't set a dataset-level version in the HDF5
+        if timeseries.version is not None:
+            self.set_version(timeseries.version, dataset_name=timeseries.dataset_name)
+
+        # Set the file's global version to indicate its schema
+        if timeseries.global_version is not None:
+            self['/'].attrs['version'] = timeseries.global_version
+
+        # Insert/update dataset metadata
+        for key, value in timeseries.dataset_metadata.items():
+            # special hack for column names
+            if key == COLUMN_NAME_KEY:
+                # note: the behavior of numpy.string_(x) where
+                # type(x) == numpy.array is _different_ in python2 vs. python3.
+                # Python3 happily converts each element to a numpy.string_,
+                # while Python2 first calls a.__repr__ to turn it into a single
+                # string, then converts that to numpy.string_.
+                dataset_hdf5.attrs[key] = numpy.array([numpy.string_(x) for x in value])
+            elif tokio.common.isstr(value):
+                dataset_hdf5.attrs[key] = numpy.string_(value)
+            elif value is None:
+                warnings.warn("Skipping attribute %s (null value) for %s" % (key, timeseries.dataset_name))
+            else:
+                dataset_hdf5.attrs[key] = value
+
+        # Insert/update group metadata
+        for key, value in timeseries.group_metadata.items():
+            if tokio.common.isstr(value):
+                dataset_hdf5.parent.attrs[key] = numpy.string_(value)
+            else:
+                dataset_hdf5.parent.attrs[key] = value
+
 def missing_values(dataset, inverse=False):
     """Identify matrix values that are missing
 
@@ -1007,3 +1180,22 @@ def reduce_dataset_name(key):
     if key.endswith('/missing'):
         return tuple(key.rsplit('/', 1))
     return key, None
+
+def get_insert_indices(my_timestamps, existing_timestamps):
+    """
+    Given new timestamps and an existing series of timestamps, find the indices
+    overlap so that new data can be inserted into the middle of an existing
+    dataset
+    """
+    existing_timestep = existing_timestamps[1] - existing_timestamps[0]
+    my_timestep = my_timestamps[1] - my_timestamps[0]
+
+    # make sure the time delta is ok
+    if existing_timestep != my_timestep:
+        raise Exception("Existing dataset has different timestep (mine=%d, existing=%d)"
+                        % (my_timestep, existing_timestep))
+
+    my_offset = (my_timestamps[0] - existing_timestamps[0]) // existing_timestep
+    my_end = my_offset + len(my_timestamps)
+
+    return my_offset, my_end
diff --git a/tokio/timeseries.py b/tokio/timeseries.py
index e40bc46..4e3c878 100644
--- a/tokio/timeseries.py
+++ b/tokio/timeseries.py
@@ -10,7 +10,6 @@ import time
 import datetime
 import warnings
 import numpy
-import tokio.connectors.hdf5
 from tokio.common import isstr
 
 class TimeSeries(object):
@@ -22,7 +21,7 @@ class TimeSeries(object):
     def __init__(self, dataset_name=None,
                  start=None, end=None, timestep=None, num_columns=None,
                  column_names=None, timestamp_key=None,
-                 hdf5_file=None, sort_hex=False):
+                 sort_hex=False):
 
         # numpy.ndarray of timestamp measurements
         self.timestamps = None
@@ -50,15 +49,10 @@ class TimeSeries(object):
         # string describing schema version
         self.global_version = None
 
-        # attempt to attach the object if requested.
-        if dataset_name is not None:
-            attached = False
-            if hdf5_file is not None:
-                attached = self.attach(hdf5_file, dataset_name)
-            # if attach fails due to dataset being uninitialized, initialize it instead
-            if not attached and start and end and timestep and num_columns:
-                self.init(start, end, timestep, num_columns, dataset_name,
-                          column_names, timestamp_key)
+        # attempt to initialize the object if fields are supplied
+        if dataset_name is not None and start and end and timestep and num_columns:
+            self.init(start, end, timestep, num_columns, dataset_name,
+                column_names, timestamp_key)
 
     def init(self, start, end, timestep, num_columns, dataset_name,
              column_names=None, timestamp_key=None):
@@ -106,168 +100,12 @@ class TimeSeries(object):
         """Set the timestamp key
 
         Args:
-            timestamp_key (str or None): The key for the timestamp dataset.  If
-                None, use the default
+            timestamp_key (str): The key for the timestamp dataset.
             safe (bool): If true, do not overwrite an existing timestamp key
         """
         # Root the timestamp_key at the same parent as the dataset
         if not safe or self.timestamp_key is None:
-            if timestamp_key is None:
-                self.timestamp_key = '/'.join(self.dataset_name.split('/')[0:-1] \
-                                     + [tokio.connectors.hdf5.DEFAULT_TIMESTAMP_DATASET])
-            else:
-                self.timestamp_key = timestamp_key
-
-    def attach(self, hdf5_file, dataset_name, light=False):
-        """
-        Populate a TimeSeries dataset object with the data from an existing HDF5
-        dataset.  If light is True, don't actually load datasets into memory;
-        reference them directly into the HDF5 file
-
-        Responsible for setting self.dataset_name, self.columns, self.dataset,
-        self.dataset_metadata, self.group_metadata, self.timestamp_key
-        """
-        self.dataset_name = dataset_name
-
-        try:
-            dataset = hdf5_file[dataset_name]
-        except KeyError:
-            # can't attach because dataset doesn't exist; pass this back to caller so it can init
-            return False
-
-        self.dataset = dataset if light else dataset[:, :]
-
-        # load and decode version of dataset and file schema
-        self.global_version = hdf5_file['/'].attrs.get('version')
-        self.version = hdf5_file.get_version(dataset_name)
-        if isinstance(self.version, bytes):
-            self.version = self.version.decode()
-
-        # copy columns into memory
-        columns = hdf5_file.get_columns(dataset_name)
-        self.set_columns(columns)
-
-        # copy metadata into memory
-        for key, value in dataset.attrs.items():
-            if isinstance(value, bytes):
-                self.dataset_metadata[key] = value.decode()
-            else:
-                self.dataset_metadata[key] = value
-        for key, value in dataset.parent.attrs.items():
-            if isinstance(value, bytes):
-                self.group_metadata[key] = value.decode()
-            else:
-                self.group_metadata[key] = value
-
-        self.timestamp_key = tokio.connectors.hdf5.get_timestamps_key(hdf5_file, dataset_name)
-        self.timestamps = hdf5_file[self.timestamp_key]
-        self.timestamps = self.timestamps if light else self.timestamps[:]
-
-        self.timestep = self.timestamps[1] - self.timestamps[0]
-        return True
-
-    def commit_dataset(self, hdf5_file, **kwargs):
-        """
-        Write contents of this object into an HDF5 file group
-        """
-        extra_dataset_args = {
-            'dtype': 'f8',
-            'chunks': True,
-            'compression': 'gzip',
-        }
-        extra_dataset_args.update(kwargs)
-
-        # Create the dataset in the HDF5 file (if necessary)
-        if self.dataset_name in hdf5_file:
-            dataset_hdf5 = hdf5_file[self.dataset_name]
-        else:
-            dataset_hdf5 = hdf5_file.create_dataset(name=self.dataset_name,
-                                                    shape=self.dataset.shape,
-                                                    **extra_dataset_args)
-
-        # Create the timestamps in the HDF5 file (if necessary) and calculate
-        # where to insert our data into the HDF5's dataset
-        if self.timestamp_key not in hdf5_file:
-            timestamps_hdf5 = hdf5_file.create_dataset(name=self.timestamp_key,
-                                                       shape=self.timestamps.shape,
-                                                       dtype='i8')
-            # Copy the in-memory timestamp dataset into the HDF5 file
-            timestamps_hdf5[:] = self.timestamps[:]
-            t_start = 0
-            t_end = self.timestamps.shape[0]
-            start_timestamp = self.timestamps[0]
-            end_timestamp = self.timestamps[-1] + self.timestep
-        else:
-            existing_timestamps = tokio.connectors.hdf5.get_timestamps(hdf5_file, self.dataset_name)
-            t_start, t_end = get_insert_indices(self.timestamps, existing_timestamps)
-
-            if t_start < 0 \
-            or t_start > (len(existing_timestamps) - 2) \
-            or t_end < 1 \
-            or t_end > len(existing_timestamps):
-                raise IndexError("cannot commit dataset that is not a subset of existing data")
-
-            start_timestamp = existing_timestamps[0]
-            end_timestamp = existing_timestamps[-1] + self.timestep
-
-        # Make sure that the start/end timestamps are consistent with the HDF5
-        # file's global time range
-        if 'start' not in hdf5_file.attrs:
-            hdf5_file.attrs['start'] = start_timestamp
-            hdf5_file.attrs['end'] = end_timestamp
-        else:
-            if hdf5_file.attrs['start'] != start_timestamp \
-            or hdf5_file.attrs['end'] != end_timestamp:
-#               warnings.warn(
-                raise IndexError("Mismatched start or end values:  %d != %d or %d != %d" % (
-                    start_timestamp, hdf5_file.attrs['start'],
-                    end_timestamp, hdf5_file.attrs['end']))
-
-        # If we're updating an existing HDF5, use its column names and ordering.
-        # Otherwise sort the columns before committing them.
-        if tokio.connectors.hdf5.COLUMN_NAME_KEY in dataset_hdf5.attrs:
-            self.rearrange_columns(self.columns)
-        else:
-            self.sort_columns()
-
-        # Copy the in-memory dataset into the HDF5 file
-        dataset_hdf5[t_start:t_end, :] = self.dataset[:, :]
-
-        # Copy column names into metadata before committing metadata
-        self.dataset_metadata[tokio.connectors.hdf5.COLUMN_NAME_KEY] = self.columns
-        self.dataset_metadata['updated'] = int(time.mktime(datetime.datetime.now().timetuple()))
-
-        # If self.version was never set, don't set a dataset-level version in the HDF5
-        if self.version is not None:
-            hdf5_file.set_version(self.version, dataset_name=self.dataset_name)
-
-        # Set the file's global version to indicate its schema
-        if self.global_version is not None:
-            hdf5_file['/'].attrs['version'] = self.global_version
-
-        # Insert/update dataset metadata
-        for key, value in self.dataset_metadata.items():
-            # special hack for column names
-            if (key == tokio.connectors.hdf5.COLUMN_NAME_KEY):
-                # note: the behavior of numpy.string_(x) where
-                # type(x) == numpy.array is _different_ in python2 vs. python3.
-                # Python3 happily converts each element to a numpy.string_,
-                # while Python2 first calls a.__repr__ to turn it into a single
-                # string, then converts that to numpy.string_.
-                dataset_hdf5.attrs[key] = numpy.array([numpy.string_(x) for x in value])
-            elif isstr(value):
-                dataset_hdf5.attrs[key] = numpy.string_(value)
-            elif value is None:
-                warnings.warn("Skipping attribute %s (null value) for %s" % (key, self.dataset_name))
-            else:
-                dataset_hdf5.attrs[key] = value
-
-        # Insert/update group metadata
-        for key, value in self.group_metadata.items():
-            if isstr(value):
-                dataset_hdf5.parent.attrs[key] = numpy.string_(value)
-            else:
-                dataset_hdf5.parent.attrs[key] = value
+            self.timestamp_key = timestamp_key
 
     def update_column_map(self):
         """
@@ -401,15 +239,6 @@ class TimeSeries(object):
             self.dataset[t_index, c_index] = value
         return True
 
-    def missing_matrix(self, inverse=False):
-        """
-        Because we initialize datasets with -0.0, we can scan the sign bit of every
-        element of an array to determine how many data were never populated.  This
-        converts negative zeros to ones and all other data into zeros then count up
-        the number of missing elements in the array.
-        """
-        return tokio.connectors.hdf5.missing_values(self.dataset, inverse)
-
     def convert_to_deltas(self):
         """
         Convert a matrix of monotonically increasing rows into deltas.  Replaces
@@ -531,22 +360,3 @@ def timeseries_deltas(dataset):
                     prev_nonzero[icol] = this_element
 
     return diff_matrix
-
-def get_insert_indices(my_timestamps, existing_timestamps):
-    """
-    Given new timestamps and an existing series of timestamps, find the indices
-    overlap so that new data can be inserted into the middle of an existing
-    dataset
-    """
-    existing_timestep = existing_timestamps[1] - existing_timestamps[0]
-    my_timestep = my_timestamps[1] - my_timestamps[0]
-
-    # make sure the time delta is ok
-    if existing_timestep != my_timestep:
-        raise Exception("Existing dataset has different timestep (mine=%d, existing=%d)"
-                        % (my_timestep, existing_timestep))
-
-    my_offset = (my_timestamps[0] - existing_timestamps[0]) // existing_timestep
-    my_end = my_offset + len(my_timestamps)
-
-    return my_offset, my_end


Test patch:
diff --git a/tests/test_connectors_hdf5.py b/tests/test_connectors_hdf5.py
index e90770e..4bfcadd 100644
--- a/tests/test_connectors_hdf5.py
+++ b/tests/test_connectors_hdf5.py
@@ -5,10 +5,12 @@ Test the HDF5 connector
 
 import datetime
 import random
+import warnings
+import shutil
 import nose
 import numpy
 import tokiotest
-import tokio.connectors
+import tokio.connectors.hdf5
 
 DATASETS_1D = [
     'FSStepsGroup/FSStepsDataSet',
@@ -35,6 +37,21 @@ GET_SET_TRUE_VERSIONS = {
 
 INVALID_DATASET = '/abc/def'
 
+def generate_light_timeseries(file_name=tokiotest.SAMPLE_COLLECTDES_HDF5):
+    """
+    Return a TimeSeries object that's initialized against the tokiotest sample
+    input.  Uses light=True when attaching to prevent loading the entire dataset
+    into memory.
+    """
+    output_hdf5 = tokio.connectors.hdf5.Hdf5(file_name, 'r')
+    timeseries = tokio.timeseries.TimeSeries()
+
+    output_hdf5.to_timeseries(dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET, light=True)
+#   timeseries.attach(output_hdf5, dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET, light=True)
+    return timeseries
+
+
+
 def test_h5lmt():
     """connectors.hdf5.Hdf5() h5lmt support
     """
@@ -229,7 +246,12 @@ def _test_get_index(input_file):
                 target_datetime = datetime.datetime.fromtimestamp(timestamps[target_index]) \
                                   + datetime.timedelta(seconds=fuzz)
                 new_index = hdf5_file.get_index(dataset_name, target_datetime)
-                print("%d(%s) == %d(%s)? %s" % (target_index, type(target_index), new_index, type(new_index), target_index == new_index))
+                print("%d(%s) == %d(%s)? %s" % (
+                    target_index,
+                    type(target_index),
+                    new_index,
+                    type(new_index),
+                    target_index == new_index))
                 assert target_index == new_index
                 assert (dataset[new_index] == dataset[target_index]).all()
 
@@ -376,3 +398,159 @@ def test_set_versions():
 
     hdf5.close()
     test_get_versions(hdf5_filename=tokiotest.TEMP_FILE.name)
+
+@nose.tools.with_setup(tokiotest.create_tempfile, tokiotest.delete_tempfile)
+def test_commit_timeseries():
+    """connectors.hdf5.Hdf5.commit_timeseries()
+    """
+    tokiotest.TEMP_FILE.close()
+
+    # Connect to the sample input file
+    timeseries1 = tokiotest.generate_timeseries()
+
+    # Create a new HDF5 file into which timeseries1 will be copied
+    hdf5_file = tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name, 'w')
+
+    # Write the output out as a new HDF5 file
+    hdf5_file.commit_timeseries(timeseries1)
+    hdf5_file.close()
+
+    # Read that newly generated HDF5 file back in
+    timeseries2 = tokiotest.generate_timeseries(file_name=tokiotest.TEMP_FILE.name)
+
+    # Compare the original to the reprocessed
+    print("Comparing before/after read/write/read")
+    tokiotest.compare_timeseries(timeseries2, timeseries1, verbose=True)
+
+@nose.tools.with_setup(tokiotest.create_tempfile, tokiotest.delete_tempfile)
+def test_commit_timeseries_bad_bounds():
+    """connectors.hdf5.Hdf5.commit_timeseries() with out-of-bounds
+    """
+    tokiotest.TEMP_FILE.close()
+
+    # Connect to the sample input file
+    timeseries1 = tokiotest.generate_timeseries(dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET)
+    timeseries2 = tokiotest.generate_timeseries(dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET2)
+
+    # Write the output as a new HDF5 file (should work like normal)
+    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name, 'w') as hdf5_file:
+        hdf5_file.commit_timeseries(timeseries1)
+
+    # Now trim down the dataset so it no longer spans the same range as the
+    # existing HDF5 file - this should work, because the HDF5 will simply retain
+    # its start/end since the new data being committed is a complete subset
+    print("Attempting trimmed down dataset")
+    timeseries1.trim_rows(3)
+    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name) as hdf5_file:
+        hdf5_file.commit_timeseries(timeseries1)
+
+    # Add back the rows we took off, and then some - this should NOT work
+    # because the data now goes beyond the original maximum timestamp for the
+    # file
+    print("Attempting bloated existing dataset")
+    timeseries1.add_rows(12)
+    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name) as hdf5_file:
+        print('Global start: %s' % hdf5_file.attrs.get('start'))
+        print('Global end:   %s' % hdf5_file.attrs.get('end'))
+        caught = False
+        try:
+            hdf5_file.commit_timeseries(timeseries1)
+        except IndexError:
+            caught = True
+        assert caught
+
+    # Now commit a completely new dataset that doesn't fit - this should throw
+    # a warning (or should we make it throw an exception?)
+    print("Attempting bloated non-existent dataset")
+    timeseries2.add_rows(12)
+    timeseries2.dataset_name = '/blah/blah'
+    timeseries2.timestamp_key = '/blah/timestamps'
+    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name) as hdf5_file:
+        print('Global start: %s' % hdf5_file.attrs.get('start'))
+        print('Global end:   %s' % hdf5_file.attrs.get('end'))
+        caught = False
+        try:
+            hdf5_file.commit_timeseries(timeseries2)
+        except IndexError:
+            caught = True
+#       with warnings.catch_warnings(record=True) as warn:
+#           caught = True
+#           # cause all warnings to always be triggered
+#           warnings.simplefilter("always")
+#           hdf5_file.commit_timeseries(timeseries2)
+#           print len(warn)
+#           assert len(warn) == 1
+#           print warn[-1].category, type(warn[-1].category)
+#           assert issubclass(warn[-1].category, UserWarning)
+#           assert "some warning message" in str(warn[-1].message)
+        assert caught
+
+@nose.tools.with_setup(tokiotest.create_tempfile, tokiotest.delete_tempfile)
+def test_uneven_columns():
+    """connectors.hdf5.Hdf5.to_timeseries(with uneven columns)
+    """
+
+    # We expect to trigger some warnings here
+    warnings.filterwarnings('ignore')
+
+    tokiotest.TEMP_FILE.close()
+    # copy the input file into a new temporary file with which we can tinker
+    with open(tokiotest.SAMPLE_COLLECTDES_HDF5, 'rb') as input_file:
+        with open(tokiotest.TEMP_FILE.name, 'wb') as output_file:
+            shutil.copyfileobj(input_file, output_file)
+
+    print("Ensure that the input dataset has even column lengths before we make them uneven")
+    h5_file = tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name, 'r+')
+    dataset = h5_file[tokiotest.SAMPLE_COLLECTDES_DSET]
+    orig_col_names = list(h5_file.get_columns(dataset.name))
+    result = len(orig_col_names) == dataset.shape[1]
+    print(orig_col_names)
+    print(dataset.attrs['columns'])
+    print("%-3d == %3d? %s" % (len(orig_col_names), dataset.shape[1], result))
+    assert result
+
+    print("Test case where there are more column names than shape of dataset")
+    extra_columns = orig_col_names + ['argle', 'bargle', 'fffff']
+    timeseries = tokiotest.generate_timeseries(file_name=tokiotest.TEMP_FILE.name)
+    timeseries.set_columns(extra_columns)
+    result = len(timeseries.columns) == timeseries.dataset.shape[1]
+    print("%-3d == %3d? %s" % (len(timeseries.columns), timeseries.dataset.shape[1], result))
+    assert result
+    for icol in range(timeseries.dataset.shape[1]):
+        print("%-15s(%-15s) == %15s(%-15s)? %s" % (
+            extra_columns[icol],
+            type(extra_columns[icol]),
+            timeseries.columns[icol],
+            type(timeseries.columns[icol]),
+            extra_columns[icol] == timeseries.columns[icol]))
+        assert extra_columns[icol] == timeseries.columns[icol]
+
+    print("Test cases where column names are incomplete compared to shape of dataset")
+    fewer_columns = orig_col_names[0:-(3*len(orig_col_names)//4)]
+    timeseries = tokiotest.generate_timeseries(file_name=tokiotest.TEMP_FILE.name)
+    timeseries.set_columns(fewer_columns)
+    result = len(timeseries.columns) < timeseries.dataset.shape[1]
+    print("%-3d < %3d? %s" % (len(timeseries.columns), timeseries.dataset.shape[1], result))
+    assert result
+    for icol, fewer_col in enumerate(fewer_columns):
+        print("%-15s == %15s? %s" % (fewer_col, timeseries.columns[icol],
+                                     fewer_col == timeseries.columns[icol]))
+        assert fewer_col == timeseries.columns[icol]
+
+def test_light_attach():
+    """connectors.hdf5.Hdf5.attach_dataset(light=True)
+    """
+    tokiotest.TEMP_FILE.close()
+
+    full = tokiotest.generate_timeseries()
+    light = generate_light_timeseries()
+    tokiotest.compare_timeseries(light, full, verbose=True)
+
+def test_get_insert_pos():
+    """connectors.hdf5.get_insert_pos()
+    """
+    # TODO
+    # 1. ensure correctness of turning timestamp into an index
+    # 2. ensure that create_col=False works
+    # 3. ensure that create_col=True works
+    raise nose.SkipTest("test not implemented")
diff --git a/tests/test_timeseries.py b/tests/test_timeseries.py
index ed86a4c..51991a4 100644
--- a/tests/test_timeseries.py
+++ b/tests/test_timeseries.py
@@ -4,50 +4,9 @@ tokio.timeseries.TimeSeries methods
 """
 
 import random
-import shutil
-import warnings
-import nose
 import numpy
 import tokio
-from tokio.connectors.hdf5 import Hdf5
-import tokiotest
-
-def compare_timeseries(timeseries1, timeseries2, verbose=False):
-    """
-    Compare two TimeSeries objects' datasets column by column
-    """
-    for timeseries1_index, column in enumerate(list(timeseries1.columns)):
-        timeseries2_index = timeseries2.column_map[column]
-        if verbose:
-            col_sum1 = timeseries1.dataset[:, timeseries1_index].sum()
-            col_sum2 = timeseries2.dataset[:, timeseries2_index].sum()
-            print("%-14s: %.16e vs. %.16e" % (column, col_sum1, col_sum2))
-        assert numpy.array_equal(timeseries1.dataset[:, timeseries1_index],
-                                 timeseries2.dataset[:, timeseries2_index])
-
-def generate_timeseries(file_name=tokiotest.SAMPLE_COLLECTDES_HDF5,
-                        dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET):
-    """
-    Return a TimeSeries object that's initialized against the tokiotest sample
-    input
-    """
-    output_hdf5 = Hdf5(file_name, 'r')
-    print("Creating timeseries from %s" % file_name)
-    timeseries = tokio.timeseries.TimeSeries(dataset_name=dataset_name,
-                                             hdf5_file=output_hdf5)
-    return timeseries
-
-def generate_light_timeseries(file_name=tokiotest.SAMPLE_COLLECTDES_HDF5):
-    """
-    Return a TimeSeries object that's initialized against the tokiotest sample
-    input.  Uses light=True when attaching to prevent loading the entire dataset
-    into memory.
-    """
-    output_hdf5 = Hdf5(file_name, 'r')
-    timeseries = tokio.timeseries.TimeSeries()
-
-    timeseries.attach(output_hdf5, dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET, light=True)
-    return timeseries
+from tokiotest import generate_timeseries, compare_timeseries
 
 def test_rearrange():
     """
@@ -229,95 +188,6 @@ def test_timeseries_deltas():
     print()
     assert (close_matrix | fix_matrix).all()
 
-
-@nose.tools.with_setup(tokiotest.create_tempfile, tokiotest.delete_tempfile)
-def test_commit_dataset():
-    """
-    TimeSeries.commit_dataset()
-    """
-    tokiotest.TEMP_FILE.close()
-
-    # Connect to the sample input file
-    timeseries1 = generate_timeseries()
-
-    # Create a new HDF5 file into which timeseries1 will be copied
-    hdf5_file = tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name, 'w')
-
-    # Write the output out as a new HDF5 file
-    timeseries1.commit_dataset(hdf5_file)
-    hdf5_file.close()
-
-    # Read that newly generated HDF5 file back in
-    timeseries2 = generate_timeseries(file_name=tokiotest.TEMP_FILE.name)
-
-    # Compare the original to the reprocessed
-    print("Comparing before/after read/write/read")
-    compare_timeseries(timeseries2, timeseries1, verbose=True)
-
-@nose.tools.with_setup(tokiotest.create_tempfile, tokiotest.delete_tempfile)
-def test_commit_dataset_bad_bounds():
-    """
-    TimeSeries.commit_dataset() with out-of-bounds
-    """
-    tokiotest.TEMP_FILE.close()
-
-    # Connect to the sample input file
-    timeseries1 = generate_timeseries(dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET)
-    timeseries2 = generate_timeseries(dataset_name=tokiotest.SAMPLE_COLLECTDES_DSET2)
-
-    # Write the output as a new HDF5 file (should work like normal)
-    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name, 'w') as hdf5_file:
-        timeseries1.commit_dataset(hdf5_file)
-
-    # Now trim down the dataset so it no longer spans the same range as the
-    # existing HDF5 file - this should work, because the HDF5 will simply retain
-    # its start/end since the new data being committed is a complete subset
-    print("Attempting trimmed down dataset")
-    timeseries1.trim_rows(3)
-    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name) as hdf5_file:
-        timeseries1.commit_dataset(hdf5_file)
-
-    # Add back the rows we took off, and then some - this should NOT work
-    # because the data now goes beyond the original maximum timestamp for the
-    # file
-    print("Attempting bloated existing dataset")
-    timeseries1.add_rows(12)
-    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name) as hdf5_file:
-        print('Global start: %s' % hdf5_file.attrs.get('start'))
-        print('Global end:   %s' % hdf5_file.attrs.get('end'))
-        caught = False
-        try:
-            timeseries1.commit_dataset(hdf5_file)
-        except IndexError:
-            caught = True
-        assert caught
-
-    # Now commit a completely new dataset that doesn't fit - this should throw
-    # a warning (or should we make it throw an exception?)
-    print("Attempting bloated non-existent dataset")
-    timeseries2.add_rows(12)
-    timeseries2.dataset_name = '/blah/blah'
-    timeseries2.timestamp_key = '/blah/timestamps'
-    with tokio.connectors.hdf5.Hdf5(tokiotest.TEMP_FILE.name) as hdf5_file:
-        print('Global start: %s' % hdf5_file.attrs.get('start'))
-        print('Global end:   %s' % hdf5_file.attrs.get('end'))
-        caught = False
-        try:
-            timeseries2.commit_dataset(hdf5_file)
-        except IndexError:
-            caught = True
-#       with warnings.catch_warnings(record=True) as warn:
-#           caught = True
-#           # cause all warnings to always be triggered
-#           warnings.simplefilter("always")
-#           timeseries2.commit_dataset(hdf5_file)
-#           print len(warn)
-#           assert len(warn) == 1
-#           print warn[-1].category, type(warn[-1].category)
-#           assert issubclass(warn[-1].category, UserWarning)
-#           assert "some warning message" in str(warn[-1].message)
-        assert caught
-
 def test_add_rows():
     """
     TimeSeries.add_rows()
@@ -340,75 +210,3 @@ def test_add_rows():
     assert prev_deltim > 0
     assert timeseries.timestamps.shape[0] == timeseries.dataset.shape[0]
     assert (timeseries.timestamps.shape[0] - add_rows) == orig_row_count
-
-@nose.tools.with_setup(tokiotest.create_tempfile, tokiotest.delete_tempfile)
-def test_uneven_columns():
-    """
-    TimeSeries.attach_dataset with uneven columns
-    """
-
-    # We expect to trigger some warnings here
-    warnings.filterwarnings('ignore')
-
-    tokiotest.TEMP_FILE.close()
-    # copy the input file into a new temporary file with which we can tinker
-    with open(tokiotest.SAMPLE_COLLECTDES_HDF5, 'rb') as input_file:
-        with open(tokiotest.TEMP_FILE.name, 'wb') as output_file:
-            shutil.copyfileobj(input_file, output_file)
-
-    print("Ensure that the input dataset has even column lengths before we make them uneven")
-    h5_file = Hdf5(tokiotest.TEMP_FILE.name, 'r+')
-    dataset = h5_file[tokiotest.SAMPLE_COLLECTDES_DSET]
-    orig_col_names = list(h5_file.get_columns(dataset.name))
-    result = len(orig_col_names) == dataset.shape[1]
-    print(orig_col_names)
-    print(dataset.attrs['columns'])
-    print("%-3d == %3d? %s" % (len(orig_col_names), dataset.shape[1], result))
-    assert result
-
-    print("Test case where there are more column names than shape of dataset")
-    extra_columns = orig_col_names + ['argle', 'bargle', 'fffff']
-    timeseries = generate_timeseries(file_name=tokiotest.TEMP_FILE.name)
-    timeseries.set_columns(extra_columns)
-    result = len(timeseries.columns) == timeseries.dataset.shape[1]
-    print("%-3d == %3d? %s" % (len(timeseries.columns), timeseries.dataset.shape[1], result))
-    assert result
-    for icol in range(timeseries.dataset.shape[1]):
-        print("%-15s(%-15s) == %15s(%-15s)? %s" % (
-            extra_columns[icol],
-            type(extra_columns[icol]),
-            timeseries.columns[icol],
-            type(timeseries.columns[icol]),
-            extra_columns[icol] == timeseries.columns[icol]))
-        assert extra_columns[icol] == timeseries.columns[icol]
-
-    print("Test cases where column names are incomplete compared to shape of dataset")
-    fewer_columns = orig_col_names[0:-(3*len(orig_col_names)//4)]
-    timeseries = generate_timeseries(file_name=tokiotest.TEMP_FILE.name)
-    timeseries.set_columns(fewer_columns)
-    result = len(timeseries.columns) < timeseries.dataset.shape[1]
-    print("%-3d < %3d? %s" % (len(timeseries.columns), timeseries.dataset.shape[1], result))
-    assert result
-    for icol, fewer_col in enumerate(fewer_columns):
-        print("%-15s == %15s? %s" % (fewer_col, timeseries.columns[icol],
-                                     fewer_col == timeseries.columns[icol]))
-        assert fewer_col == timeseries.columns[icol]
-
-def test_light_attach():
-    """
-    TimeSeries.attach_dataset with light attach
-    """
-    tokiotest.TEMP_FILE.close()
-
-    full = generate_timeseries()
-    light = generate_light_timeseries()
-    compare_timeseries(light, full, verbose=True)
-
-def test_get_insert_pos():
-    """TimeSeries.get_insert_pos()
-    """
-    # TODO
-    # 1. ensure correctness of turning timestamp into an index
-    # 2. ensure that create_col=False works
-    # 3. ensure that create_col=True works
-    raise nose.SkipTest("test not implemented")
diff --git a/tests/tokiotest.py b/tests/tokiotest.py
index ca523b7..9eb4845 100644
--- a/tests/tokiotest.py
+++ b/tests/tokiotest.py
@@ -10,6 +10,7 @@ import errno
 import tempfile
 import subprocess
 import datetime
+import numpy # for compare_timeseries
 
 try:
     import StringIO as io
@@ -332,3 +333,32 @@ def try_unlink(output_filename):
     if os.path.exists(output_filename):
         print("Destroying %s" % output_filename)
         os.unlink(output_filename)
+
+# TimeSeries and HDF5 tests have some mutual dependency, so factor out those
+# test components here
+
+def compare_timeseries(timeseries1, timeseries2, verbose=False):
+    """
+    Compare two TimeSeries objects' datasets column by column
+    """
+    for timeseries1_index, column in enumerate(list(timeseries1.columns)):
+        timeseries2_index = timeseries2.column_map[column]
+        if verbose:
+            col_sum1 = timeseries1.dataset[:, timeseries1_index].sum()
+            col_sum2 = timeseries2.dataset[:, timeseries2_index].sum()
+            print("%-14s: %.16e vs. %.16e" % (column, col_sum1, col_sum2))
+        assert numpy.array_equal(timeseries1.dataset[:, timeseries1_index],
+                                 timeseries2.dataset[:, timeseries2_index])
+
+def generate_timeseries(file_name=SAMPLE_COLLECTDES_HDF5,
+                        dataset_name=SAMPLE_COLLECTDES_DSET):
+    """
+    Return a TimeSeries object that's initialized against the sample input
+    """
+    output_hdf5 = tokio.connectors.hdf5.Hdf5(file_name, 'r')
+    print("Creating timeseries from %s" % file_name)
+    timeseries = output_hdf5.to_timeseries(dataset_name=dataset_name)
+
+    return timeseries
+
+