examples/InteractiveStreamingExample/ADCSampleFactory.h

0001
0002 // Copyright 2020, Jefferson Science Associates, LLC.
0003 // Subject to the terms in the LICENSE file found in the top-level directory.
0004
0005
0006 #ifndef _ADCSampleFactory_h_
0007 #define _ADCSampleFactory_h_
0008
0009 #include <JANA/JFactoryT.h>
0010
0011 #include "ADCSample.h"
0012 #include "INDRAMessage.h"
0013
0014 #include <fstream>
0015
0016 class ADCSampleFactory : public JFactoryT<ADCSample> {
0017
0018     // parameters to simulate a bottle neck, spread is in sigmas
0019     size_t m_cputime_ms = 0;
0020     double m_cputime_spread = 0.25;
0021
0022     std::vector<ADCSample> m_samples;
0023     std::vector<ADCSample*> m_sample_ptrs;
0024     // Rather than creating and destroying lots of individual hit objects for each event,
0025     // we maintain a block of them and set NOT_OBJECT_OWNER. These are owned by the JFactory
0026     // so there won't be a memory leak.
0027
0028 public:
0029
0030     void Init() override {
0031         auto app = GetApplication();
0032         app->GetParameter("streamDet:rawhit_ms",     m_cputime_ms);
0033         app->GetParameter("streamDet:rawhit_spread", m_cputime_spread);
0034         SetFactoryFlag(JFactory_Flags_t::NOT_OBJECT_OWNER);
0035     }
0036
0037     void Process(const std::shared_ptr<const JEvent> &event) override {
0038
0039         // acquire the DASEventMessage via zmq
0040         // each DASEventMessage corresponds to one hardware event (readout window)
0041         // for now we pretend that hardware events = physics events
0042         auto message   = event->GetSingle<DASEventMessage>();
0043         auto source_id = message->as_indra_message()->source_id;
0044
0045         // obtain a view into our DASEventMessage payload and assign variables
0046         const char* payload_buffer;
0047         size_t payload_buffer_size;
0048         message->as_payload(&payload_buffer, &payload_buffer_size);
0049         size_t max_samples  = message->get_sample_count();
0050         size_t max_channels = message->get_channel_count();
0051
0052         if (m_samples.size() != max_channels*max_samples) {
0053             // If size actually changes (hopefully not often), create or destroy new ADCSamples as needed
0054             // They will be initialized with garbage data though
0055             m_samples.resize(max_channels*max_samples);
0056
0057             // Also recreate the vector of pointers into this buffer
0058             m_sample_ptrs.resize(max_channels*max_samples);
0059             for (size_t i=0; i < m_samples.size(); ++i) {
0060                 m_sample_ptrs[i] = &m_samples[i];
0061             }
0062         }
0063
0064         size_t i = 0;
0065         // decode the message and populate the associated jobject (hit) for the event
0066         for (uint16_t sample = 0; sample < max_samples; ++sample) {
0067             for (uint16_t channel = 0; channel < max_channels; ++channel) {
0068
0069                 // Parse in the simplest way possible
0070                 uint16_t current_value = (payload_buffer[0]-48) * 1000 + (payload_buffer[1]-48) * 100 + (payload_buffer[2]-48) * 10 + (payload_buffer[3]-48);
0071                 assert(current_value >= 0);
0072                 assert(current_value <= 1024);
0073                 payload_buffer += 5;
0074                 ADCSample& hit = m_samples[i++];
0075                 hit.source_id  = source_id;
0076                 hit.sample_id  = sample;
0077                 hit.channel_id = channel;
0078                 hit.adc_value  = current_value;
0079             }
0080         }
0081         Set(m_sample_ptrs); // Copy all of the pointers into m_samples over in one go
0082     }
0083 };
0084
0085 #endif  // _ADCSampleFactory_h_