sddscombinelogfiles.c File Reference

Detailed Description

Combines multiple SDDS log files into a single file, retaining only common timestamps.

This program processes SDDS log files in the one-PV-per-file format, merging them into a single SDDS file. It retains only the timestamps common across all input files and supports flexible input-output configurations such as pipe-based I/O and overwriting existing output files.

Usage

sddscombinelogfiles [<SDDSinputfilelist>] [<SDDSoutputfile>] 
                    [-pipe=[output]] 
                    [-overwrite]
                    [-threads=<number>]

Options

Option	Description
-pipe	Use pipe output for the SDDS file instead of writing to a file.
-overwrite	Overwrite the output file if it already exists.
-threads	Number of input files to load concurrently.

Copyright

• (c) 2002 The University of Chicago, as Operator of Argonne National Laboratory.
• (c) 2002 The Regents of the University of California, as Operator of Los Alamos National Laboratory.

License

This file is distributed under the terms of the Software License Agreement found in the file LICENSE included with this distribution.

Author

R. Soliday, L. Emery

Definition in file sddscombinelogfiles.c.

#include "mdb.h"
#include "scan.h"
#include "SDDS.h"

Go to the source code of this file.

Functions
static int	LoadLogFile (const char *filename, LOG_FILE_DATA *logFile)
static void	FreeLogFileData (LOG_FILE_DATA *logFile)
int	main (int argc, char **argv)

Function Documentation

FreeLogFileData()

static void FreeLogFileData ( LOG_FILE_DATA * logFile )

static

Definition at line 621 of file sddscombinelogfiles.c.

                                                    {
  int page;
  if (!logFile)
    return;
  for (page = 0; page < logFile->pages; page++) {
    free(logFile->page[page].timeValues);
    free(logFile->page[page].dataValues);
    free(logFile->page[page].dataName);
  }
  free(logFile->page);
  logFile->page = NULL;
  logFile->pages = 0;
}

LoadLogFile()

static int LoadLogFile ( const char * filename, LOG_FILE_DATA * logFile )

static

Definition at line 480 of file sddscombinelogfiles.c.

                                                                     {
  SDDS_DATASET SDDS_input;
  char **columnname = NULL;
  int32_t columnnames = 0;
  int dataIndex = -1;
  int32_t readCode;
  int64_t j;
  int initialized = 0;
 
  logFile->page = NULL;
  logFile->pages = 0;
  logFile->status = 0;
  logFile->error[0] = 0;
 
  if (!SDDS_InitializeInput(&SDDS_input, (char *)filename)) {
    snprintf(logFile->error, sizeof(logFile->error), "Error: Unable to open '%s'.", filename);
    return 0;
  }
  initialized = 1;
 
  columnname = SDDS_GetColumnNames(&SDDS_input, &columnnames);
  if (columnname == NULL) {
    snprintf(logFile->error, sizeof(logFile->error), "Error: Unable to read column names from '%s'.", filename);
    goto fail;
  }
 
  if (columnnames > 3 || columnnames < 2) {
    snprintf(logFile->error, sizeof(logFile->error), "Error: Unexpected number of columns in '%s'.", filename);
    goto fail;
  }
 
  if (columnnames == 2) {
    if (strcmp("Time", columnname[0]) == 0) {
      dataIndex = 1;
    } else if (strcmp("Time", columnname[1]) == 0) {
      dataIndex = 0;
    } else {
      snprintf(logFile->error, sizeof(logFile->error), "Error: 'Time' column is missing in '%s'.", filename);
      goto fail;
    }
  }
 
  if (columnnames == 3) {
    if (strcmp("CAerrors", columnname[0]) == 0) {
      if (strcmp("Time", columnname[1]) == 0) {
        dataIndex = 2;
      } else if (strcmp("Time", columnname[2]) == 0) {
        dataIndex = 1;
      } else {
        snprintf(logFile->error, sizeof(logFile->error), "Error: 'Time' column is missing in '%s'.", filename);
        goto fail;
      }
    } else if (strcmp("CAerrors", columnname[1]) == 0) {
      if (strcmp("Time", columnname[0]) == 0) {
        dataIndex = 2;
      } else if (strcmp("Time", columnname[2]) == 0) {
        dataIndex = 0;
      } else {
        snprintf(logFile->error, sizeof(logFile->error), "Error: 'Time' column is missing in '%s'.", filename);
        goto fail;
      }
    } else if (strcmp("CAerrors", columnname[2]) == 0) {
      if (strcmp("Time", columnname[0]) == 0) {
        dataIndex = 1;
      } else if (strcmp("Time", columnname[1]) == 0) {
        dataIndex = 0;
      } else {
        snprintf(logFile->error, sizeof(logFile->error), "Error: 'Time' column is missing in '%s'.", filename);
        goto fail;
      }
    } else {
      snprintf(logFile->error, sizeof(logFile->error), "Error: 'CAerrors' column is missing in '%s'.", filename);
      goto fail;
    }
  }
 
  while ((readCode = SDDS_ReadTable(&SDDS_input)) > 0) {
    int page = logFile->pages;
    LOG_PAGE_DATA *pageData;
    pageData = realloc(logFile->page, sizeof(*logFile->page) * (page + 1));
    if (!pageData) {
      snprintf(logFile->error, sizeof(logFile->error), "Error: Memory allocation failure.");
      goto fail;
    }
    logFile->page = pageData;
    logFile->page[page].timeValues = NULL;
    logFile->page[page].dataValues = NULL;
    logFile->page[page].dataName = NULL;
    logFile->page[page].rows = SDDS_RowCount(&SDDS_input);
    logFile->pages++;
 
    if (!SDDS_CopyString(&logFile->page[page].dataName, columnname[dataIndex])) {
      snprintf(logFile->error, sizeof(logFile->error), "Error: Memory allocation failure.");
      goto fail;
    }
    if (logFile->page[page].rows > 0) {
      logFile->page[page].timeValues = SDDS_GetColumnInDoubles(&SDDS_input, "Time");
      if (logFile->page[page].timeValues == NULL) {
        snprintf(logFile->error, sizeof(logFile->error), "Error: Unable to read 'Time' data from '%s'.", filename);
        goto fail;
      }
 
      logFile->page[page].dataValues = SDDS_GetColumnInDoubles(&SDDS_input, columnname[dataIndex]);
      if (logFile->page[page].dataValues == NULL) {
        snprintf(logFile->error, sizeof(logFile->error), "Error: Unable to read '%s' data from '%s'.", columnname[dataIndex], filename);
        goto fail;
      }
    }
  }
  if (readCode == 0) {
    snprintf(logFile->error, sizeof(logFile->error), "Error: Unable to read data from '%s'.", filename);
    goto fail;
  }
 
  for (j = 0; j < columnnames; j++)
    free(columnname[j]);
  free(columnname);
  columnname = NULL;
 
  if (!SDDS_Terminate(&SDDS_input)) {
    initialized = 0;
    snprintf(logFile->error, sizeof(logFile->error), "Error: Unable to close '%s'.", filename);
    goto fail;
  }
  initialized = 0;
 
  logFile->status = 1;
  return 1;
 
fail:
  if (columnname) {
    for (j = 0; j < columnnames; j++)
      free(columnname[j]);
    free(columnname);
  }
  if (initialized)
    SDDS_Terminate(&SDDS_input);
  FreeLogFileData(logFile);
  return 0;
}

main()

int main	(	int	argc,
		char **	argv )

Definition at line 85 of file sddscombinelogfiles.c.

                                {
  SDDS_DATASET SDDS_output;
  SCANNED_ARG *s_arg;
  KEYED_EQUIVALENT **keyGroup = NULL;
  long keyGroups = 0;
  char **inputfile = NULL;
  int inputfiles = 0;
  char *outputfile = NULL;
  int i_arg, n, row, z;
  int threads = 1;
  int64_t i, j, m, r, s;
  unsigned long pipeFlags = 0;
  int overwrite = 0;
  double **timeValues = NULL;
  double **dataValues = NULL;
  short **flag = NULL;
  int64_t *rows = NULL;
  char **dataNames = NULL;
  char **uniqueDataName = NULL;
  int uniqueDataNames = 0;
  int page = 0;
  int pages;
  int found;
  double *outputTimeValues = NULL;
  double **outputDataValues = NULL;
  int64_t allocated_rows = 0;
  int **array = NULL;
  int *arrayCount;
  LOG_FILE_DATA *logFile = NULL;
 
  SDDS_RegisterProgramName(argv[0]);
  argc = scanargs(&s_arg, argc, argv);
 
  if (argc < 3) {
    fprintf(stderr, "%s", USAGE);
    return EXIT_FAILURE;
  }
 
  for (i_arg = 1; i_arg < argc; i_arg++) {
    if (s_arg[i_arg].arg_type == OPTION) {
      switch (match_string(s_arg[i_arg].list[0], option, N_OPTIONS, 0)) {
      case SET_OVERWRITE:
        overwrite = 1;
        break;
      case SET_THREADS:
        if (s_arg[i_arg].n_items != 2 ||
            sscanf(s_arg[i_arg].list[1], "%d", &threads) != 1 ||
            threads < 1) {
          fprintf(stderr, "Error: Invalid -threads option syntax.\n");
          return EXIT_FAILURE;
        }
        break;
      case SET_PIPE:
        if (!processPipeOption(s_arg[i_arg].list + 1,
                               s_arg[i_arg].n_items - 1,
                               &pipeFlags)) {
          fprintf(stderr, "Error: Invalid -pipe option syntax.\n");
          return EXIT_FAILURE;
        }
        if (pipeFlags & USE_STDIN) {
          fprintf(stderr, "Error: -pipe=in is not supported.\n");
          return EXIT_FAILURE;
        }
        break;
      default:
        fprintf(stderr, "Error: Unrecognized option.\n%s", USAGE);
        return EXIT_FAILURE;
      }
    } else {
      inputfile = trealloc(inputfile, sizeof(*inputfile) * (inputfiles + 1));
      inputfile[inputfiles++] = s_arg[i_arg].list[0];
    }
  }
 
  if (inputfiles > 1) {
    if (!(pipeFlags & USE_STDOUT)) {
      outputfile = inputfile[--inputfiles];
      if (fexists(outputfile) && !overwrite) {
        fprintf(stderr, "Error: Output file '%s' already exists. Use -overwrite to replace it.\n", outputfile);
        return EXIT_FAILURE;
      }
    }
  } else if (inputfiles == 1) {
    if ((pipeFlags & USE_STDOUT) && outputfile) {
      fprintf(stderr, "Error: Too many filenames provided with -pipe=output.\n");
      return EXIT_FAILURE;
    }
  } else {
    fprintf(stderr, "Error: No input filenames provided.\n%s", USAGE);
    return EXIT_FAILURE;
  }
 
  if (threads > inputfiles)
    threads = inputfiles;
 
  logFile = calloc(inputfiles, sizeof(*logFile));
  if (!logFile) {
    fprintf(stderr, "Error: Memory allocation failure.\n");
    return EXIT_FAILURE;
  }
 
#pragma omp parallel for if (threads > 1 && inputfiles > 1) num_threads(threads) schedule(dynamic)
  for (i = 0; i < inputfiles; i++) {
    LoadLogFile(inputfile[i], &logFile[i]);
  }
 
  pages = 0;
  for (i = 0; i < inputfiles; i++) {
    if (!logFile[i].status) {
      fprintf(stderr, "%s\n", logFile[i].error);
      for (j = 0; j < inputfiles; j++)
        FreeLogFileData(&logFile[j]);
      free(logFile);
      return EXIT_FAILURE;
    }
    pages += logFile[i].pages;
  }
 
  timeValues = malloc(sizeof(*timeValues) * pages);
  dataValues = malloc(sizeof(*dataValues) * pages);
  dataNames = malloc(sizeof(*dataNames) * pages);
  rows = malloc(sizeof(*rows) * pages);
  if (!timeValues || !dataValues || !dataNames || !rows) {
    fprintf(stderr, "Error: Memory allocation failure.\n");
    for (i = 0; i < inputfiles; i++)
      FreeLogFileData(&logFile[i]);
    free(logFile);
    return EXIT_FAILURE;
  }
 
  page = 0;
  for (i = 0; i < inputfiles; i++) {
    int filePage;
    for (filePage = 0; filePage < logFile[i].pages; filePage++) {
      timeValues[page] = logFile[i].page[filePage].timeValues;
      dataValues[page] = logFile[i].page[filePage].dataValues;
      dataNames[page] = logFile[i].page[filePage].dataName;
      rows[page] = logFile[i].page[filePage].rows;
      logFile[i].page[filePage].timeValues = NULL;
      logFile[i].page[filePage].dataValues = NULL;
      logFile[i].page[filePage].dataName = NULL;
      page++;
    }
    FreeLogFileData(&logFile[i]);
  }
  free(logFile);
 
  pages = page;
 
  /* Identify unique data names */
  for (page = 0; page < pages; page++) {
    found = 0;
    for (i = 0; i < uniqueDataNames; i++) {
      if (strcmp(dataNames[page], uniqueDataName[i]) == 0) {
        found = 1;
        break;
      }
    }
    if (!found) {
      uniqueDataName = realloc(uniqueDataName, sizeof(*uniqueDataName) * (uniqueDataNames + 1));
      SDDS_CopyString(&uniqueDataName[uniqueDataNames], dataNames[page]);
      uniqueDataNames++;
    }
  }
 
  /* Initialize output SDDS file */
  if (!SDDS_InitializeOutput(&SDDS_output, SDDS_BINARY, 0, NULL, NULL, outputfile)) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  if (!SDDS_DefineSimpleColumn(&SDDS_output, "Time", "s", SDDS_DOUBLE)) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  for (i = 0; i < uniqueDataNames; i++) {
    if (!SDDS_DefineSimpleColumn(&SDDS_output, uniqueDataName[i], NULL, SDDS_DOUBLE)) {
      SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
      return EXIT_FAILURE;
    }
  }
 
  outputDataValues = malloc(sizeof(*outputDataValues) * uniqueDataNames);
  if (uniqueDataNames == 1) {
    /* Single PV: Concatenate all data */
    for (page = 0; page < pages; page++) {
      allocated_rows += rows[page];
    }
 
    outputTimeValues = malloc(sizeof(*outputTimeValues) * allocated_rows);
    outputDataValues[0] = malloc(sizeof(*(outputDataValues[0])) * allocated_rows);
 
    i = 0;
    for (page = 0; page < pages; page++) {
      for (j = 0; j < rows[page]; j++) {
        outputTimeValues[i] = timeValues[page][j];
        outputDataValues[0][i] = dataValues[page][j];
        i++;
      }
    }
  } else {
    /* Multiple PVs: Retain only common timestamps */
    flag = malloc(sizeof(*flag) * pages);
    for (page = 0; page < pages; page++) {
      flag[page] = calloc(rows[page], sizeof(*(flag[page])));
    }
 
    array = malloc(sizeof(*array) * uniqueDataNames);
    arrayCount = calloc(uniqueDataNames, sizeof(*arrayCount));
 
    for (i = 0; i < uniqueDataNames; i++) {
      for (page = 0; page < pages; page++) {
        if (strcmp(dataNames[page], uniqueDataName[i]) == 0) {
          arrayCount[i]++;
          if (arrayCount[i] == 1) {
            array[i] = malloc(sizeof(*(array[i])));
          } else {
            array[i] = realloc(array[i], sizeof(*(array[i])) * arrayCount[i]);
          }
          array[i][arrayCount[i] - 1] = page;
        }
      }
    }
 
    for (i = 0; i < arrayCount[0]; i++) {
      keyGroup = MakeSortedKeyGroups(&keyGroups, SDDS_DOUBLE, timeValues[array[0][i]], rows[array[0][i]]);
      for (n = 1; n < uniqueDataNames; n++) {
        for (m = 0; m < arrayCount[n]; m++) {
          if ((i == m) && (rows[array[0][i]] == rows[array[n][m]]) && (rows[array[0][i]] > 10)) {
            if ((timeValues[array[0][i]][0] == timeValues[array[n][m]][0]) &&
                (timeValues[array[0][i]][1] == timeValues[array[n][m]][1]) &&
                (timeValues[array[0][i]][rows[array[0][i]] - 2] == timeValues[array[n][m]][rows[array[n][m]] - 2]) &&
                (timeValues[array[0][i]][rows[array[0][i]] - 1] == timeValues[array[n][m]][rows[array[n][m]] - 1])) {
              /* Assume the entire page matches because it has the same number of rows and key timestamps match */
              for (r = 0; r < rows[array[n][m]]; r++) {
                if (flag[array[n][m]][r]) {
                  continue;
                }
                flag[array[0][i]][r] += 1;
                flag[array[n][m]][r] = 1;
              }
            }
          }
 
          for (r = 0; r < rows[array[n][m]]; r++) {
            if (flag[array[n][m]][r]) {
              continue;
            }
            row = FindMatchingKeyGroup(keyGroup, keyGroups, SDDS_DOUBLE, &(timeValues[array[n][m]][r]), 1);
            if (row >= 0) {
              flag[array[0][i]][row] += 1;
              flag[array[n][m]][r] = 1;
            }
          }
        }
      }
 
      for (j = 0; j < keyGroups; j++) {
        free(keyGroup[j]->equivalent);
        free(keyGroup[j]);
      }
      free(keyGroup);
    }
 
    z = uniqueDataNames - 1;
    for (n = 0; n < arrayCount[0]; n++) {
      for (m = 0; m < rows[array[0][n]]; m++) {
        if (flag[array[0][n]][m] >= z) {
          allocated_rows++;
        }
      }
    }
 
    outputTimeValues = malloc(sizeof(*outputTimeValues) * allocated_rows);
    for (i = 0; i < uniqueDataNames; i++) {
      outputDataValues[i] = malloc(sizeof(*(outputDataValues[i])) * allocated_rows);
    }
 
    s = 0;
    for (i = 0; i < arrayCount[0]; i++) {
      for (j = 0; j < rows[array[0][i]]; j++) {
        if (flag[array[0][i]][j] >= z) {
          outputTimeValues[s] = timeValues[array[0][i]][j];
          outputDataValues[0][s] = dataValues[array[0][i]][j];
          s++;
        }
      }
    }
 
    if (s == 0) {
      fprintf(stderr, "Error: No matching 'Time' rows found in input files.\n");
      return EXIT_FAILURE;
    }
 
    keyGroup = MakeSortedKeyGroups(&keyGroups, SDDS_DOUBLE, outputTimeValues, s);
 
    for (n = 1; n < uniqueDataNames; n++) {
      for (m = 0; m < arrayCount[n]; m++) {
        for (r = 0; r < rows[array[n][m]]; r++) {
          if (flag[array[n][m]][r]) {
            row = FindMatchingKeyGroup(keyGroup, keyGroups, SDDS_DOUBLE, &(timeValues[array[n][m]][r]), 1);
            if (row >= 0) {
              outputDataValues[n][row] = dataValues[array[n][m]][r];
            }
          }
        }
      }
    }
 
    for (i = 0; i < uniqueDataNames; i++) {
      free(array[i]);
    }
 
    for (j = 0; j < keyGroups; j++) {
      if (keyGroup[j]->equivalent)
        free(keyGroup[j]->equivalent);
      if (keyGroup[j])
        free(keyGroup[j]);
    }
 
    for (page = 0; page < pages; page++) {
      free(flag[page]);
    }
 
    free(array);
    free(keyGroup);
    free(arrayCount);
    free(flag);
  }
 
  /* Free allocated memory for input data */
  for (page = 0; page < pages; page++) {
    if (timeValues[page])
      free(timeValues[page]);
    if (dataValues[page])
      free(dataValues[page]);
    free(dataNames[page]);
  }
  free(timeValues);
  free(dataValues);
  free(dataNames);
 
  /* Write the output SDDS file */
  if (!SDDS_WriteLayout(&SDDS_output)) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  if (!SDDS_StartPage(&SDDS_output, allocated_rows)) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  if (!SDDS_SetColumnFromDoubles(&SDDS_output, SDDS_SET_BY_NAME, outputTimeValues, allocated_rows, "Time")) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  for (i = 0; i < uniqueDataNames; i++) {
    if (!SDDS_SetColumnFromDoubles(&SDDS_output, SDDS_SET_BY_NAME, outputDataValues[i], allocated_rows, uniqueDataName[i])) {
      SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
      return EXIT_FAILURE;
    }
  }
 
  if (!SDDS_WriteTable(&SDDS_output)) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  if (!SDDS_Terminate(&SDDS_output)) {
    SDDS_PrintErrors(stderr, SDDS_VERBOSE_PrintErrors);
    return EXIT_FAILURE;
  }
 
  /* Free allocated memory for output data */
  for (i = 0; i < uniqueDataNames; i++) {
    free(uniqueDataName[i]);
    free(outputDataValues[i]);
  }
  free(outputTimeValues);
  free(outputDataValues);
  free(uniqueDataName);
  free(rows);
 
  if (inputfiles > 0) {
    free(inputfile);
  }
 
  free_scanargs(&s_arg, argc);
 
  return EXIT_SUCCESS;
}