core: Update file_meta_sink to inject headers into data stream.

When a tag with updated metadata information is received, close out the previous header (by setting the segment size) and create a new header with the new data. Specifically for sample rate and time stamps. Will be useful for extra_dict when implemented.
author: Tom Rondeau <trondeau@vt.edu> 2012-11-24 19:09:02 -0500
committer: Tom Rondeau <trondeau@vt.edu> 2012-11-24 19:09:02 -0500
commit: 9dc8f8b18043e71b50b3a254cb52bf355e97e6fa (patch)
tree: f1a00dc48cecc25ca2fbc57aa0267bb40871ac63 /gnuradio-core/src/lib/io
parent: 54c23abf9b9703b8d52704fa01b9d6069d9edf91 (diff)
3 files changed, 160 insertions, 25 deletions
diff --git a/gnuradio-core/src/lib/io/gr_file_meta_sink.cc b/gnuradio-core/src/lib/io/gr_file_meta_sink.cc
index aa34bf9b70..9d092d79fa 100644
--- a/gnuradio-core/src/lib/io/gr_file_meta_sink.cc
+++ b/gnuradio-core/src/lib/io/gr_file_meta_sink.cc
@@ -28,25 +28,31 @@
 #include <gr_io_signature.h>
 #include <stdexcept>
 
+#define HEADER_SIZE 117
+
 gr_file_meta_sink_sptr
 gr_make_file_meta_sink(size_t itemsize, const char *filename,
-		       double samp_rate, gr_file_types type, bool complex,
+		       double samp_rate, double relative_rate,
+		       gr_file_types type, bool complex,
 		       const std::string &extra_dict)
 {
   return gnuradio::get_initial_sptr
     (new gr_file_meta_sink(itemsize, filename,
-			   samp_rate, type, complex,
+			   samp_rate, relative_rate,
+			   type, complex,
 			   extra_dict));
 }
 
 gr_file_meta_sink::gr_file_meta_sink(size_t itemsize, const char *filename,
-				     double samp_rate, gr_file_types type, bool complex,
+				     double samp_rate, double relative_rate,
+				     gr_file_types type, bool complex,
 				     const std::string &extra_dict)
   : gr_sync_block("file_meta_sink",
 		  gr_make_io_signature(1, 1, itemsize),
 		  gr_make_io_signature(0, 0, 0)),
     gr_file_sink_base(filename, true),
-    d_itemsize(itemsize)
+    d_itemsize(itemsize), d_relative_rate(relative_rate),
+    d_total_seg_size(0)
 {
   if(!open(filename))
     throw std::runtime_error("file_meta_sink: can't open file\n");
@@ -55,11 +61,11 @@ gr_file_meta_sink::gr_file_meta_sink(size_t itemsize, const char *filename,
 				   pmt_from_double(0));
 
   d_header = pmt_make_dict();
-  d_header = pmt_dict_add(d_header, mp("sr"), mp(samp_rate));
-  d_header = pmt_dict_add(d_header, mp("time"), timestamp);
+  d_header = pmt_dict_add(d_header, mp("rx_rate"), mp(samp_rate));
+  d_header = pmt_dict_add(d_header, mp("rx_time"), timestamp);
   d_header = pmt_dict_add(d_header, mp("type"), pmt_from_long(type));
   d_header = pmt_dict_add(d_header, mp("cplx"), complex ? PMT_T : PMT_F);
-  d_header = pmt_dict_add(d_header, mp("strt"), pmt_from_uint64(109));
+  d_header = pmt_dict_add(d_header, mp("strt"), pmt_from_uint64(HEADER_SIZE));
   d_header = pmt_dict_add(d_header, mp("size"), pmt_from_uint64(0));
   // handle extra dictionary
 
@@ -72,6 +78,8 @@ gr_file_meta_sink::write_header(pmt_t header)
   do_update();
 
   std::string header_str = pmt_serialize_str(header);
+  if(header_str.size() != HEADER_SIZE)
+      throw std::runtime_error("file_meta_sink: header is wrong size.\n");
 
   size_t nwritten = 0;
   while(nwritten < header_str.size()) {
@@ -80,21 +88,55 @@ gr_file_meta_sink::write_header(pmt_t header)
     nwritten += count;
     if((count == 0) && (ferror(d_fp))) {
       fclose(d_fp);
-      throw std::runtime_error("file_meta_sink: error writing header to file\n");
+      throw std::runtime_error("file_meta_sink: error writing header to file.\n");
     }
   }
 }
 
-gr_file_meta_sink::~gr_file_meta_sink()
+bool
+gr_file_meta_sink::update_header(pmt_t key, pmt_t value)
 {
-  fseek(d_fp, 0, SEEK_SET);
+  // Special handling caveat to transform rate from radio source into
+  // the rate at this sink.
+  if(pmt_eq(key, mp("rx_rate"))) {
+    double rate = pmt_to_double(value);
+    value = pmt_from_double(rate*d_relative_rate);
+  }
 
-  // Replace the dictionary item with the data size now that we're
-  // done.
-  uint64_t s = nitems_read(0) * d_itemsize;
-  d_header = pmt_dict_delete(d_header, mp("size"));
-  d_header = pmt_dict_add(d_header, mp("size"), pmt_from_uint64(s));
-  write_header(d_header);
+  if(pmt_dict_has_key(d_header, key)) {
+    d_header = pmt_dict_add(d_header, key, value);
+    return true;
+  }
+  else {
+    return false;
+  }
+}
+
+uint64_t
+gr_file_meta_sink::get_last_header_loc()
+{
+  uint64_t loc = 0;
+  pmt_t v = pmt_dict_ref(d_header, mp("strt"), PMT_NIL);
+  if(!pmt_eq(v, PMT_NIL))
+    loc = pmt_to_uint64(v) - HEADER_SIZE;
+  return loc;
+}
+
+gr_file_meta_sink::~gr_file_meta_sink()
+{
+  // Replace the last header block with the final count of the number
+  // of items.
+  uint64_t loc = get_last_header_loc();
+  uint64_t seg_size = nitems_read(0) * d_itemsize - d_total_seg_size;
+  pmt_t s = pmt_from_uint64(seg_size);
+  //std::cerr << "Destructor" << std::endl;
+  //std::cerr << "  location of last header: " << loc << std::endl;
+  //std::cerr << "  nitems_read:  " << nitems_read(0)*d_itemsize << std::endl;
+  //std::cerr << "  Segment Size: " << seg_size << std::endl;
+  if(update_header(mp("size"), s)) {
+    fseek(d_fp, loc, SEEK_SET);
+    write_header(d_header);
+  }
 }
 
 int
@@ -110,6 +152,60 @@ gr_file_meta_sink::work(int noutput_items,
   if(!d_fp)
     return noutput_items;		// drop output on the floor
 
+  uint64_t abs_N = nitems_read(0);
+  uint64_t end_N = abs_N + (uint64_t)(noutput_items);
+  std::vector<gr_tag_t> all_tags;
+  get_tags_in_range(all_tags, 0, abs_N, end_N);
+  
+  std::vector<gr_tag_t>::iterator itr;
+  for(itr = all_tags.begin(); itr != all_tags.end(); itr++) {
+    // Special case where info is carried on the first tag, so we just
+    // overwrite the first header.
+    if(itr->offset == 0) {
+      if(update_header(itr->key, itr->value)) {
+	fseek(d_fp, 0, SEEK_SET);
+	write_header(d_header);
+      }
+    }
+    else {
+      // Update the last header info with the number of samples this
+      // block represents.
+      uint64_t loc = get_last_header_loc();
+      uint64_t seg_size = itr->offset * d_itemsize - d_total_seg_size;
+      pmt_t s = pmt_from_uint64(seg_size);
+      //std::cerr << "Found Tag at: " << itr->offset*d_itemsize << std::endl;
+      //std::cerr << "  last header starts at: " << loc << std::endl;
+      //std::cerr << "  segment size is:       " << seg_size << std::endl;    
+      if(update_header(mp("size"), s)) {
+	fseek(d_fp, loc, SEEK_SET);
+	write_header(d_header);
+      }
+
+      if(update_header(itr->key, itr->value)) {
+	// Otherwise, set current size of chunk to 0 and start of
+	// chunk based on current index + header size.
+	d_total_seg_size += seg_size;
+	s = pmt_from_uint64(0);
+	if(update_header(mp("size"), s)) {
+	  // If we have multiple tags on the same offset, this makes
+	  // sure we just overwrite the same header each time instead
+	  // of creating a new header per tag.
+	  uint64_t seg_start = loc;
+	  if(seg_size != 0)
+	    seg_start += HEADER_SIZE + seg_size;
+	  pmt_t s = pmt_from_uint64(seg_start + HEADER_SIZE);
+	  if(update_header(mp("strt"), s)) {
+	    //std::cerr << "Adding new header" << std::endl;
+	    //std::cerr << "  new header start at: " << seg_start-HEADER_SIZE << std::endl;
+	    //std::cerr << "  new seg start at:    " << seg_start << std::endl;
+	    fseek(d_fp, seg_start, SEEK_SET);
+	    write_header(d_header);
+	  }
+	}
+      }
+    }
+  }
+
   while(nwritten < noutput_items) {
     int count = fwrite(inbuf, d_itemsize, noutput_items - nwritten, d_fp);
     if(count == 0)	// FIXME add error handling
diff --git a/gnuradio-core/src/lib/io/gr_file_meta_sink.h b/gnuradio-core/src/lib/io/gr_file_meta_sink.h
index 741408de18..c0f30bdc08 100644
--- a/gnuradio-core/src/lib/io/gr_file_meta_sink.h
+++ b/gnuradio-core/src/lib/io/gr_file_meta_sink.h
@@ -46,34 +46,71 @@ typedef boost::shared_ptr<gr_file_meta_sink> gr_file_meta_sink_sptr;
 
 GR_CORE_API gr_file_meta_sink_sptr
 gr_make_file_meta_sink(size_t itemsize, const char *filename,
-		       double samp_rate, gr_file_types type, bool complex,
+		       double samp_rate, double relative_rate,
+		       gr_file_types type, bool complex,
 		       const std::string &extra_dict="");
 
 /*!
- * \brief Write stream to file.
+ * \brief Write stream to file with meta-data headers.
  * \ingroup sink_blk
+ *
+ * These files represent data as binary information in between
+ * meta-data headers. The headers contain information about the type
+ * of data and properties of the data in the next segment of
+ * samples. The information includes:
+ *
+ *   rx_rate (double): sample rate of data.
+ *   rx_time (uint64_t, double): time stamp of first sample in segment.
+ *   type (gr_file_types as int32_t): data type.
+ *   cplx (bool): Is data complex?
+ *   strt (uint64_t): Starting byte of data in this segment.
+ *   size (uint64_t): Size in bytes of data in this segment.
+ *
+ * Tags can be sent to the file to update the information, which will
+ * create a new header. Headers are found by searching from the first
+ * header (at position 0 in the file) and reading where the data
+ * segment starts plus the data segment size. Following will either be
+ * a new header or EOF.
  */
-
 class GR_CORE_API gr_file_meta_sink : public gr_sync_block, public gr_file_sink_base
 {
-  
+  /*!
+   * \brief Create a meta-data file sink.
+   *
+   * \param itemsize (size_t): Size of data type.
+   * \param filename (string): Name of file to write data to.
+   * \param samp_rate (double): Sample rate of data. If sample rate will be
+   *    set by a tag, such as rx_tag from a UHD source, this is
+   *    basically ignored.
+   * \param relative_rate (double): Rate chance from source of sample
+   *    rate tag to sink.
+   * \param type (gr_file_types): Data type (int, float, etc.)
+   * \param complex (bool): If data stream is complex
+   * \param extra_dict (string): a serialized PMT dictionary of extra
+   *    information. Currently not supported.
+   */
   friend GR_CORE_API gr_file_meta_sink_sptr
     gr_make_file_meta_sink(size_t itemsize, const char *filename,
-			   double samp_rate, gr_file_types type, bool complex,
+			   double samp_rate, double relative_rate,
+			   gr_file_types type, bool complex,
 			   const std::string &extra_dict);
 
  private:
   size_t d_itemsize;
+  double d_relative_rate;
+  uint64_t d_total_seg_size;
   pmt_t d_header;
   pmt_t d_extra_dict;
 
  protected:
   gr_file_meta_sink(size_t itemsize, const char *filename,
-		    double samp_rate, gr_file_types type, bool complex,
+		    double samp_rate, double relative_rate,
+		    gr_file_types type, bool complex,
 		    const std::string &extra_dict);
 
-
   void write_header(pmt_t header);
+  bool update_header(pmt_t key, pmt_t value);
+  uint64_t get_last_header_loc();
 
  public:
   ~gr_file_meta_sink();
diff --git a/gnuradio-core/src/lib/io/gr_file_meta_sink.i b/gnuradio-core/src/lib/io/gr_file_meta_sink.i
index 27fb4debbf..7d90ca3049 100644
--- a/gnuradio-core/src/lib/io/gr_file_meta_sink.i
+++ b/gnuradio-core/src/lib/io/gr_file_meta_sink.i
@@ -35,14 +35,16 @@ enum gr_file_types {
 
 gr_file_meta_sink_sptr
 gr_make_file_meta_sink(size_t itemsize, const char *filename,
-		       double samp_rate, gr_file_types type, bool complex,
+		       double samp_rate, double relative_rate,
+		       gr_file_types type, bool complex,
 		       const std::string & extra_dict="");
 
 class gr_file_meta_sink : public gr_sync_block, public gr_file_sink_base
 {
  protected:
   gr_file_meta_sink(size_t itemsize, const char *filename,
-		    double samp_rate, gr_file_types type, bool complex,
+		    double samp_rate, double relative_rate,
+		    gr_file_types type, bool complex,
 		    const std::string & extra_dict);
 
  public:
author	Tom Rondeau <trondeau@vt.edu>	2012-11-24 19:09:02 -0500
committer	Tom Rondeau <trondeau@vt.edu>	2012-11-24 19:09:02 -0500
commit	9dc8f8b18043e71b50b3a254cb52bf355e97e6fa (patch)
tree	f1a00dc48cecc25ca2fbc57aa0267bb40871ac63 /gnuradio-core/src/lib/io
parent	54c23abf9b9703b8d52704fa01b9d6069d9edf91 (diff)