libdsproc3  2.0
 All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Macros Groups
dsproc_dataset_fetch.c
Go to the documentation of this file.
1 /*******************************************************************************
2 *
3 * COPYRIGHT (C) 2012 Battelle Memorial Institute. All Rights Reserved.
4 *
5 ********************************************************************************
6 *
7 * Author:
8 * name: Brian Ermold
9 * phone: (509) 375-2277
10 * email: brian.ermold@pnl.gov
11 *
12 ********************************************************************************
13 *
14 * REPOSITORY INFORMATION:
15 * $Revision: 56490 $
16 * $Author: ermold $
17 * $Date: 2014-09-15 19:46:07 +0000 (Mon, 15 Sep 2014) $
18 *
19 ********************************************************************************
20 *
21 * NOTE: DOXYGEN is used to generate documentation for this file.
22 *
23 *******************************************************************************/
24 
25 /** @file dsproc_dataset_fetch.c
26  * Dataset Fetch Functions.
27  */
28 
29 #include "dsproc3.h"
30 #include "dsproc_private.h"
31 
32 extern DSProc *_DSProc; /**< Internal DSProc structure */
33 
34 /** @privatesection */
35 
36 /*******************************************************************************
37  * Static Functions Visible Only To This Module
38  */
39 
40 /**
41  * Static: Get the time range of the previously fetched data.
42  *
43  * @param ds pointer to the DataStream structure
44  * @param begin output: begin time
45  * @param end output: end time
46  *
47  * @retval 1 previously fetched data exists
48  * @retval 0 previously fetched data does not exist
49  */
50 static int _dsproc_get_fetched_range(
51  DataStream *ds,
52  timeval_t *begin,
53  timeval_t *end)
54 {
55  CDSGroup *dataset;
56  CDSVar *var;
57  size_t count;
58 
59  /* Initialize variables */
60 
61  memset(begin, 0, sizeof(timeval_t));
62  memset(end, 0, sizeof(timeval_t));
63 
64  /* Check if any previously fetched data exists */
65 
66  if (!ds ||
67  !ds->fetched_cds ||
68  !ds->fetched_cds->ngroups) {
69 
70  return(0);
71  }
72 
73  /* Get start time */
74 
75  dataset = ds->fetched_cds->groups[0];
76  var = cds_find_time_var(dataset);
77  if (!var) return(0);
78 
79  count = 1;
80  cds_get_sample_timevals(var, 0, &count, begin);
81 
82  /* Get end time */
83 
84  dataset = ds->fetched_cds->groups[ds->fetched_cds->ngroups - 1];
85  var = cds_find_time_var(dataset);
86  if (!var) return(0);
87 
88  count = 1;
89  cds_get_sample_timevals(var, var->sample_count - 1, &count, end);
90 
91  return(1);
92 }
93 
94 /*******************************************************************************
95  * Private Functions Visible Only To This Library
96  */
97 
98 /**
99  * Private: Fetch previously stored data from a datastream file.
100  *
101  * If an error occurs in this function it will be appended to the log and
102  * error mail messages, and the process status will be set appropriately.
103  *
104  * @param dsfile - pointer to the DSFile structure
105  * @param start - index of the first record to retrieve
106  * @param count - number of records to retrieve
107  * @param nvars - the number of variables to retrieve
108  * (or 0 to retrieve all variables)
109  * @param var_names - the list if variable names to retrieve
110  * (or NULL to retrieve all variables)
111  * @param parent - pointer to the parent dataset to store the
112  * observation in or NULL for no parent group.
113  *
114  * @retval obs pointer to the retrieved observation
115  * @retval NULL if an error occurred
116  */
117 CDSGroup *_dsproc_fetch_dsfile_dataset(
118  DSFile *dsfile,
119  size_t start,
120  size_t count,
121  size_t nvars,
122  const char **var_names,
123  CDSGroup *parent)
124 {
125  size_t tmp_count;
126  CDSGroup *dataset;
127  CDSVar *var;
128  int varid;
129  int status;
130  size_t vi;
131 
132  /* Create the dataset group for this file */
133 
134  dataset = cds_define_group(parent, dsfile->name);
135 
136  if (!dataset) {
137  dsproc_set_status(DSPROC_ENOMEM);
138  return((CDSGroup *)NULL);
139  }
140 
141  /* Load the data */
142 
143  if (!_dsproc_open_dsfile(dsfile, 0)) {
144  cds_delete_group(dataset);
145  return((CDSGroup *)NULL);
146  }
147 
148  if (!nvars) {
149 
150  /* Read in the NetCDF header */
151 
152  if (!ncds_read_group(dsfile->ncid, 0, dataset)) {
153 
154  ERROR( DSPROC_LIB_NAME,
155  "Could not read in netcdf header from: %s\n",
156  dsfile->full_path);
157 
158  dsproc_set_status(DSPROC_ENCREAD);
159  cds_delete_group(dataset);
160  return((CDSGroup *)NULL);
161  }
162 
163  /* Read in the NetCDF data */
164 
165  if (!ncds_read_group_data(
166  dsfile->ncid, start, count, 0, dataset, 0)) {
167 
168  ERROR( DSPROC_LIB_NAME,
169  "Could not read in netcdf data from: %s\n",
170  dsfile->full_path);
171 
172  dsproc_set_status(DSPROC_ENCREAD);
173  cds_delete_group(dataset);
174  return((CDSGroup *)NULL);
175  }
176  }
177  else {
178 
179  /* Read in all global attributes */
180 
181  status = ncds_read_atts(dsfile->ncid, dataset);
182 
183  if (status < 0) {
184 
185  ERROR( DSPROC_LIB_NAME,
186  "Could not read in global attributes from: %s\n",
187  dsfile->full_path);
188 
189  dsproc_set_status(DSPROC_ENCREAD);
190  cds_delete_group(dataset);
191  return((CDSGroup *)NULL);
192  }
193 
194  /* Read in the time variable */
195 
196  status = ncds_get_time_info(
197  dsfile->ncid, NULL, &varid, NULL, NULL, NULL, NULL);
198 
199  if (status <= 0) {
200 
201  ERROR( DSPROC_LIB_NAME,
202  "Could not get time variable id from: %s\n",
203  dsfile->full_path);
204 
205  dsproc_set_status(DSPROC_ENCREAD);
206  cds_delete_group(dataset);
207  return((CDSGroup *)NULL);
208  }
209 
210  tmp_count = count;
211 
212  var = ncds_get_var_by_id(
213  dsfile->ncid, varid, start, &tmp_count, dataset,
214  NULL, 0, NULL, 0, 0, NULL, NULL, NULL, NULL);
215 
216  if (!var) {
217 
218  ERROR( DSPROC_LIB_NAME,
219  "Could not read in time variable data from: %s\n",
220  dsfile->full_path);
221 
222  dsproc_set_status(DSPROC_ENCREAD);
223  cds_delete_group(dataset);
224  return((CDSGroup *)NULL);
225  }
226 
227  /* Read in all requested variables */
228 
229  for (vi = 0; vi < nvars; vi++) {
230 
231  tmp_count = count;
232  var = ncds_get_var(
233  dsfile->ncid, var_names[vi], start, &tmp_count, dataset,
234  NULL, 0, NULL, 0, 0, NULL, NULL, NULL, NULL);
235 
236  if (var == (CDSVar *)-1) {
237 
238  ERROR( DSPROC_LIB_NAME,
239  "Could not read in %s variable data from: %s\n",
240  var_names[vi], dsfile->full_path);
241 
242  dsproc_set_status(DSPROC_ENCREAD);
243  cds_delete_group(dataset);
244  return((CDSGroup *)NULL);
245  }
246 
247  if (var == (CDSVar *)NULL) {
248 
249  WARNING( DSPROC_LIB_NAME,
250  "Requested variable %s not found in: %s\n",
251  var_names[vi], dsfile->full_path);
252  }
253  }
254  }
255 
256  return(dataset);
257 }
258 
259 /**
260  * Private: Get the DOD for a datastream file.
261  *
262  * If an error occurs in this function it will be appended to the log and
263  * error mail messages, and the process status will be set appropriately.
264  *
265  * @param file - pointer to the DSFile structure.
266  *
267  * @return
268  * - pointer to the CDSGroup containing the DOD
269  * - NULL if an error occurred
270  */
271 CDSGroup *_dsproc_fetch_dsfile_dod(DSFile *file)
272 {
273  if (file->dod) {
274  return(file->dod);
275  }
276 
277  if (!_dsproc_open_dsfile(file, 0)) {
278  return((CDSGroup *)NULL);
279  }
280 
281  file->dod = cds_define_group(NULL, file->name);
282 
283  if (!file->dod) {
284 
285  ERROR( DSPROC_LIB_NAME,
286  "Could not get DOD for file: %s\n"
287  " -> memory allocation error",
288  file->full_path);
289 
290  dsproc_set_status(DSPROC_ENOMEM);
291  return((CDSGroup *)NULL);
292  }
293 
294  if (!ncds_read_group(file->ncid, 0, file->dod)) {
295 
296  ERROR( DSPROC_LIB_NAME,
297  "Could not get DOD for file: %s\n",
298  file->full_path);
299 
300  dsproc_set_status(DSPROC_ENCREAD);
301  return((CDSGroup *)NULL);
302  }
303 
304  if (!ncds_read_static_data(file->ncid, file->dod)) {
305 
306  ERROR( DSPROC_LIB_NAME,
307  "Could not get DOD for file: %s\n",
308  file->full_path);
309 
310  dsproc_set_status(DSPROC_ENCREAD);
311  return((CDSGroup *)NULL);
312  }
313 
314  return(file->dod);
315 }
316 
317 /**
318  * Private: Fetch previously stored datasets.
319  *
320  * This function will search the specified datastream files and retrieve
321  * all data for the specified time range. The _dsproc_find_dsfiles()
322  * function should be used to get the dsfiles list.
323  *
324  * If the begin_timeval is not specified, data for the time just prior to
325  * the end_timeval will be retrieved.
326  *
327  * If the end_timeval is not specified, data for the time just after the
328  * begin_timeval will be retrieved.
329  *
330  * If an error occurs in this function it will be appended to the log and
331  * error mail messages, and the process status will be set appropriately.
332  *
333  * @param ndsfiles - number of datastream Files
334  * @param dsfiles - list of datastream Files to search
335  * @param begin_timeval - beginning of the time range to search
336  * @param end_timeval - end of the time range to search
337  * @param nvars - the number of variables to retrieve
338  * (or 0 to retrieve all variables)
339  * @param var_names - the list if variable names to retrieve
340  * (or NULL to retrieve all variables)
341  * @param merge_obs - flag specifying if multiple observations should
342  * be merged when possible (0 == false, 1 == true)
343  * @param parent - pointer to the parent dataset to store the
344  * retrieved observations in.
345  *
346  * @retval nobs the number of observations retrieved
347  * @retval -1 if an error occurred
348  */
349 int _dsproc_fetch_dataset(
350  int ndsfiles,
351  DSFile **dsfiles,
352  timeval_t *begin_timeval,
353  timeval_t *end_timeval,
354  size_t nvars,
355  const char **var_names,
356  int merge_obs,
357  CDSGroup *parent)
358 {
359  DSFile *dsfile;
360  DSFile *prev_dsfile;
361  int prev_start;
362  int start;
363  int end;
364  size_t count;
365  int nobs;
366  int fi;
367 
368  if (!ndsfiles) {
369  return(0);
370  }
371 
372  if ((!begin_timeval || !begin_timeval->tv_sec) &&
373  (!end_timeval || !end_timeval->tv_sec)) {
374 
375  return(0);
376  }
377 
378  /* Loop over all datastream files */
379 
380  prev_dsfile = (DSFile *)NULL;
381  prev_start = 0;
382  nobs = 0;
383 
384  for (fi = 0; fi < ndsfiles; fi++) {
385 
386  dsfile = dsfiles[fi];
387 
388  if (!dsfile->ntimes) {
389  continue;
390  }
391 
392  if (!begin_timeval || !begin_timeval->tv_sec) {
393 
394  /* We want the dataset for the time just prior to the end_timeval */
395 
396  start = cds_find_timeval_index(
397  dsfile->ntimes, dsfile->timevals, *end_timeval, CDS_LT);
398 
399  if (start >= 0) {
400  prev_dsfile = dsfile;
401  prev_start = start;
402  }
403  else {
404  break;
405  }
406  }
407  else if (!end_timeval || !end_timeval->tv_sec) {
408 
409  /* We want the dataset for the time just after the begin_timeval */
410 
411  start = cds_find_timeval_index(
412  dsfile->ntimes, dsfile->timevals, *begin_timeval, CDS_GT);
413 
414  if (start >= 0) {
415  prev_dsfile = dsfile;
416  prev_start = start;
417  break;
418  }
419  }
420  else {
421 
422  /* We want the datasets for all times in the specified range */
423 
424  start = cds_find_timeval_index(
425  dsfile->ntimes, dsfile->timevals, *begin_timeval, CDS_GTEQ);
426 
427  end = cds_find_timeval_index(
428  dsfile->ntimes, dsfile->timevals, *end_timeval, CDS_LTEQ);
429 
430  if ((start >= 0) &&
431  (end >= start)) {
432 
433  count = end - start + 1;
434 
435  if (!_dsproc_fetch_dsfile_dataset(
436  dsfile, start, count, nvars, var_names, parent)) {
437 
438  return(-1);
439  }
440 
441  nobs++;
442  }
443  }
444 
445  } /* end loop over all datastream files */
446 
447  if (prev_dsfile) {
448 
449  if (!_dsproc_fetch_dsfile_dataset(
450  dsfile, prev_start, 1, nvars, var_names, parent)) {
451 
452  return(-1);
453  }
454 
455  nobs++;
456  }
457 
458  if (merge_obs &&
459  nobs > 1) {
460 
461  nobs = _dsproc_merge_obs(parent);
462  }
463 
464  return(nobs);
465 }
466 
467 /**
468  * Private: Fetch the times of previously stored data.
469  *
470  * This function will search the specified datastream files and retrieve
471  * all times within the specified time range. The _dsproc_find_dsfiles()
472  * function should be used to get the dsfiles list.
473  *
474  * If the begin_timeval is not specified, the time just prior to
475  * the end_timeval will be returned.
476  *
477  * If the end_timeval is not specified, the time just after
478  * the begin_timeval will be returned.
479  *
480  * Memory will be allocated for the returned array of times if the output
481  * array is NULL. In this case the calling process is responsible for
482  * freeing the allocated memory.
483  *
484  * If an error occurs in this function it will be appended to the log and
485  * error mail messages, and the process status will be set appropriately.
486  *
487  * @param ds - pointer to the DataStream structure
488  * @param ndsfiles - number of datastream Files
489  * @param dsfiles - list of datastream Files to search
490  * @param begin_timeval - beginning of the time range to search
491  * @param end_timeval - end of the time range to search
492  * @param ntimevals - pointer to the number of timevals
493  * - input:
494  * - length of the output array
495  * - ignored if the output array is NULL
496  * - output:
497  * - number of timevals returned
498  * - 0 if no times were found in the specified range
499  * - (size_t)-1 if a memory allocation error occurs
500  * @param timevals - pointer to the output array
501  * or NULL to dynamically allocate the memory needed.
502  *
503  * @return
504  * - pointer to the array of timevals
505  * - NULL if:
506  * - no times were found in the specified range (ntimevals == 0)
507  * - an error occurred (ntimevals == (size_t)-1)
508  */
509 timeval_t *_dsproc_fetch_timevals(
510  DataStream *ds,
511  int ndsfiles,
512  DSFile **dsfiles,
513  timeval_t *begin_timeval,
514  timeval_t *end_timeval,
515  size_t *ntimevals,
516  timeval_t *timevals)
517 {
518  DSFile *dsfile;
519  size_t max_ntimes;
520  int free_timevals;
521  int start;
522  int end;
523  int fi, fti;
524  size_t oti;
525 
526  if (!ndsfiles) {
527  *ntimevals = 0;
528  return((timeval_t *)NULL);
529  }
530 
531  if ((!begin_timeval || !begin_timeval->tv_sec) &&
532  (!end_timeval || !end_timeval->tv_sec)) {
533 
534  *ntimevals = 0;
535  return((timeval_t *)NULL);
536  }
537 
538  /* Determine the maximum number of times to get */
539 
540  if (timevals) {
541  free_timevals = 0;
542  max_ntimes = *ntimevals;
543  }
544  else {
545 
546  free_timevals = 1;
547  max_ntimes = dsfiles[0]->ntimes;
548 
549  for (fi = 1; fi < ndsfiles; fi++) {
550  max_ntimes += dsfiles[fi]->ntimes;
551  }
552 
553  timevals = (timeval_t *)calloc(max_ntimes, sizeof(timeval_t));
554  if (!timevals) {
555 
556  ERROR( DSPROC_LIB_NAME,
557  "Could not fetch times from datastream: %s\n"
558  " -> memory allocation error",
559  ds->name);
560 
561  dsproc_set_status(DSPROC_ENOMEM);
562  *ntimevals = (size_t)-1;
563  return((timeval_t *)NULL);
564  }
565  }
566 
567  /* Loop over all datastream files */
568 
569  oti = 0;
570 
571  for (fi = 0; fi < ndsfiles; fi++) {
572 
573  dsfile = dsfiles[fi];
574 
575  if (!dsfile->ntimes) {
576  continue;
577  }
578 
579  if (!begin_timeval || !begin_timeval->tv_sec) {
580 
581  /* We want the time just prior to the end_timeval */
582 
583  start = cds_find_timeval_index(
584  dsfile->ntimes, dsfile->timevals, *end_timeval, CDS_LT);
585 
586  if (start >= 0) {
587  timevals[0] = dsfile->timevals[start];
588  oti = 1;
589  }
590  else {
591  break;
592  }
593  }
594  else if (!end_timeval || !end_timeval->tv_sec) {
595 
596  /* We want the time just after the begin_timeval */
597 
598  start = cds_find_timeval_index(
599  dsfile->ntimes, dsfile->timevals, *begin_timeval, CDS_GT);
600 
601  if (start >= 0) {
602  timevals[0] = dsfile->timevals[start];
603  oti = 1;
604  break;
605  }
606  }
607  else {
608 
609  /* We want all times in the specified range */
610 
611  start = cds_find_timeval_index(
612  dsfile->ntimes, dsfile->timevals, *begin_timeval, CDS_GTEQ);
613 
614  end = cds_find_timeval_index(
615  dsfile->ntimes, dsfile->timevals, *end_timeval, CDS_LTEQ);
616 
617  if ((start >= 0) &&
618  (end >= start)) {
619 
620  for (fti = start; fti <= end; ) {
621  timevals[oti++] = dsfile->timevals[fti++];
622  if (oti == max_ntimes) break;
623  }
624 
625  if (oti == max_ntimes) break;
626  }
627  }
628 
629  } /* end loop over all datastream files */
630 
631  if (!oti) {
632  if (free_timevals) free(timevals);
633  timevals = (timeval_t *)NULL;
634  }
635 
636  *ntimevals = oti;
637 
638  return(timevals);
639 }
640 
641 /** @publicsection */
642 
643 /*******************************************************************************
644  * Internal Functions Visible To The Public
645  */
646 
647 /*******************************************************************************
648  * Public Functions
649  */
650 
651 /**
652  * Fetch a dataset from previously stored data.
653  *
654  * This function will retrieve a dataset from the previously stored data
655  * for the specified datastream and time range.
656  *
657  * If the begin_timeval is not specified, data for the time just prior to
658  * the end_timeval will be retrieved.
659  *
660  * If the end_timeval is not specified, data for the time just after the
661  * begin_timeval will be retrieved.
662  *
663  * If both the begin and end times are not specified, the data previously
664  * retrieved by this function will be returned.
665  *
666  * This memory used by the returned dataset belongs to the internal datastream
667  * structure and must not be freed by the calling process. This dataset will
668  * remain valid until the next call to this function using a different time
669  * range and/or different variable names.
670  *
671  * If an error occurs in this function it will be appended to the log and
672  * error mail messages, and the process status will be set appropriately.
673  *
674  * @param ds_id - datastream ID
675  * @param begin_timeval - beginning of the time range to search
676  * @param end_timeval - end of the time range to search
677  * @param nvars - the number of variables to retrieve
678  * (or 0 to retrieve all variables)
679  * @param var_names - the list if variable names to retrieve
680  * (or NULL to retrieve all variables)
681  * @param merge_obs - flag specifying if multiple observations should
682  * be merged when possible (0 == false, 1 == true)
683  * @param dataset - pointer to the retrieved dataset
684  *
685  * @retval nobs the number of observations in the returned dataset
686  * @retval -1 if an error occurred
687  */
688 int dsproc_fetch_dataset(
689  int ds_id,
690  timeval_t *begin_timeval,
691  timeval_t *end_timeval,
692  size_t nvars,
693  const char **var_names,
694  int merge_obs,
695  CDSGroup **dataset)
696 {
697  DataStream *ds = _DSProc->datastreams[ds_id];
698  timeval_t search_begin = {0, 0};
699  timeval_t search_end = {0, 0};
700  timeval_t data_begin = {0, 0};
701  timeval_t data_end = {0, 0};
702  int ndsfiles;
703  DSFile **dsfiles;
704  int new_request;
705  int nobs;
706  char ts1[32], ts2[32];
707  size_t vi;
708 
709  /************************************************************
710  * Initialize Variables
711  *************************************************************/
712 
713  *dataset = (CDSGroup *)NULL;
714 
715  if (begin_timeval) search_begin = *begin_timeval;
716  if (end_timeval) search_end = *end_timeval;
717 
718  if (msngr_debug_level || msngr_provenance_level) {
719 
720  if (search_begin.tv_sec) format_timeval(&search_begin, ts1);
721  else strcpy(ts1, "N/A");
722 
723  if (search_end.tv_sec) format_timeval(&search_end, ts2);
724  else strcpy(ts2, "N/A");
725 
726  DEBUG_LV1( DSPROC_LIB_NAME,
727  "%s: Fetching previously stored datasets\n"
728  " - search begin: %s\n"
729  " - search end: %s\n", ds->name, ts1, ts2);
730  }
731 
732  /************************************************************
733  * If the begin and end times were not specified, return the
734  * dataset previously retrieved by this function.
735  *************************************************************/
736 
737  if (!search_begin.tv_sec &&
738  !search_end.tv_sec) {
739 
740  if (ds->fetched_cds) {
741 
742  DEBUG_LV1( DSPROC_LIB_NAME,
743  " - returning dataset from previous request\n");
744 
745  *dataset = ds->fetched_cds;
746  return(ds->fetched_cds->ngroups);
747  }
748  else {
749 
750  DEBUG_LV1( DSPROC_LIB_NAME,
751  " - no previous dataset to return\n");
752 
753  return(0);
754  }
755  }
756 
757  /************************************************************
758  * Check if we have already retrieved the dataset for this
759  * request, or clear the results from the previous request.
760  *************************************************************/
761 
762  new_request = 0;
763 
764  if (!_dsproc_get_fetched_range(ds, &data_begin, &data_end)) {
765  new_request = 1;
766  }
767  else if (
768  (search_begin.tv_sec == 0 || ds->fetch_begin.tv_sec == 0) &&
769  (search_begin.tv_sec != ds->fetch_begin.tv_sec) ) {
770 
771  new_request = 1;
772  }
773  else if (
774  (search_begin.tv_sec != 0) &&
775  (TV_LT(search_begin, ds->fetch_begin) ||
776  TV_GT(search_begin, data_begin)) ) {
777 
778  new_request = 1;
779  }
780  else if (
781  (search_end.tv_sec == 0 || ds->fetch_end.tv_sec == 0) &&
782  (search_end.tv_sec != ds->fetch_end.tv_sec) ) {
783 
784  new_request = 1;
785  }
786  else if (
787  (search_end.tv_sec != 0) &&
788  (TV_LT(search_end, data_end) ||
789  TV_GT(search_end, ds->fetch_end)) ) {
790 
791  new_request = 1;
792  }
793  else if (ds->fetch_nvars != 0) {
794 
795  if (nvars == 0) {
796  new_request = 1;
797  }
798  else {
799  for (vi = 0; vi < nvars; vi++) {
800  if (!cds_get_var(ds->fetched_cds, var_names[vi])) break;
801  }
802 
803  if (vi != nvars) {
804  new_request = 1;
805  }
806  }
807  }
808 
809  if (new_request) {
810  _dsproc_free_datastream_fetched_cds(ds);
811  }
812  else {
813 
814  DEBUG_LV1( DSPROC_LIB_NAME,
815  " - returning dataset from previous request\n");
816 
817  *dataset = ds->fetched_cds;
818  return(ds->fetched_cds->ngroups);
819  }
820 
821  /************************************************************
822  * Get the list of datastream files in the requested range
823  *************************************************************/
824 
825  ndsfiles = _dsproc_find_dsfiles(
826  ds->dir, &search_begin, &search_end, &dsfiles);
827 
828  if (ndsfiles <= 0) {
829 
830  if (ndsfiles == 0) {
831  DEBUG_LV1( DSPROC_LIB_NAME,
832  " - no stored data found for requested range\n");
833  }
834 
835  return(ndsfiles);
836  }
837 
838  /************************************************************
839  * Fetch the data.
840  *************************************************************/
841 
842  ds->fetched_cds = cds_define_group(NULL, ds->name);
843  if (!ds->fetched_cds) {
844  dsproc_set_status(DSPROC_ENOMEM);
845  return(-1);
846  }
847 
848  nobs = _dsproc_fetch_dataset(
849  ndsfiles, dsfiles, &search_begin, &search_end,
850  nvars, var_names, merge_obs, ds->fetched_cds);
851 
852  free(dsfiles);
853 
854  if (nobs <= 0) {
855 
856  if (nobs == 0) {
857  DEBUG_LV1( DSPROC_LIB_NAME,
858  " - no stored data found for requested range\n");
859  }
860 
861  _dsproc_free_datastream_fetched_cds(ds);
862  return(nobs);
863  }
864 
865  if (msngr_debug_level || msngr_provenance_level) {
866 
867  _dsproc_get_fetched_range(ds, &data_begin, &data_end);
868 
869  format_timeval(&data_begin, ts1);
870  format_timeval(&data_end, ts2);
871 
872  DEBUG_LV1( DSPROC_LIB_NAME,
873  " - data begin: %s\n"
874  " - data end: %s\n", ts1, ts2);
875  }
876 
877  ds->fetch_begin = search_begin;
878  ds->fetch_end = search_end;
879  ds->fetch_nvars = nvars;
880 
881  *dataset = ds->fetched_cds;
882 
883  return(ds->fetched_cds->ngroups);
884 }
885 
886 /**
887  * Fetch the times of previously stored data.
888  *
889  * This function will retrieve the times of previously stored data for the
890  * specified datastream and time range.
891  *
892  * If the begin_timeval is not specified, the time just prior to the
893  * end_timeval will be returned.
894  *
895  * If the end_timeval is not specified, the time just after the
896  * begin_timeval will be returned.
897  *
898  * Memory will be allocated for the returned array of times if the output
899  * array is NULL. In this case the calling process is responsible for
900  * freeing the allocated memory.
901  *
902  * If an error occurs in this function it will be appended to the log and
903  * error mail messages, and the process status will be set appropriately.
904  *
905  * @param ds_id - datastream ID
906  * @param begin_timeval - beginning of the time range to search
907  * @param end_timeval - end of the time range to search
908  * @param ntimevals - pointer to the number of timevals
909  * - input:
910  * - length of the output array
911  * - ignored if the output array is NULL
912  * - output:
913  * - number of timevals returned
914  * - 0 if no times were found in the specified range
915  * - (size_t)-1 if a memory allocation error occurs
916  * @param timevals - pointer to the output array
917  * or NULL to dynamically allocate the memory needed.
918  *
919  * @return
920  * - pointer to the array of timevals
921  * - NULL if:
922  * - no times were found in the specified range (ntimevals == 0)
923  * - an error occurred (ntimevals == (size_t)-1)
924  */
925 timeval_t *dsproc_fetch_timevals(
926  int ds_id,
927  timeval_t *begin_timeval,
928  timeval_t *end_timeval,
929  size_t *ntimevals,
930  timeval_t *timevals)
931 {
932  DataStream *ds = _DSProc->datastreams[ds_id];
933  int ndsfiles;
934  DSFile **dsfiles;
935  char ts1[32], ts2[32];
936 
937  if (msngr_debug_level || msngr_provenance_level) {
938 
939  if (begin_timeval && begin_timeval->tv_sec) {
940  format_timeval(begin_timeval, ts1);
941  }
942  else {
943  strcpy(ts1, "N/A");
944  }
945 
946  if (end_timeval && end_timeval->tv_sec) {
947  format_timeval(end_timeval, ts2);
948  }
949  else {
950  strcpy(ts2, "N/A");
951  }
952 
953  DEBUG_LV1( DSPROC_LIB_NAME,
954  "%s: Fetching times of previously stored data\n"
955  " - search begin: %s\n"
956  " - search end: %s\n", ds->name, ts1, ts2);
957  }
958 
959  /************************************************************
960  * Get the list of datastream files in the requested range
961  *************************************************************/
962 
963  ndsfiles = _dsproc_find_dsfiles(
964  ds->dir, begin_timeval, end_timeval, &dsfiles);
965 
966  if (ndsfiles <= 0) {
967 
968  if (ndsfiles == 0) {
969  DEBUG_LV1( DSPROC_LIB_NAME,
970  " - no stored data found for requested range\n");
971  }
972 
973  *ntimevals = (size_t)ndsfiles;
974  return((timeval_t *)NULL);
975  }
976 
977  /************************************************************
978  * Fetch the times
979  *************************************************************/
980 
981  timevals = _dsproc_fetch_timevals(
982  ds, ndsfiles, dsfiles, begin_timeval, end_timeval, ntimevals, timevals);
983 
984  free(dsfiles);
985 
986  if (!timevals) {
987 
988  if (*ntimevals == 0) {
989  DEBUG_LV1( DSPROC_LIB_NAME,
990  " - no stored data found for requested range\n");
991  }
992 
993  return((timeval_t *)NULL);
994  }
995 
996  if (msngr_debug_level || msngr_provenance_level) {
997 
998  format_timeval(&timevals[0], ts1);
999  format_timeval(&timevals[*ntimevals - 1], ts2);
1000 
1001  DEBUG_LV1( DSPROC_LIB_NAME,
1002  " - data begin: %s\n"
1003  " - data end: %s\n", ts1, ts2);
1004  }
1005 
1006  return(timevals);
1007 }