Updates to daqd_dc_zmq as part of debugging.

Added a program to check the output of the daqd against the FE simulated channels.

git-svn-id: https://redoubt.ligo-wa.caltech.edu/svn/advLigoRTS/trunk@4551 6dcd42c9-f523-4c6d-aada-af552506706e

CMakeLists.txt

@@ -21,6 +21,7 @@ FIND_PACKAGE(FrameCPP REQUIRED)
 FIND_PACKAGE(MX)
 FIND_PACKAGE(OpenMX)
 FIND_PACKAGE(ZMQ4)
+FIND_PACKAGE(NDS2Client)
 
 CHECK_CXX_SOURCE_COMPILES("#include <iostream>
 #include <FlexLexer.h>

config/cmake/FindNDS2Client.cmake

+
+if (cds_find_nds2_client_included)
+    return()
+endif(cds_find_nds2_client_included)
+set(cds_find_nds2_client_included TRUE)
+
+pkg_check_modules(libNDS2Client nds2-client>=0.14.3)
+
+if (libNDS2Client_FOUND)
+    find_library(libNDS2Client_lib_path ndsxxwrap
+            PATHS ${libNDS2Client_LIBRARY_DIRS})
+
+    find_path(libNDS2client_include_path nds.hh
+            PATHS ${libNDS2Client_INCLUDE_DIRS})
+
+    add_library(_nds2client SHARED IMPORTED)
+    set_target_properties(_nds2client PROPERTIES
+            IMPORTED_LOCATION ${libNDS2Client_lib_path})
+
+    add_library(_nds2client_intl INTERFACE)
+    target_include_directories(_nds2client_intl INTERFACE ${libNDS2client_include_path})
+    target_link_libraries(_nds2client_intl INTERFACE _nds2client)
+
+    add_library(nds2client::cxx ALIAS _nds2client_intl)
+
+
+endif(libNDS2Client_FOUND)
+

src/fe_stream_test/CMakeLists.txt

-add_library(fe_stream_generator INTERFACE)
-target_include_directories(fe_stream_generator INTERFACE ${CMAKE_CURRENT_SOURCE_DIR})
+add_library(fe_stream_generator STATIC
+        fe_stream_generator.hh
+        str_split.cc
+        str_split.hh)
+target_include_directories(fe_stream_generator PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
 
 add_executable(fe_stream_test
         fe_stream_test.cc
@@ -7,4 +10,16 @@ add_executable(fe_stream_test
         )
 target_link_libraries(fe_stream_test
         PRIVATE
-        fe_stream_generator)
\ No newline at end of file
+        fe_stream_generator)
+
+if (${libNDS2Client_FOUND})
+
+    add_executable(fe_stream_check
+            fe_stream_check.cc
+            )
+    target_link_libraries(fe_stream_check
+            PRIVATE
+            fe_stream_generator
+            nds2client::cxx)
+
+endif (${libNDS2Client_FOUND})
\ No newline at end of file

src/fe_stream_test/Makefile

 all: fe_stream_test
 
-fe_stream_test.o: fe_stream_test.cc gps.hh ../drv/rfm.c ../drv/mbuf/mbuf.c
+fe_stream_test.o: fe_stream_test.cc gps.hh ../drv/rfm.c ../drv/mbuf/mbuf.c str_split.hh
 	g++ -c -O -g3 fe_stream_test.cc -o fe_stream_test.o -I.
 
-fe_stream_test: fe_stream_test.o
-	g++ fe_stream_test.o -o fe_stream_test -lrt
+str_split.o: str_split.cc str_split.hh
+	g++ -c -O -g3 str_split.cc -o str_split.o -I.
+
+fe_stream_test: fe_stream_test.o str_split.o
+	g++ fe_stream_test.o str_split.o -o fe_stream_test -lrt
 
 clean:
 	rm fe_stream_test.o fe_stream_test

src/fe_stream_test/fe_stream_check.cc

+#include <arpa/inet.h>
+
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+#include <sstream>
+#include <stdexcept>
+
+#include <nds.hh>
+
+#include "fe_stream_generator.hh"
+
+struct Config {
+    std::string hostname;
+    unsigned short port;
+    NDS::buffer::gps_second_type gps;
+    std::vector<std::string> channels;
+};
+
+void usage()
+{
+    using namespace std;
+    cerr << "Usage:\n";
+    cerr << "\t<server> <port> 'live'|gps_time channellist" << std::endl;
+    exit(1);
+}
+
+Config get_config(int argc, char *argv[])
+{
+    Config cfg;
+    if (argc < 5)
+        throw std::runtime_error("Insufficient arguments");
+    cfg.hostname = argv[1];
+    {
+        std::istringstream is(argv[2]);
+        is >> cfg.port;
+    }
+    if (std::strcmp("live", argv[3]) == 0) {
+        cfg.gps = 0;
+    } else {
+        std::istringstream is(argv[3]);
+        is >> cfg.gps;
+    }
+    for (int i = 4; i < argc; ++i)
+    {
+        cfg.channels.push_back(argv[i]);
+    }
+    return cfg;
+}
+
+int main(int argc, char *argv[])
+{
+    Config cfg;
+    std::vector<GeneratorPtr> generators;
+    try {
+        cfg = get_config(argc, argv);
+    } catch(...) {
+        usage();
+    }
+
+    generators.reserve(cfg.channels.size());
+    for (int i = 0; i < cfg.channels.size(); ++i)
+        generators.push_back(create_generator(cfg.channels[i]));
+
+    int status = 0;
+
+    std::cout << "Starting check at " << cfg.gps << std::endl;
+
+    std::tr1::shared_ptr<NDS::connection> conn(new NDS::connection(cfg.hostname, cfg.port));
+    for (int i = 0; i < generators.size(); ++i)
+    {
+        Generator& cur_gen = *(generators[i].get());
+        NDS::connection::channel_names_type chlist;
+        chlist.push_back(generators[i]->full_channel_name());
+
+        NDS::buffers_type bufs;
+        if (cfg.gps == 0)
+        {
+
+        }
+        else
+        {
+            bufs = conn->fetch(cfg.gps, cfg.gps+1, chlist);
+        }
+        NDS::buffer& sample_buf = *(bufs[0].get());
+        if (sample_buf.samples_to_bytes(sample_buf.Samples()) != cur_gen.bytes_per_sec())
+        {
+            std::cerr << "Length mismatch on " << cur_gen.full_channel_name() << std::endl;
+            status = 1;
+            continue;
+        }
+        std::vector<char> ref_buf(cur_gen.bytes_per_sec());
+
+        {
+            char *out = &ref_buf[0];
+            int gps_sec = bufs[0]->Start();
+            int gps_nano = bufs[0]->StartNano();
+            int step = 1000000000/16;
+            for (int j = 0; j < 16; ++j)
+            {
+                out = cur_gen.generate(gps_sec, gps_nano, out);
+                gps_nano += step;
+            }
+        }
+
+        if (std::memcmp(&ref_buf[0], &sample_buf[0], ref_buf.size()) != 0)
+        {
+            std::cerr << "There is a difference in " << cur_gen.full_channel_name() << std::endl;
+            status = 1;
+            std::vector<int> bswap_buff(ref_buf.size()/4);
+            int* cur = (int*)(&ref_buf[0]);
+            for (int j = 0; j < bswap_buff.size(); ++j)
+            {
+                bswap_buff[j] = htonl(*cur);
+                ++cur;
+            }
+            if (std::memcmp(&bswap_buff[0], &sample_buf[0], ref_buf.size()) == 0)
+            {
+                std::cerr << "Byte order difference!!!!" << std::endl;
+            }
+        }
+    }
+    return status;
+}

src/fe_stream_test/fe_stream_generator.hh

@@ -13,6 +13,8 @@
 
 #include <stdexcept>
 
+#include "str_split.hh"
+
 class SimChannel
 {
 private:
@@ -64,6 +66,8 @@ public:
         return os.str();
     }
 
+    virtual size_t bytes_per_sec() const = 0;
+
     virtual char* generate(int gps_sec, int gps_nano, char *out) = 0;
 
     virtual void output_ini_entry(std::ostream& os) = 0;
@@ -84,6 +88,13 @@ namespace Generators {
         SimChannelGenerator(const SimChannel& ch):
                 ch_(ch) {}
 
+        virtual size_t bytes_per_sec() const
+        {
+            if (ch_.data_type() != 2)
+                throw std::runtime_error("Unsupported type");
+            return ch_.data_rate()*4;
+        }
+
         virtual void output_ini_entry(std::ostream& os)
         {
             os << "[" << full_channel_name() << "]\n";
@@ -150,7 +161,7 @@ namespace Generators {
 
 GeneratorPtr create_generator(const std::string& generator, const SimChannel& ch)
 {
-    if (ch.data_type() != 4)
+    if (ch.data_type() != 2)
         throw std::runtime_error("Invalid/unsupported data type for a generator");
     if (generator == "gps_sec") {
         return GeneratorPtr(new Generators::GPSSecondGenerator(ch));
@@ -158,4 +169,37 @@ GeneratorPtr create_generator(const std::string& generator, const SimChannel& ch
     throw std::runtime_error("Unknown generator type");
 }
 
+GeneratorPtr create_generator(const std::string& channel_name)
+{
+    std::vector<std::string> parts = split(channel_name, "--");
+    if (parts.size() < 4)
+        throw std::runtime_error("Generator name has too few parts, invalid input");
+    int data_type = 0;
+    {
+        std::istringstream is (parts[parts.size()-2]);
+        is >> data_type;
+    }
+    int rate = 0;
+    {
+        std::istringstream is (parts[parts.size()-1]);
+        is >> rate;
+    }
+    if (data_type != 2 || rate < 16)
+        throw std::runtime_error("Invalid data type or rate found");
+    std::string& base = parts[0];
+    std::string& name = parts[1];
+    int arg_count = parts.size()-4; // ignore base channel name, data type, rate
+    if (name == "gpssoff1p" && arg_count == 1)
+    {
+        std::istringstream is(parts[2]);
+        int offset = 0;
+        is >> offset;
+        return GeneratorPtr(new Generators::GPSSecondWithOffset(SimChannel(base, data_type, rate, 0), offset));
+    }
+    else
+    {
+        throw std::runtime_error("Unknown generator type");
+    }
+}
+
 #endif //DAQD_FE_STREAM_GENERATOR_HH

src/fe_stream_test/fe_stream_test.cc

@@ -60,8 +60,8 @@ void output_ini_files(const std::string& ini_dir, const std::string& system_name
     string fname_par = generate_par_filename(ini_dir, system_name);
     ofstream os_ini(fname_ini.c_str());
     ofstream os_par(fname_par.c_str());
-    os_ini << "[default]\ngain=1.0\nacquire=1\ndcuid=" << dcuid << "\nifoid=0\n";
-    os_ini << "datatype=4\ndatarate=2048\noffset=0\nslope=1.0\nunits=undef\n\n";
+    os_ini << "[default]\ngain=1.0\nacquire=3\ndcuid=" << dcuid << "\nifoid=0\n";
+    os_ini << "datatype=2\ndatarate=2048\noffset=0\nslope=1.0\nunits=undef\n\n";
 
     vector<GeneratorPtr>::iterator cur = channels.begin();
     for(; cur != channels.end(); ++cur)
@@ -185,7 +185,7 @@ int main(int argc, char *argv[]) {
 
         std::ostringstream ss;
         ss << system_name << "-" << i;
-        generators.push_back(GeneratorPtr(new Generators::GPSSecondWithOffset(SimChannel(ss.str(), 4, 1024, chnum), i%21)));
+        generators.push_back(GeneratorPtr(new Generators::GPSSecondWithOffset(SimChannel(ss.str(), 2, 1024, chnum), i%21)));
     }
 
     if (true)

src/fe_stream_test/str_split.cc

+#include "str_split.hh"
+
+std::vector<std::string> split(const std::string& source, const std::string& sep, split_type filter_mode) {
+    std::vector<std::string> results;
+
+    if (source == "") return results;
+    std::string::size_type prev = 0;
+    std::string::size_type pos = 0;
+    if (sep.size() > 0) {
+        std::string::size_type sep_size = sep.size();
+        while (pos < source.size() && pos != std::string::npos) {
+            pos = source.find(sep, prev);
+            if (pos != std::string::npos) {
+                std::string tmp = source.substr(prev, pos - prev);
+                if (tmp != "" || filter_mode == INCLUDE_EMPTY_STRING)
+                    results.push_back(tmp);
+                pos += sep_size;
+                prev = pos;
+            }
+        }
+    } else {
+        while (pos < source.size() && pos != std::string::npos) {
+            pos = source.find_first_of(" \t\n\r", prev);
+            if (pos != std::string::npos) {
+                std::string tmp = source.substr(prev, pos - prev);
+                if (tmp != "" || filter_mode == INCLUDE_EMPTY_STRING)
+                    results.push_back(tmp);
+                ++pos;
+                prev = pos;
+            }
+        }
+    }
+    std::string tmp = source.substr(prev);
+    if (tmp != "" || filter_mode == INCLUDE_EMPTY_STRING)
+        results.push_back(tmp);
+    return results;
+}
\ No newline at end of file

src/fe_stream_test/str_split.hh

+//
+// Created by jonathan.hanks on 11/10/17.
+//
+
+#ifndef DAQD_TRUNK_STR_SPLIT_HH
+#define DAQD_TRUNK_STR_SPLIT_HH
+
+#include <string>
+#include <vector>
+
+enum split_type {
+    INCLUDE_EMPTY_STRING=0,
+    EXCLUDE_EMPTY_STRING=1,
+};
+
+extern std::vector<std::string> split(const std::string& source, const std::string& sep, split_type filter_mode=INCLUDE_EMPTY_STRING);
+
+#endif //DAQD_TRUNK_STR_SPLIT_HH