#! /usr/bin/env python __author__ = "Ben Raanan" __copyright__ = "Copyright 2021, MBARI" __credits__ = ["MBARI"] __license__ = "GPL-3.0" __maintainer__ = "Ben Raanan" __email__ = "byraanan at mbari.org" __doc__ = ''' Defines the export logs class for converting the LCM binary data logged by the LRAUV backseat helper service to HDF5 and NetCDF4 data formats. @author: __author__ @license: __license__ ''' import os import datetime import h5py import netCDF4 import numpy as np import supervisor.Logger logger, _ = supervisor.Logger.configure_logger(name='backseat', log_to_file=False) from LCM.LogReader import LcmLogReader class ExportLogs(LcmLogReader): def __init__(self, ): LcmLogReader.__init__(self) self.logs = [] def find_logs(self, directory='./logs', log_name='lcmlog'): """ Walk the dir tree and locate log files. :param directory: root search dir :param log_name: log name to match :return: updates the 'logs' class member """ log_name = log_name.lower() for dir_path, _, files in os.walk(directory): fold_logs = [] for name in files: if log_name and name.lower().startswith(log_name): fold_logs.append(os.path.join(dir_path, name)) if fold_logs: self.logs.append(fold_logs) def export(self, log_dir='./logs', distribute_to='', export_format=('netcdf', 'hdf5')): """ Export lcm logs nested under log_dir to specified destination :param log_dir: :param distribute_to: :param export_format: :return: """ self.find_logs(log_dir) for log in self.logs: self.export_log(log, distribute_to, export_to=export_format) self.clear() def export_log(self, log_names: str, distribute_to: str, export_to=('netcdf', 'hdf5'), out_prefix='backseat'): """ :param log_names: :param distribute_to: :param export_to: :param out_prefix: :return: """ if not log_names: # nothing to do here... return # match output filename to the log name (appropriate file extension will be added later) destination_log = "" log = os.path.dirname(log_names[0]) if distribute_to: # export the lcm log to overlapping LRAUV log folder destination_log = self.match_log(os.path.basename(log), distribute_to) if not destination_log: # export the lcm log origin folder destination_log = log outfile = os.path.join(destination_log, out_prefix + '_' + os.path.basename(log)) # Read a lcm log(s) into memory for log in log_names: self.read_log(log) # export the data if 'netcdf' in export_to: self.write_netcdf(outfile) if 'hdf5' in export_to: self.write_hdf5(outfile) def write_netcdf(self, outfile: str, extension='.nc4'): """ Export data to netCDF4 file. :param outfile: output file name :param extension: file name suffix :return: void """ logger.info('Writing data in NETCDF4 format to: {}'.format(outfile + extension)) with netCDF4.Dataset(outfile + extension, 'w') as nc: for channel in self.data.keys(): logger.info('Processing data from LCM channel: {}'.format(channel)) chan_grp = nc.createGroup(channel) for k, v in self.data[channel].items(): logger.info('Adding {}::{}'.format(channel, k)) t_ = np.asarray(v['time']) v_ = np.asarray(v['value']) # dimensions dim = k + "_time", chan_grp.createDimension(dim[0], None) time_ = chan_grp.createVariable(dim[0], t_.dtype.str, dim) time_.unit = 'seconds since 1970-00-00T00:00:00.0Z' dimensions = [] for i, d in enumerate(np.shape(v_)[1:], start=1): # add additional dimensions for matrices, etc. new_dim = (k + "_dim" + str(i),) dim = dim + new_dim # create a new dimension... logger.info('{}::{}: adding dim {} = {}'.format(channel, k, new_dim[0], d)) chan_grp.createDimension(new_dim[0], d) dimension = chan_grp.createVariable(new_dim[0], 'i', new_dim) dimensions.append(dimension) # variables if v['unit'] == 'str': # handle StringVector case... v_ = np.asarray(v['value'], dtype='object') val_ = chan_grp.createVariable(k, str, dim) else: val_ = chan_grp.createVariable(k, v_.dtype.str, dim) val_.units = v['unit'] # data time_[:] = t_ val_[:] = v_ logger.info('Done.\n') def write_hdf5(self, outfile, extension='.h5'): """ Export data to HDF5 file. :param outfile: str containing output file name :param extension: str containing output file extension :return: void """ logger.info('Writing data in HDF5 format to: {}'.format(outfile + extension)) with h5py.File(outfile + extension, 'w') as hf: for channel in self.data.keys(): logger.info('Processing data from LCM channel: {}'.format(channel)) for k, v in self.data[channel].items(): var_name = '{}/{}'.format(channel, k) logger.info('Adding {}'.format(var_name)) hf['{}/time'.format(var_name)] = v['time'] hf['{}/units'.format(var_name)] = v['unit'] try: hf['{}/value'.format(var_name)] = v['value'] except TypeError: # handle StringVector case... hf['{}/value'.format(var_name)] = np.array(v['value'], dtype=h5py.special_dtype(vlen=str)) logger.info('Done.\n') def match_log(self, log, destination_root): """ Match lcm log to overlapping LRAUV log. Note: assumes the log folder names are log start times in the format '%Y%m%dT%H%M%S' :param log: string containing the lcm-log file path :param destination_root: string containing the root path of lrauv logs :return: path to lrauv log that overlaps the lcm-log """ # list subdirectories in destination folder _, logs, _ = next(os.walk(destination_root)) logs.sort() # compute the time difference between the lcm-log and destination log log_t = self.date(log) log_delta_t = [(log_t - self.date(l)).total_seconds() for l in logs] # index the closest preceding log... idx = min(dt for dt in log_delta_t if dt and dt >= 0) # ...and get its name log_match = logs[log_delta_t.index(idx)] log_path = "" if log_match: log_path = os.path.join(destination_root, log_match) return log_path @staticmethod def date(log_name: str, date_format='%Y%m%dT%H%M%S'): try: return datetime.datetime.strptime(log_name, date_format) except ValueError: logger.error('Handling ValueError for: {}'.format(log_name)) return datetime.datetime.max if __name__ == "__main__": import argparse # parse command-line arguments parser = argparse.ArgumentParser(description='Export lcm logs to netCDF4/HDF5.') parser.add_argument("-p", "--path", default='./logs', type=str, help="set path to logs root folder") parser.add_argument("-f", "--format", default=['netcdf', 'hdf5'], nargs='+', help="export file format") parser.add_argument("-d", "--distribute", default='', type=str, help="export logs to matching lrauv log folder found under this path") args = parser.parse_args() export_data = ExportLogs() export_data.export(log_dir=args.path, distribute_to=args.distribute, export_format=tuple(args.format))