Merge pull request #222 from aglowacki/master

update load netcdf callback

Author: aglowacki

src/io/file/netcdf_io.cpp

@@ -104,7 +104,11 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
                                 size_t detector,
                                 data_struct::Spectra_Line<T_real>* spec_line,
                                 size_t line_size,
-                                data_struct::Spectra<T_real>* spectra)
+                                data_struct::Spectra<T_real>* spectra,
+                                size_t cur_row,
+                                size_t max_rows,
+                                data_struct::IO_Callback_Func_Def<T_real> *callback_fun,
+                                void* user_data)
 {
     std::lock_guard<std::mutex> lock(_mutex);
 
@@ -123,6 +127,7 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
     T_real elapsed_realtime = 0.;
     T_real input_counts = 0.;
     T_real output_counts = 0.;
+    data_struct::Spectra<T_real>* callback_spectra = nullptr;
 
     size_t dim2size[NC_MAX_VAR_DIMS] = {0};
 
@@ -222,11 +227,11 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
     }
 
     size_t spec_cntr = 0;
-    if (ltype == E_load_type::LINE || ltype == E_load_type::CALLBACKF)
+    if (ltype == E_load_type::LINE)
     {
         spec_cntr = spec_line->size();
     }
-    else if (ltype == E_load_type::INTEGRATED)
+    else if (ltype == E_load_type::INTEGRATED || ltype == E_load_type::CALLBACKF)
     {
         spec_cntr = line_size;
     }
@@ -237,6 +242,10 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
 		{
 			(*spec_line)[j].resize(spectra_size); // should be renames to resize
 		}
+        else if (ltype == E_load_type::CALLBACKF)
+        {
+            callback_spectra = new data_struct::Spectra<T_real>(spectra_size);
+        }
 
         //read header
         if( (retval = _nc_get_vars_real(ncid, varid, start, count, stride, &data_in[0][0][0]) ) != 0)
@@ -264,7 +273,7 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
         unsigned short i1 = data_in[0][0][ELAPSED_LIVETIME_OFFSET+(detector*8)];
         unsigned short i2 = data_in[0][0][ELAPSED_LIVETIME_OFFSET+(detector*8)+1];
         unsigned int ii = i1 | i2<<16;
-        if (ltype == E_load_type::LINE || ltype == E_load_type::CALLBACKF)
+        if (ltype == E_load_type::LINE)
         {
             elapsed_livetime = ((float)ii) * 320e-9f; // need to multiply by this value becuase of the way it is saved
             if (elapsed_livetime == 0)
@@ -289,12 +298,15 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
         {
             elapsed_livetime += ((float)ii) * 320e-9f; 
         }
-        
+        else if(ltype == E_load_type::CALLBACKF && callback_spectra != nullptr)
+        {
+            callback_spectra->elapsed_livetime( ((float)ii) * 320e-9f);
+        }
 
         i1 = data_in[0][0][ELAPSED_REALTIME_OFFSET+(detector*8)];
         i2 = data_in[0][0][ELAPSED_REALTIME_OFFSET+(detector*8)+1];
         ii = i1 | i2<<16;
-        if (ltype == E_load_type::LINE || ltype == E_load_type::CALLBACKF)
+        if (ltype == E_load_type::LINE)
         {
             elapsed_realtime = ((float)ii) * 320e-9f; // need to multiply by this value becuase of the way it is saved
             if (elapsed_realtime == 0)
@@ -319,7 +331,10 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
         {
             elapsed_realtime += ((float)ii) * 320e-9f;
         }
-        
+        else if(ltype == E_load_type::CALLBACKF && callback_spectra != nullptr)
+        {
+            callback_spectra->elapsed_realtime(((float)ii) * 320e-9f);
+        }
 
         i1 = data_in[0][0][INPUT_COUNTS_OFFSET+(detector*8)];
         i2 = data_in[0][0][INPUT_COUNTS_OFFSET+(detector*8)+1];
@@ -349,6 +364,11 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
         {
             input_counts += ((float)ii) / elapsed_livetime;
         }
+        else if(ltype == E_load_type::CALLBACKF && callback_fun != nullptr)
+        {
+            callback_spectra->input_counts(((float)ii) / elapsed_livetime);
+        }
+
 
         i1 = data_in[0][0][OUTPUT_COUNTS_OFFSET+(detector*8)];
         i2 = data_in[0][0][OUTPUT_COUNTS_OFFSET+(detector*8)+1];
@@ -378,8 +398,12 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
         {
             output_counts += ((float)ii) / elapsed_realtime;
         }
+        else if(ltype == E_load_type::CALLBACKF && callback_fun != nullptr)
+        {
+            callback_spectra->output_counts(((float)ii) / elapsed_realtime);
+        }
 
-        if (ltype == E_load_type::LINE || ltype == E_load_type::CALLBACKF)
+        if (ltype == E_load_type::LINE)
         {
             (*spec_line)[j].elapsed_livetime(elapsed_livetime);
             (*spec_line)[j].elapsed_realtime(elapsed_realtime);
@@ -413,6 +437,12 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
                 (*spectra)(k) += data_in[0][0][k];
             }
         }
+        else if(ltype == E_load_type::CALLBACKF && callback_fun != nullptr)
+        {
+            callback_spectra->recalc_elapsed_livetime();
+            (*callback_fun)(cur_row, j, max_rows, line_size, detector, callback_spectra, user_data);
+        }
+
         start[2] += spectra_size * (4 - detector);
         count[2] = header_size;
 
@@ -455,15 +485,15 @@ size_t NetCDF_IO<T_real>::_load_spectra(E_load_type ltype,
 template<typename T_real>
 size_t NetCDF_IO<T_real>::load_spectra_line(std::string path, size_t detector, data_struct::Spectra_Line<T_real>* spec_line)
 {
-    return _load_spectra(E_load_type::LINE, path, detector, spec_line, -1, nullptr);
+    return _load_spectra(E_load_type::LINE, path, detector, spec_line, -1, nullptr, 0, 0, nullptr, nullptr);
 }
 
 //-----------------------------------------------------------------------------
 
 template<typename T_real>
 size_t NetCDF_IO<T_real>::load_spectra_line_integrated(std::string path, size_t detector, size_t line_size, data_struct::Spectra<T_real>* spectra)
 {
-    return _load_spectra(E_load_type::INTEGRATED, path, detector, nullptr, line_size, spectra);
+    return _load_spectra(E_load_type::INTEGRATED, path, detector, nullptr, line_size, spectra, 0, 0, nullptr, nullptr);
 }
 
 //-----------------------------------------------------------------------------
@@ -477,207 +507,12 @@ bool NetCDF_IO<T_real>::load_spectra_line_with_callback(std::string path,
                                                 data_struct::IO_Callback_Func_Def<T_real> callback_fun,
                                                 void* user_data)
 {
-
-    std::lock_guard<std::mutex> lock(_mutex);
-
-    size_t header_size = 256;
-    int ncid, varid, retval;
-    size_t start[] = {0, 0, 0};
-    size_t count[] = {1, 1, header_size};
-    ptrdiff_t stride[] = {1, 1, 1};
-    T_real *data_in;
-    size_t spectra_size;
-	size_t num_detectors = detector_num_arr.size();
-    int dataidx = 0;
-    T_real elapsed_livetime = 0.;
-    T_real elapsed_realtime = 0.;
-    T_real input_counts = 0.;
-    T_real output_counts = 0.;
-
-    nc_type rh_type;
-    int rh_ndims;
-    int  rh_dimids[NC_MAX_VAR_DIMS] = {0};
-    int rh_natts;
-
-    size_t dim2size[NC_MAX_VAR_DIMS] = {0};
-
-    if( (retval = nc_open(path.c_str(), NC_NOWRITE, &ncid)) != 0 )
-    {
-        logE<< path << " :: " << nc_strerror(retval)<<"\n";
-        return false;
-    }
-
-    if( (retval = nc_inq_varid(ncid, "array_data", &varid)) != 0)
-    {
-        logE<< path << " :: " << nc_strerror(retval)<<"\n";
-        nc_close(ncid);
-        return false;
-    }
-
-    if( (retval = nc_inq_var (ncid, varid, 0, &rh_type, &rh_ndims, rh_dimids, &rh_natts) ) != 0)
-    {
-        logE<< path << " :: " << nc_strerror(retval)<<"\n";
-        nc_close(ncid);
-        return false;
-    }
-
-    for (int i=0; i <  rh_ndims; i++)
-    {
-        if( (retval = nc_inq_dimlen(ncid, rh_dimids[i], &dim2size[i]) ) != 0)
-        {
-            logE<< path << " :: " << nc_strerror(retval)<<"\n";
-            nc_close(ncid);
-            return false;
-        }
-    }
-
-    data_in = new T_real[dim2size[0] * dim2size[1] * dim2size[2]];
-    count[2] = dim2size[2];
-    //read in last col sector to get total number of cols
-    //start[0] = dim2size[0] - 1;
-    if( (retval = _nc_get_vars_real(ncid, varid, start, count, stride, &data_in[0]) ) != 0)
-    {
-        delete[] data_in;
-        logE<< path << " :: " << nc_strerror(retval)<<"\n";
-        nc_close(ncid);
-        return false;
-    }
-
-    if (data_in[0] != 21930 || data_in[1] != -21931)
-    {
-        delete[] data_in;
-        logE<<"NetCDF header not found! Stopping load : "<<path<<"\n";
-        nc_close(ncid);
-        return false;
-    }
-
-    //can't read from file because it can change inbetween rows ...
-    //max_cols = (124 * (dim2size[0] - 1) ) + data_in[0][0][8];
-    header_size = data_in[2];
-    //num_cols = data_in[][0][8];  //sum all across the first dim looking at value 8
-    spectra_size = data_in[20];
-
-    start[2] += header_size;
-    count[2] = header_size + (spectra_size * MAX_NUM_SUPPORTED_DETECOTRS_PER_COL); //only 4 element detector supported per col
-
-    //loop through col sectors
-    for(size_t j = 0; j < max_cols; j++)
+    if (detector_num_arr.size() > 0)
     {
-        /*
-        //read header
-        if( (retval = _nc_get_vars_real(ncid, varid, start, count, stride, &data_in[0][0][0]) ) != 0)
-        {
-            logE<< path << " :: " << nc_strerror(retval)<<"\n";
-            nc_close(ncid);
-            return false;
-        }
-        */
-        int midx = start[2];
-
-        if (data_in[midx] != 13260 || data_in[midx + 1] != -13261)
-        {
-            delete[] data_in;
-            if(j < max_cols -2)
-            {
-                logE<<"NetCDF sub header not found! Stopping load at Col: "<<j<<" path :"<<path<<"\n";
-                nc_close(ncid);
-                return false;
-            }
-            //last two may not be filled with data
-            //TODO: send end of row stream_block down pipeline
-            nc_close(ncid);
-            return true;
-        }
-
-        for(size_t detector_num : detector_num_arr)
-        {
-            
-            if (detector_num > 3)
-            {
-                dataidx = 1;
-                detector_num -= MAX_NUM_SUPPORTED_DETECOTRS_PER_COL;
-            }
-            else
-            {
-                dataidx = 0;
-            }
-
-            midx = (dataidx * dim2size[2]) + start[2];
-
-            unsigned short i1 = data_in[midx+(ELAPSED_LIVETIME_OFFSET+(detector_num*8))];
-            unsigned short i2 = data_in[midx + (ELAPSED_LIVETIME_OFFSET+(detector_num*8)+1)];
-            unsigned int ii = i1 | i2<<16;
-            elapsed_livetime = ((float)ii) * 320e-9f; // need to multiply by this value becuase of the way it is saved
-            if(elapsed_livetime == 0)
-            {
-                if(j < max_cols-2) // usually the last two are missing which spams the log ouput.
-                {
-                    logW<<"Reading in elapsed lifetime for Col:"<<j<<" is 0. Setting it to 1.0 " << path <<"\n";
-                    elapsed_livetime = 1.0;
-                }
-            }
-
-            i1 = data_in[midx + (ELAPSED_REALTIME_OFFSET+(detector_num*8))];
-            i2 = data_in[midx + (ELAPSED_REALTIME_OFFSET+(detector_num*8)+1)];
-            ii = i1 | i2<<16;
-            elapsed_realtime = ((float)ii) * 320e-9f; // need to multiply by this value becuase of the way it is saved
-            if(elapsed_realtime == 0)
-            {
-                if(j < max_cols-2) // usually the last two are missing which spams the log ouput.
-                {
-                    logW<<"Reading in elapsed realtime for Col:"<<j<<" is 0. Setting it to 1.0 "<<path<<"\n";
-                    elapsed_realtime = 1.0;
-                }
-            }
-
-            i1 = data_in[midx + (INPUT_COUNTS_OFFSET+(detector_num*8))];
-            i2 = data_in[midx + (INPUT_COUNTS_OFFSET+(detector_num*8)+1)];
-            ii = i1 | i2<<16;
-            input_counts = ((float)ii) / elapsed_livetime;
-
-            i1 = data_in[midx + (OUTPUT_COUNTS_OFFSET+(detector_num*8))];
-            i2 = data_in[midx + (OUTPUT_COUNTS_OFFSET+(detector_num*8)+1)];
-            ii = i1 | i2<<16;
-            output_counts = ((float)ii) / elapsed_realtime;
-
-            data_struct::Spectra<T_real>* spectra = new data_struct::Spectra<T_real>(spectra_size);
-
-            spectra->elapsed_livetime(elapsed_livetime);
-            spectra->elapsed_realtime(elapsed_realtime);
-            spectra->input_counts(input_counts);
-            spectra->output_counts(output_counts);
-            spectra->recalc_elapsed_livetime();
-
-            int idx = header_size + (detector_num*spectra_size);
-
-            for(size_t k=0; k<spectra_size; k++)
-            {
-                (*spectra)[k] = data_in[midx + (idx+k)];
-            }
-
-            callback_fun(row, j, max_rows, max_cols, detector_num, spectra, user_data);
-
-        }
-
-        start[2] += count[2];
-
-        if(start[2] >= dim2size[2])
-        {
-            start[0]++;
-            start[2] = header_size;
-        }
+        size_t detector = detector_num_arr[0];
+        return _load_spectra(E_load_type::CALLBACKF, path, detector, nullptr, max_cols, nullptr, row, max_rows, &callback_fun, nullptr);
     }
-
-    delete[] data_in;
-
-    if((retval = nc_close(ncid)) != 0)
-    {
-        logE<< path << " :: " << nc_strerror(retval)<<"\n";
-        return false;
-    }
-
-    return true;
-
+    return false;
 }
 
 //-----------------------------------------------------------------------------

src/io/file/netcdf_io.h

@@ -115,7 +115,11 @@ class DLL_EXPORT NetCDF_IO
                         size_t detector,
                         data_struct::Spectra_Line<T_real>* spec_line,
                         size_t line_size, 
-                        data_struct::Spectra<T_real>* spectra);
+                        data_struct::Spectra<T_real>* spectra,
+                        size_t cur_row,
+                        size_t max_rows,
+                        data_struct::IO_Callback_Func_Def<T_real> *callback_fun,
+                        void* user_data);
 
     static NetCDF_IO *_this_inst;