Commit be84a2a9 authored by Vincent POLLET's avatar Vincent POLLET
Browse files

[DOC] Documentation of StreamFile class

parent 50226420
Pipeline #42784 passed with stage
in 14 minutes and 6 seconds
......@@ -2,16 +2,60 @@ import json
import numpy as np
from bob.io.base import HDF5File
from bob.ip.stereo import load_camera_config
from .config import load_data_config
class StreamFile:
"""File class to read from HDF5 streams files.
Exposes methods to read a stream's data and meta-data.
Attributes
----------
hdf5_file : :py:class:`bob.io.base.HDF5File`
HDF5 file containing the streams data.
data_format_config : str
Path to configuration json with the streams data meta-data (names, shape, etc...)
camera_config : str
Path to json file with the camera information (camera_matrix, distortion coefficients, etc...)
during streams recording
"""
def __init__(self, hdf5_file=None, data_format_config_file_path=None, camera_config_file_path=None, mode="r"):
"""Open the HDF5 and register the data and camera configuration via :meth:`bob.io.stream.StreamFile.set_source`.
"""
self.hdf5_file = None
self.data_format_config = None
self.camera_config = None
self.set_source(hdf5_file, data_format_config_file_path, camera_config_file_path, mode)
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
if isinstance(self.hdf5_file, HDF5File):
self.hdf5_file.close()
def set_source(self, hdf5_file=None, data_format_config_file_path=None, camera_config_file_path=None, mode="r"):
"""Open the HDF5 file and load data and camera config.
Parameters
----------
hdf5_file : :obj:`bob.io.base.HDF5File` or str or None
File handle or path to the streams HDF5 File, by default None.
data_format_config_file_path : str or None
Path to the data config file, by default None.
camera_config_file_path : str or None
Path to the camera config file, by default None.
mode : str
File opening mode, by default "r".
"""
if isinstance(hdf5_file, str): # case string: it is a path, use bob.io.base.HDF5File
self.hdf5_file = HDF5File(hdf5_file, mode)
elif hdf5_file is not None: # otherwise expect an opened file object (h5py.File, etc...)
elif hdf5_file is not None: # otherwise expect an opened file object (bob.io.base.HDF5File, h5py.File, etc...)
self.hdf5_file = hdf5_file
else:
self.hdf5_file = None
......@@ -24,44 +68,49 @@ class StreamFile:
else:
self.camera_config = None
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
if isinstance(self.hdf5_file, HDF5File):
self.hdf5_file.close()
# set source
def set_source(
self, hdf5_file=None, data_format_config_file_path=None, camera_config_file_path=None, mode="r"
):
self.__init__(
hdf5_file=hdf5_file,
data_format_config_file_path=data_format_config_file_path,
camera_config_file_path=camera_config_file_path,
mode=mode,
)
# get available streams
def get_available_streams(self):
""":obj:`list` of :obj:`str`: Get the names of the streams in the HDF5 File."""
if self.data_format_config is not None:
return list(self.data_format_config.keys())
else:
# TODO list available datasets if no config present
return None
# get stream config
def get_stream_config(self, stream_name):
"""Get the `stream_name` configuration: stream name, data format, etc...
Parameters
----------
stream_name : str
Name of the stream in the HDF5 File which meta-data is requested.
Returns
-------
dict
Stream meta-data. If the configuration is not available, return a default config contaning only the stream
name.
"""
if self.data_format_config is not None:
data_config = self.data_format_config[stream_name]
else:
# return a generic config if no config is present
data_config = { 'path' : stream_name}
data_config = {"path": stream_name}
return data_config
# get stream shape
def get_stream_shape(self, stream_name):
"""Get the shape of the data in in `stream_name`.
Parameters
----------
stream_name : str
Name of the stream which shape is requested.
Returns
-------
:obj:`tuple` of :obj:`int`
Shape of the `stream_name`'s data.
"""
data_config = self.get_stream_config(stream_name)
data_path = data_config["path"]
descriptor = self.hdf5_file.describe(data_path)
......@@ -69,8 +118,19 @@ class StreamFile:
shape = descriptor[1][0][1]
return shape
# get stream timestamps
def get_stream_timestamps(self, stream_name):
"""Return the timestamps of each frame in `stream_name`.
Parameters
----------
stream_name : str
Name of the stream which timestamps are requested.
Returns
-------
:obj:`numpy.ndarray`
Timestamps of each frame in `stream_name`
"""
data_config = self.get_stream_config(stream_name)
data_path = data_config["path"]
if not self.hdf5_file.has_attribute("timestamps", data_path):
......@@ -81,12 +141,31 @@ class StreamFile:
if isinstance(timestamps, bytes):
timestamps = timestamps.decode("utf-8")
if isinstance(timestamps, str):
return np.array(json.loads("[" + timestamps.strip().strip("[").strip("]") + "]"))
else:
return timestamps
timestamps = np.array(json.loads("[" + timestamps.strip().strip("[").strip("]") + "]"))
return timestamps
# get stream camera
def get_stream_camera(self, stream_name):
"""Get the camera config of a stream.
Returns None if no camera is associated with the data in the stream.
Parameters
----------
stream_name : str
Name of the stream which camera configuration is requested.
Returns
-------
dict or None
Stream's camera config.
Raises
------
ValueError
If the stream's camera exists, but it's configuration is missing from the stream's data config (likely a
name missmatch).
"""
data_config = self.get_stream_config(stream_name)
if "use_config_from" in data_config:
data_config = self.get_stream_config(data_config["use_config_from"])
......@@ -95,24 +174,45 @@ class StreamFile:
if camera_name in self.camera_config:
return self.camera_config[camera_name]
else:
raise('invalid camera name')
raise ValueError("invalid camera name")
else:
return None
# load stream data
def load_stream_data(self, stream_name, index):
"""Load the `index` frame(s) of data from `stream_name`.
Loads only the requested indices from the file.
If the stream's data configuration requests it, some axis in the loaded data are flipped.
Parameters
----------
stream_name : str
Name of the stream which data should to be loaded
index : int or :obj:`list` of :obj:`int`
Index of the frame(s) to load.
Returns
-------
:obj:`numpy.ndarray`
Stream's data at frames index.
Raises
------
ValueError
If `index` has not a valid type.
"""
data_config = self.get_stream_config(stream_name)
data_path = data_config["path"]
if "use_config_from" in data_config:
data_config = self.get_stream_config(data_config["use_config_from"])
# load only relevant data
# load only relevant data using lread.
if isinstance(index, int):
data = np.stack([self.hdf5_file.lread(data_path, index)])
elif isinstance(index, list):
data = np.stack([self.hdf5_file.lread(data_path, i) for i in index])
else:
raise Exception("index can only be int or list")
raise ValueError("index can only be int or list")
# flip if requested
array_flip = None
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment