working simple streamer

examples/numpy_write_example.cpp

@@ -1,6 +1,6 @@
 // Your First C++ Program
-#include "aare/file_io/File.hpp"
 #include "aare/core/Frame.hpp"
+#include "aare/file_io/File.hpp"
 #include <iostream>
 
 #define AARE_ROOT_DIR_VAR "PROJECT_ROOT_DIR"

examples/restream_example.cpp

@@ -1,26 +1,88 @@
-#include <iostream>
-#include <string>
-#include <filesystem>
-#include <vector>
-#include <memory>
 #include <boost/program_options.hpp>
+#include <chrono>
+#include <thread>
 
+#include "aare/file_io/File.hpp"
 #include "aare/network_io/ZmqSocketSender.hpp"
 
 #include <boost/program_options/options_description.hpp>
 
-
 using namespace aare;
 using namespace std;
 namespace po = boost::program_options;
 
 int main(int argc, char **argv) {
+    aare::logger::set_verbosity(aare::logger::DEBUG);
+
     po::options_description desc("Allowed options");
-    desc.add_options()
-    ("help", "produce help message")
-    // ("input,i", po::value<string>(), "input file");
-    ("port,p", po::value<int>(), "port number");
-    // ("loop,l", "loop over the file");
-    // po::positional_options_description pd;
-    // pd.add("input,i", 1);
+    desc.add_options()("help", "produce help message")("file,f", po::value<string>(), "input file")(
+        "port,p", po::value<uint16_t>(), "port number")("fps", po::value<uint16_t>()->default_value(1),
+                                                        "frames per second (default 1)")("loop,l",
+                                                                                         "loop over the file");
+    po::positional_options_description pd;
+    pd.add("file", -1);
+
+    po::variables_map vm;
+    try {
+        auto parsed = po::command_line_parser(argc, argv).options(desc).positional(pd).run();
+        po::store(parsed, vm);
+        po::notify(vm);
+
+    } catch (const boost::program_options::error &e) {
+        cout << e.what() << "\n";
+        cout << desc << "\n";
+        return 1;
+    }
+
+    if (vm.count("help")) {
+        cout << desc << "\n";
+        return 1;
+    }
+    if (vm.count("file") != 1) {
+        aare::logger::error("file is required");
+        cout << desc << "\n";
+        return 1;
+    }
+    if (vm.count("port") != 1) {
+        aare::logger::error("file is required");
+        cout << desc << "\n";
+        return 1;
+    }
+
+    std::string path = vm["file"].as<string>();
+    uint16_t port = vm["port"].as<uint16_t>();
+    bool loop = vm.count("loop") == 1 ? true : false;
+    uint16_t fps = vm["fps"].as<uint16_t>();
+
+    aare::logger::debug("ARGS: file:", path, "port:", port, "fps:", fps, "loop:", loop);
+    auto d = round<std::chrono::milliseconds>(std::chrono::duration<double>{1. / fps});
+    aare::logger::debug("sleeping for", d.count(), "ms");
+
+    if (!std::filesystem::exists(path)) {
+        aare::logger::error("file does not exist");
+        return 1;
+    }
+
+    std::filesystem::path tmp(path);
+
+    File file(tmp, "r");
+    string endpoint = "tcp://*:" + std::to_string(port);
+    ZmqSocketSender sender(endpoint);
+    sender.bind();
+    std::this_thread::sleep_for(d); // slow joiner problem should fix this
+
+    for (size_t frameidx = 0; frameidx < file.total_frames(); frameidx++) {
+
+        Frame frame = file.read();
+        ZmqHeader header;
+        header.frameNumber = frameidx;
+        header.data = true;
+        header.npixelsx = frame.rows();
+        header.npixelsy = frame.cols();
+        header.dynamicRange = frame.bitdepth();
+        header.imageSize = frame.size();
+
+        sender.send({header, frame});
+        std::this_thread::sleep_for(d);
+    }
 }

examples/zmq_receiver_example.cpp

@@ -7,7 +7,7 @@
 using namespace aare;
 
 int main() {
-    // aare::logger::set_verbosity(aare::logger::DEBUG);
+    aare::logger::set_verbosity(aare::logger::DEBUG);
     std::string endpoint = "tcp://localhost:5555";
     aare::ZmqSocketReceiver socket(endpoint);
     socket.connect();
@@ -19,15 +19,15 @@ int main() {
         aare::logger::info("Frame size:", v[0].frame.size());
         aare::logger::info("Header:", v[0].header.to_string());
 
-        for (ZmqFrame zmq_frame : v) {
-            auto &[header, frame] = zmq_frame;
-            for (int i = 0; i < 1024; i++) {
-                for (int j = 0; j < 1024; j++) {
-                    assert(*(uint32_t *)frame.get(i, j) == (uint32_t)i + j);
-                }
-            }
-            aare::logger::info("Frame verified");
-        }
+        // for (ZmqFrame zmq_frame : v) {
+        //     auto &[header, frame] = zmq_frame;
+        //     for (int i = 0; i < 1024; i++) {
+        //         for (int j = 0; j < 1024; j++) {
+        //             assert(*(uint32_t *)frame.get(i, j) == (uint32_t)i + j);
+        //         }
+        //     }
+        //     aare::logger::info("Frame verified");
+        // }
     }
     return 0;
 }

file_io/include/aare/file_io/NumpyFile.hpp

@@ -1,8 +1,8 @@
 #pragma once
 #include "aare/core/DType.hpp"
+#include "aare/core/defs.hpp"
 #include "aare/file_io/FileInterface.hpp"
 #include "aare/file_io/NumpyHelpers.hpp"
-#include "aare/core/defs.hpp"
 #include <filesystem>
 #include <iostream>
 #include <numeric>

file_io/include/aare/file_io/NumpyFileFactory.hpp

@@ -1,7 +1,7 @@
 #pragma once
+#include "aare/core/defs.hpp"
 #include "aare/file_io/FileFactory.hpp"
 #include "aare/file_io/NumpyFile.hpp"
-#include "aare/core/defs.hpp"
 #include <fstream>
 
 namespace aare {

file_io/include/aare/file_io/RawFile.hpp

@@ -1,8 +1,8 @@
 #pragma once
-#include "aare/file_io/FileInterface.hpp"
 #include "aare/core/Frame.hpp"
-#include "aare/file_io/SubFile.hpp"
 #include "aare/core/defs.hpp"
+#include "aare/file_io/FileInterface.hpp"
+#include "aare/file_io/SubFile.hpp"
 
 namespace aare {
 

file_io/src/RawFileFactory.cpp

@@ -1,7 +1,7 @@
 #include "aare/file_io/RawFileFactory.hpp"
+#include "aare/core/defs.hpp"
 #include "aare/file_io/RawFile.hpp"
 #include "aare/file_io/SubFile.hpp"
-#include "aare/core/defs.hpp"
 #include "aare/file_io/helpers.hpp"
 #include "aare/utils/logger.hpp"
 

network_io/include/aare/network_io/ZmqSocketReceiver.hpp

@@ -20,7 +20,6 @@ class ZmqSocketReceiver : public ZmqSocket {
   public:
     ZmqSocketReceiver(const std::string &endpoint);
     void connect();
-    ZmqFrame receive(ZmqFrame &zmq_frame);
     std::vector<ZmqFrame> receive_n();
 
   private:

network_io/src/ZmqSocketReceiver.cpp

@@ -43,7 +43,9 @@ void ZmqSocketReceiver::connect() {
 ZmqHeader ZmqSocketReceiver::receive_header() {
 
     // receive string ZmqHeader
+    aare::logger::debug("Receiving header");
     size_t header_bytes_received = zmq_recv(m_socket, m_header_buffer, m_max_header_size, 0);
+    aare::logger::debug("Bytes: ", header_bytes_received);
 
     m_header_buffer[header_bytes_received] = '\0'; // make sure we zero terminate
     if (header_bytes_received < 0) {
@@ -80,6 +82,7 @@ int ZmqSocketReceiver::receive_data(std::byte *data, size_t size) {
 ZmqFrame ZmqSocketReceiver::receive_zmqframe() {
     // receive header from zmq and parse it
     ZmqHeader header = receive_header();
+
     if (!header.data) {
         // no data following header
         return {header, Frame(0, 0, 0)};

network_io/src/ZmqSocketSender.cpp

@@ -17,7 +17,10 @@ void ZmqSocketSender::bind() {
     m_context = zmq_ctx_new();
     m_socket = zmq_socket(m_context, ZMQ_PUB);
     size_t rc = zmq_bind(m_socket, m_endpoint.c_str());
-    assert(rc == 0);
+    if (rc != 0) {
+        std::string error = zmq_strerror(zmq_errno());
+        throw network_io::NetworkError("zmq_bind failed: " + error);
+    }
 }
 
 /**
@@ -33,6 +36,7 @@ size_t ZmqSocketSender::send(const ZmqHeader &header, const std::byte *data, siz
     //     rc = zmq_send(m_socket, &header, sizeof(ZmqHeader), ZMQ_SNDMORE);
     //     assert(rc == sizeof(ZmqHeader));
     std::string header_str = header.to_string();
+    aare::logger::debug("Header :", header_str);
     rc = zmq_send(m_socket, header_str.c_str(), header_str.size(), ZMQ_SNDMORE);
     assert(rc == header_str.size());
     if (data == nullptr) {