NeXusEventCallback.cpp

Go to the documentation of this file.

#include "stdafx.h"
#include "isiscrpt.h"
#include "isisraw.h"
#include "icputils.h"
#include "selogger.h"
#include "NeXusEventCallback.h"

#define PPP_TO_UAMPH    1.738E-6

const int NeXusEventCallback::m_nexus_chunk_length = 32000;
const int NeXusEventCallback::INVALID_FRAME_MARKER = -1;

Poco::Mutex NeXusEventCallback::m_frame_callback_mutex;
Poco::Mutex NeXusEventCallback::m_event_write_mutex;
Poco::Condition NeXusEventCallback::m_event_write_cond;


NeXusEventCallback::NeXusEventCallback(const ISISCRPT_STRUCT* crpt, IXNeXusFile* file, const std::vector<int>& ev_source_ids) : 
                        m_current_frame(-1), m_current_period(-1), m_period("period_number", crpt->start_time),m_vetoing("is_vetoing",crpt->start_time),
                        m_crpt(crpt), m_file(file), m_discarded_events(0),m_monitor_events_not_saved(0),m_good_frames(0),m_raw_frames(0),m_is_vetoing(-1),
                        m_mempool(4 * EventStore::maxNumCallbackEvents()),m_entry_name("raw_data_1"),
                        m_good_uamph_sum(0), m_raw_uamph_sum(0), m_good_events_sum(0),m_raw_events_sum(0),m_worker(m_queue),
                        m_last_frame_time(0),m_worker_thread("NeXusWorker"),m_refeventdetcard(-1)
{
        setLoggerName("NeXusEventCallback");
        m_refeventdetcard = getRefEventDetCard(ev_source_ids);
        m_worker_thread.start(m_worker);
        while( !m_queue.hasIdleThreads() )
        {
                Poco::Thread::sleep(100); // give thread time to start and connect to queue - avoids possible race condition in destructor?
        }
        BOOST_FOREACH(int id, ev_source_ids)
        {
                m_card_frame[id] = std::pair<int,time_t>(0,0); // Not INVALID_FRAME_MARKER as we should only have event enabled sources
        }
        int nspec = crpt->getNumSpectra(true); 
        m_det_id.resize(nspec);
        std::string det_name;
        det_name_t::left_const_iterator it;
        for(int i=0, j=0; i <nspec; ++i)
        {
                makeDetName(i, det_name);
                m_det_tr[det_name] = m_crpt->spectrumDAETR(i);
                if ( m_crpt->isEventSpectrum(i) )
                {
                        m_event_det.insert(det_name);
                }
                if ( (it = m_det_name.left.find(det_name)) == m_det_name.left.end() )
                {
                        it = m_det_name.left.insert(det_name_t::left_value_type(det_name, j++)).first;
                }
                m_det_id[i] = it->second;   // we are building a fast lookup of spectrum -> det_name  
        }
        if ( Poco::Util::Application::instance().config().getBool("isisicp.snseventfile", false) )
        {
                m_entry_name = "entry";
        }
}

int NeXusEventCallback::getRefEventDetCard(const std::vector<int>& ev_source_ids)
{
        int refeventdetcard = Poco::Util::Application::instance().config().getInt("isisicp.refeventdetcard", -1);
        if ( (refeventdetcard == -1) || (std::find(ev_source_ids.begin(), ev_source_ids.end(), refeventdetcard) != ev_source_ids.end()) )
        {
                return refeventdetcard;
        }
        if ( ev_source_ids.size() == 0 )  // histogram mode, no cards in event mode
        {
                LOGSTR_INFORMATION("No event sources, assuming histogram mode and setting refeventdetcard=-1");
                return -1;
        }
        static const int INVALID_CARD = std::numeric_limits<int>::max();
        int crate0_low = INVALID_CARD, crate1_low = INVALID_CARD;
        BOOST_FOREACH(int id, ev_source_ids)
        {
                if ( (id / 100 == 0) && (id < crate0_low) )    
                {
                        crate0_low = id;
                }
                if ( (id / 100 == 1) && (id < crate1_low) )    
                {
                        crate1_low = id;
                }
        }
        if (crate1_low != INVALID_CARD)
        {
                LOGSTR_WARNING("Using card number " << crate1_low << " as reference detecor card as " << refeventdetcard << " not present");
                refeventdetcard = crate1_low;
        }
        else if (crate0_low != INVALID_CARD)
        {
                LOGSTR_WARNING("Using card number  " << crate0_low << " as reference detecor card as " << refeventdetcard << " not present");
                refeventdetcard = crate0_low;
        }
        else
        {
                LOGSTR_ERROR("Reference event detector card number " << refeventdetcard << " not present and cannot find alternative");
                refeventdetcard = -1;
        }
        return refeventdetcard;
}

void NeXusEventCallback::finish()
{
        static bool random_shift = Poco::Util::Application::instance().config().getBool("isisicp.randomeventbinshift", true);
        close();
        LOGSTR_INFORMATION("GOOD uAh summed from events = " << m_good_uamph_sum * PPP_TO_UAMPH);
        LOGSTR_INFORMATION("RAW uAh summed from events = " << m_raw_uamph_sum * PPP_TO_UAMPH);
        LOGSTR_INFORMATION("GOOD events = " << m_good_events_sum);
        LOGSTR_INFORMATION("RAW events = " << m_raw_events_sum);
        LOGSTR_INFORMATION("DISCARDED events = " << m_discarded_events);
        LOGSTR_INFORMATION("Monitor events not saved = " << m_monitor_events_not_saved);
        LOGSTR_INFORMATION("GOOD / RAW frames from events = " << m_good_frames << " / " << m_raw_frames);
        int nframes = m_frame_number.size();
        LOGSTR_INFORMATION("Number of frames in event list = " << nframes);
        if ( (m_raw_events_sum > 0) && (nframes == 0) )
        {
                LOGSTR_ERROR("Events with no frames to write - something is very wrong. Last time we saw this was on WISH when the first crate generated events/frames but the second one didn't");
                return;
        }
//      for(std::map<int,int>::const_iterator it = m_first_data_frame.begin(); it != m_first_data_frame.end(); ++it)
//      {
//              LOGSTR_INFORMATION("Spectrum " << it->first << " first data frame " << it->second);
//      }
        int i, j, frame_no, nevents, n_no_events;
        std::vector<uint64_t> event_index(nframes);
        std::vector<int> event_frame_number(nframes);
        std::vector<double> event_frame_time(nframes);
        std::vector<int> frame_events(nframes, 0);
        for(std::set<std::string>::const_iterator it=m_event_det.begin(); it != m_event_det.end(); ++it)
        {
                const std::map<int,int>& this_det = m_frame_events_write[*it];
                n_no_events = 0;
                if (nframes > 0) // nframes == 0 if we have no events at all in event mode
                {
                        event_index[0] = 0;
                }
                for(i = 0; i < nframes; ++i)
                {
                        frame_no = m_frame_number[i];
                        event_frame_time[i] = static_cast<double>(m_frame_time[i]);
                        event_frame_number[i] = m_frame_number[i];
                        std::map<int,int>::const_iterator v = this_det.find(frame_no);
                        if ( v != this_det.end() )
                        {
                                nevents = v->second;
                        }
                        else
                        {
                                nevents = 0;
                                ++n_no_events;
                        }
                        if ( i < (nframes - 1) )
                        {
                                event_index[i+1] = event_index[i] + nevents;
                        }
                        frame_events[i] += nevents;
                }
                LOGSTR_INFORMATION("Inserted " << n_no_events << " zero event frames for " << *it);

                NXlink tof_link;
                memset(&tof_link, 0, sizeof(NXlink));
                if ( m_det_tr.find(*it) != m_det_tr.end() )
                {
                        m_file->openGroup("instrument", "NXinstrument");
                        m_file->openGroup("dae","IXdae");
                        m_file->openGroup(Poco::format("time_channels_%d", m_det_tr[*it]).c_str(), "IXtime_channels");
                        m_file->getLink("time_of_flight", tof_link);
                        m_file->closeGroup();
                        m_file->closeGroup();
                        m_file->closeGroup();
                }
                else
                {
                        LOGSTR_WARNING("Cannot find event_time_bins link for " << *it);
                }
                openDataGroup(*it, true);
                m_file->writeData("event_index", event_index);
                m_file->writeData("event_frame_number", event_frame_number);
                m_file->writeData("event_time_zero", event_frame_time);
                m_file->writeAttribute("event_time_zero", "offset", m_file->ISOtime(m_crpt->start_time));
                m_file->writeAttribute("event_time_zero", "units", "second");
                m_file->writeData("total_counts", m_events_write_total[*it]);
                if (strlen(tof_link.targetPath) > 0)
                {
                        m_file->makeLink("event_time_bins", tof_link);
                }
                if (random_shift)
                {
                        m_file->writeData("event_time_offset_shift", "random");
                }
                else
                {
                        m_file->writeData("event_time_offset_shift", "centre");
                }
                std::vector<NXlink> links(7); 
                m_file->getLink("event_index", links[0]);
                m_file->getLink("event_frame_number", links[1]);
                m_file->getLink("event_time_zero", links[2]);
                m_file->getLink("total_counts", links[3]);
                m_file->getLink("event_id", links[4]);
                m_file->getLink("event_time_offset", links[5]);
                m_file->getLink("event_time_offset_shift", links[6]);
                closeDataGroup(*it);
                // now make necessary links from NXdetector to NXevent_data etc.
                if (it->find(detector_base) != std::string::npos)
                {
                        m_file->makeNewGroup((*it+"_events"), "NXevent_data");
                        std::for_each(links.begin(), links.end(), boost::bind(&IXNeXusFile::makeLink, m_file, (const char*)NULL, _1));
                        if (strlen(tof_link.targetPath) > 0)
                        {
                                m_file->makeLink("event_time_bins", tof_link);
                        }
                        m_file->closeGroup();
                }
        }
        // update SE logs
        m_file->openGroup("framelog", "NXcollection");
        m_file->addLog("frame_log", m_frame_number, "frame_number", m_frame_time, "second", m_crpt->start_time);
        m_file->addLog("period_log", m_frame_period, "period_number", m_frame_time, "second", m_crpt->start_time);
        m_file->addLog("good_frame_log", m_good_frame, "is_good", m_frame_time, "second", m_crpt->start_time);
        m_file->addLog("proton_charge", m_frame_proton_charge, "uAh", m_frame_time, "second", m_crpt->start_time);
        m_file->addLog("events_log", frame_events, "events", m_frame_time, "second", m_crpt->start_time);
        for(i = 0; i < nframes; ++i)
        {
                frame_events[i] = m_frame_events_raw[i];
        }
        m_file->addLog("raw_events_log", frame_events, "events", m_frame_time, "second", m_crpt->start_time);
        const char* units;
        for(uint_map_t::const_iterator it = m_se_uint_values.begin(); it != m_se_uint_values.end(); ++it)
        {
                units = "unknown";
                for(int j=0; j<m_crpt->n_data_dae; ++j)
                {
                        if (it->first == m_crpt->data_dae[j].name)
                        {
                                units = m_crpt->data_dae[j].scaled_units;
                        }
                        else if ( it->first == (std::string(m_crpt->data_dae[j].name)+"_raw") )
                        {
                                units = m_crpt->data_dae[j].raw_units;
                        }
                }
                m_file->addLog(it->first.c_str(), it->second, units, m_se_times[it->first], "second", m_crpt->start_time);
        }
        for(float_map_t::const_iterator it = m_se_float_values.begin(); it != m_se_float_values.end(); ++it)
        {
                units = "unknown";
                for(int j=0; j<m_crpt->n_data_dae; ++j)
                {
                        if (it->first == m_crpt->data_dae[j].name)
                        {
                                units = m_crpt->data_dae[j].scaled_units;
                        }
                        else if ( it->first == (std::string(m_crpt->data_dae[j].name)+"_raw") )
                        {
                                units = m_crpt->data_dae[j].raw_units;
                        }
                }
                m_file->addLog(it->first.c_str(), it->second, units, m_se_times[it->first], "second", m_crpt->start_time);
        }
        m_file->closeGroup();

        m_file->openGroup("runlog", "IXrunlog");
        m_period.setOffset(m_crpt->start_time);   // update offset as may have changed since start of run
        m_period.write(m_file, "period_change_log");
        m_vetoing.setOffset(m_crpt->start_time);  // update offset as may have changed since start of run
        m_vetoing.write(m_file, "veto_log");
        m_file->closeGroup();

        m_file->writeData("total_counts", sum_map_values(m_events_write_total));
        m_file->writeData("total_uncounted_counts", m_discarded_events);
        m_file->writeData("monitor_events_not_saved", m_monitor_events_not_saved);      
}

void NeXusEventCallback::close()
{
        for(det_file_t::const_iterator det_pos = m_det_file.begin(); det_pos != m_det_file.end(); ++det_pos)
        {
                BOOST_FOREACH(IXNeXusFile* nx, det_pos->second)
                {
                        nx->closeData();
                        closeDataGroup(nx, det_pos->first);
                        nx->close();
                        delete nx;
                }
        }
        m_det_file.clear();
}

void NeXusEventCallback::flush()
{
        for(det_file_t::const_iterator det_pos = m_det_file.begin(); det_pos != m_det_file.end(); ++det_pos)
        {
                BOOST_FOREACH(IXNeXusFile* nx, det_pos->second)
                {
                        nx->flush();
                }
        }
        m_file->flush();
}

NeXusEventCallback::~NeXusEventCallback()
{
        close();
        Poco::Thread::sleep(100);
        while( !m_queue.empty() )
        {
                Poco::Thread::sleep(100);
        }
        m_queue.wakeUpAll();  // gets m_worker to finish
        Poco::Thread::sleep(100);
        // we have had lots of trouble here! It seems that wakeUpAll() can get ignored some times and we end up with a waiting thread
        // so try a bit of defensive programming...
        if ( m_queue.hasIdleThreads() )
        {
                LOGSTR_INFORMATION("Retrying wakeUpAll()");
                while( m_queue.hasIdleThreads() )
                {
                        m_queue.wakeUpAll();  // gets m_worker to finish
                        Poco::Thread::sleep(100);
                }
        }
        m_worker_thread.join();
}

void NeXusEventCallback::makeDetName(int spec, std::string& det_name)
{
        int mon_num = m_crpt->spectrumToMonitorNumber(spec);
        if (mon_num > 0)
        {
                det_name = Poco::format("monitor_%d", mon_num);
        }
        else
        {
                det_name = Poco::format(detector_format, m_crpt->spectrumGroup(spec));
        }
}


void NeXusEventCallback::createEmptyFileStructure()
{
        int j;
        m_file->makeNewGroup("raw_data_1", "NXentry");
        m_file->makeNewGroup("instrument", "NXinstrument");
        m_file->makeNewGroup("dae","IXdae");

//      std::set<int> tr_used;
//      for(std::map<std::string,int>::const_iterator it = m_det_tr.begin(); it != m_det_tr.end(); ++it)
//      {
//              tr_used.insert(it->second);
//      }
//      BOOST_FOREACH(int tr, tr_used)
        for(int tr=1; tr <= m_crpt->ntrg; ++tr)
        {
                j = tr - 1;
                m_file->makeNewGroup(Poco::format("time_channels_%d", tr).c_str(), "IXtime_channels");
                m_file->writeData("time_of_flight", &(m_crpt->rtcb[j][0]), 1 + m_crpt->ntc[j]);
                m_file->writeAttribute("time_of_flight", "axis", 1);
                m_file->writeAttribute("time_of_flight", "primary", 1);
                m_file->writeAttribute("time_of_flight", "units", "microsecond");
                m_file->writeData("time_of_flight_raw", &(m_crpt->tcb[j][0]), 1 + m_crpt->ntc[j]);
                m_file->writeAttribute("time_of_flight_raw", "units", "pulses");
                m_file->writeAttribute("time_of_flight_raw", "frequency", "32 MHz");
                m_file->closeGroup();
        }
        m_file->closeGroup();
        m_file->closeGroup();
        for(det_name_t::const_iterator it = m_det_name.begin(); it != m_det_name.end(); ++it)
        {
                const std::string& name = it->left;
                if (name.substr(0, 8) == "monitor_")
                {
                        m_file->makeNewGroup(name, "NXmonitor", false);
                }
                else
                {
                        m_file->openGroup("instrument", "NXinstrument");
                        m_file->makeNewGroup(name, "NXdetector", false);
                        m_file->closeGroup();
                        m_file->makeNewGroup(name, "NXdata", false);
                }
        }
        m_file->makeNewGroup("runlog", "IXrunlog", false);
}


void NeXusEventCallback::processDataDae(const DAEEventHeader* head, const DetectorEvent32* det_ev, int n, int event_source_id)
{
        static const uint32_t SIX_MONTHS_IN_SECS = 6 * 28 * 24 * 60 * 60; 
        int i, ndiscarded = 0, device_id, device_index, nignored = 0;
        union { uint64_t u64; double d; uint32_t u32; float f; unsigned char b[8]; } data_value;
        float offset_time = -1.0;
        DAEEventList::DAETimeToOffset(head->time, m_crpt->start_time, offset_time);
        std::ostringstream msgpfx_oss;
        msgpfx_oss << "(Time: " << offset_time << ", Frame: " << head->frame_number << ") ";
        const std::string& msgpfx = msgpfx_oss.str();
        bool ignore_value;
        for(i=0; i<n; ++i)
        {
                ignore_value = false;
                if (head->info.bad_frame || det_ev[i].time_channel == 0 )   // ignore bad and out of time range frames
                {
                        ++ndiscarded;
                        InterlockedIncrement(&m_discarded_events);
                        continue;
                }
                if ( (det_ev[i].spectrum >> 8) != 0 )
                {
                        LOGSTR_ERROR(msgpfx << "Expected device id packet, but found header: " << (det_ev[i].spectrum >> 8) << " value: "  << (det_ev[i].spectrum & 0xff));
                        continue;
                }
                device_id = det_ev[i].spectrum & 0xff;
                device_index = -1;
                for(int j=0; j<m_crpt->n_data_dae; ++j)
                {
                        if (m_crpt->data_dae[j].id == device_id)
                        {
                                device_index = j;
                        }
                }
                if (device_index == -1)
                {
                        LOGSTR_ERROR(msgpfx << "Invalid device id (not specified in XML file) " << device_id);
                        continue;
                }
                const data_dae_t& data_dae = m_crpt->data_dae[device_index];
                if (data_dae.data_bytes != 4)
                {
                        LOGSTR_ERROR(msgpfx << "Invalid data size " << data_dae.data_bytes << " for device id " << device_id);
                        i += data_dae.data_bytes;
                        continue;
                }
                if (data_dae.data_type != 1 && data_dae.data_type != 2)
                {
                        LOGSTR_ERROR(msgpfx << "Invalid data type " << data_dae.data_type << " for device id " << device_id);
                        i += data_dae.data_bytes;
                        continue;
                }
                for(int j=1; j <= data_dae.data_bytes; ++j)
                {
                        if ( (det_ev[i+j].spectrum >> 8) != j )
                        {
                                LOGSTR_WARNING(msgpfx << "invalid data byte " << j << " for device id " << device_id 
                                        << " header: " << (det_ev[i+j].spectrum >> 8) << " value: " << (det_ev[i+j].spectrum & 0xff) );
                                ignore_value = true;
                        }
                        data_value.b[j-1] = det_ev[i+j].spectrum & 0xff;
                }
                int dae_tr = m_crpt->daeTR(data_dae.tr);
                if (offset_time < 0.0 || offset_time > SIX_MONTHS_IN_SECS)  // no run should last more than this time
                {
                        LOGSTR_WARNING("Invalid SE DAE time value " << head->time << "(offset " << offset_time << " seconds) for frame " << m_current_frame);   // error
                        ignore_value = true;
                }
                if (!ignore_value)
                {
                        if (data_dae.data_type == 1)
                        {
                                m_se_uint_values[std::string(data_dae.name)+"_raw"].push_back(data_value.u32);
                                m_se_float_values[data_dae.name].push_back(data_value.u32 * data_dae.scale + data_dae.offset);
                        }
                        else
                        {
                                m_se_float_values[std::string(data_dae.name)+"_raw"].push_back(data_value.f);
                                m_se_float_values[data_dae.name].push_back(data_value.f * data_dae.scale + data_dae.offset);
                        }
                        m_se_times[std::string(data_dae.name)+"_raw"].push_back(offset_time + getTimeOffset(det_ev, event_source_id, dae_tr, i) / 1e6); // time of first value
                        m_se_times[data_dae.name].push_back(offset_time + getTimeOffset(det_ev, event_source_id, dae_tr, i) / 1e6); // time of first value
                }
                else
                {
                        ++nignored;
                }
                i += data_dae.data_bytes;
        }
        if (i != n)
        {
                LOGSTR_ERROR(msgpfx << "SE Data count used mismatch " << i << " != " << n);
        }
        if (nignored > 0)
        {
                LOGSTR_WARNING("Ignored " << nignored << " SE data to dae values");
        }
}


void NeXusEventCallback::allEventCallback(const DAEEventHeader* head, const DetectorEvent32* det_ev, int n, int event_source_id, const int* mapping)
{
        if (m_mempool.blockSize() < 4 * n)
        {
                throw std::exception("NeXusEventCallback::allEventCallback");
        }
        if (n == 0)
        {
                updateCurrentWriteFrame(head->frame_number, event_source_id);
                return;
        }
        // all spectra from a card (event_source_id) must have the same DAE TR, however they may have a different CRPT TR
        // the DAE TR refers to the real event time as proframmed in the detector card, the CRPT TR is the time regime that
        // was used for on the fly rebinning and is also used for naming the NXdetector objects
        int spec, dae_tr = -1, dae_tr0 = -1;
        for(int i=0; i<n; ++i)
        {
                spec = getSpectrum(det_ev, event_source_id, i);
                if (spec == 0)
                {
                        dae_tr0 = m_crpt->spectrumDAETR(spec);
                }
                else if (spec > 0)    // we may have corrupt values from card (-1); also we want to avoid spectrum 0 as this is possibly in a different time regime
                {
                        dae_tr = m_crpt->spectrumDAETR(spec);
                        break;
                }
        }
        if (dae_tr == -1 && dae_tr0 != -1)
        {
                dae_tr = dae_tr0;
        }
        if (dae_tr == -1)
        {
                for(int i=0; i < m_crpt->n_data_dae; ++i)
                {
                        if (m_crpt->data_dae[i].crate == event_source_id)
                        {
                                processDataDae(head, det_ev, n, event_source_id);
                                return;
                        }
                }
                m_logstr->warning() << "Ignoring " << n << " events from frame "<< head->frame_number << " event source " <<  event_source_id << " as all spectra are invalid";
                return;
        }
//      std::string current_det_name;
//      int_map_t event_id;
//      float_map_t event_time_offset;
        int_map_t& event_id = *m_int_map_store.get();
        float_map_t& event_time_offset = *m_float_map_store.get();
        float_map_t::iterator event_time_it;
        int_map_t::iterator event_id_it;
        int mon_num, det_id, current_det_id = -1, nmonskippedevents = 0, ndiscarded = 0;
        for(int i=0; i<n; ++i)
        {
                spec = getSpectrum(det_ev, event_source_id, i);
                mon_num = m_crpt->spectrumToMonitorNumber(spec);
                if ( (mon_num > 0) && !m_crpt->saveMonitorEvents(mon_num) )
                {
                        ++nmonskippedevents;
#ifdef _WIN64
                        InterlockedIncrement64(&m_monitor_events_not_saved);
#endif /* _WIN64 */
                        continue;
                }
                if (head->info.bad_frame || det_ev[i].time_channel == 0 || spec == -1)   // ignore bad and out of time range frames
                {
                        ++ndiscarded;
                        InterlockedIncrement(&m_discarded_events);
                }
                else
                {
                        det_id = m_det_id[spec];
                        if (current_det_id != det_id)
                        {
                                const std::string& det_name = m_det_name.right.at(det_id);
                                if ( (event_time_it = event_time_offset.find(det_name)) == event_time_offset.end() )
                                {
                                        event_time_offset[det_name];
                                        event_time_it = event_time_offset.find(det_name);
                                }
                                if (event_time_it->second.capacity() < n)
                                {
                                        event_time_it->second.reserve(n);
                                }
                                if ( (event_id_it = event_id.find(det_name)) == event_id.end() )
                                {
                                        event_id[det_name];
                                        event_id_it = event_id.find(det_name);
                                }
                                if (event_id_it->second.capacity() < n)
                                {
                                        event_id_it->second.reserve(n);
                                }
                                current_det_id = det_id;
                        }
                        event_time_it->second.push_back(getTimeOffset(det_ev, event_source_id, dae_tr, i));
//                      event_id_it->second.push_back(m_crpt->udetFromSpec(spec));  ///< @todo for SNS compatibility - conver to detector ID
                        event_id_it->second.push_back(spec); 
                }
        }
        writeEvents(head, event_source_id, n - nmonskippedevents /* - ndiscarded */, event_id, event_time_offset);
        for(int_map_t::iterator it = event_id.begin(); it != event_id.end(); ++it)
        {
                it->second.resize(0);
        }
        m_int_map_store.put(&event_id);
        for(float_map_t::iterator it = event_time_offset.begin(); it != event_time_offset.end(); ++it)
        {
                it->second.resize(0);
        }
        m_float_map_store.put(&event_time_offset);
}

void NeXusEventCallback::allFrameCallback(const DAEEventHeader* head, int event_source_id)
{
        Poco::Mutex::ScopedLock _lock(m_frame_callback_mutex);
        static bool warneventpppmismatch = Poco::Util::Application::instance().config().getBool("isisicp.warneventpppmismatch", false);
        static const uint32_t MAX_ISIS_PPP = static_cast<uint32_t>( 0.5 + (300.0 * 3600.0) / (PPP_TO_UAMPH * 50.0) ); 
        static const uint32_t SIX_MONTHS_IN_SECS = 6 * 28 * 24 * 60 * 60; 
        uint32_t the_protons = head->protons;
        m_frame_events_raw[static_cast<int>(head->frame_number)] += head->num_events;
        if (m_refeventdetcard != -1 && m_refeventdetcard != event_source_id)
        {
                return; // don't trust the information
        }
        if ( warneventpppmismatch && (static_cast<int>(head->frame_number) == m_current_frame) )
        {
                unsigned ref_protons = static_cast<unsigned>(0.5 + m_frame_proton_charge.back() / PPP_TO_UAMPH);
                if ( head->protons != ref_protons )
                {
                        LOGSTR_WARNING("proton mismatch " << head->protons << " != " << ref_protons);  
                }
                float ref_time = m_frame_time.back();
                float this_time;
                DAEEventList::DAETimeToOffset(head->time, m_crpt->start_time, this_time);
                if (this_time != ref_time)
                {
                        LOGSTR_WARNING("absolute time mismatch " << this_time << " != " << ref_time);  
                }
        }
        if ( static_cast<int>(head->frame_number) <= m_current_frame )
        {
                return;   // we have already seen this frame from another source
        }
        if ( (head->frame_number - m_current_frame) > 1 )  // we should only jump by one frame at a time;
        {
                LOGSTR_WARNING("Jumping more whan one frame " << head->frame_number << " -> " << m_current_frame);   // error
        }
        m_current_frame = head->frame_number;
        float previous_offset_time = 0.0;
        uint32_t previous_protons = 0;
        if (m_frame_proton_charge.size() > 0)
        {
                previous_protons = static_cast<uint32_t>(0.5 + m_frame_proton_charge.back() / PPP_TO_UAMPH);
        }
        if (m_frame_time.size() > 0)
        {
                previous_offset_time = m_frame_time.back();
        }
        if (the_protons > MAX_ISIS_PPP)
        {
                LOGSTR_WARNING("Invalid DAE PPP value " << the_protons << " (" << the_protons * PPP_TO_UAMPH << " uAh) for frame " << m_current_frame);   // error
//              the_protons = previous_protons;   
        }
        float offset_time = -1.0;
        DAEEventList::DAETimeToOffset(head->time, m_crpt->start_time, offset_time);
        if (offset_time < 0.0 || offset_time > SIX_MONTHS_IN_SECS)  // no run should last more than this time
        {
                LOGSTR_WARNING("Invalid DAE time value " << head->time << "(offset " << offset_time << " seconds) for frame " << m_current_frame);   // error
//              offset_time = previous_offset_time + 0.01; // add 1/100 second, half of a 50Hz frame
        }
        if ( m_current_period != head->period )
        {
                m_current_period = head->period;
                m_period.addValue(offset_time, m_current_period + 1);
        }
        if ( m_is_vetoing != head->info.bad_frame )
        {
                m_is_vetoing = head->info.bad_frame;
                m_vetoing.addValue(offset_time, head->info.bad_frame);
        }
        ++m_raw_frames;
        m_raw_uamph_sum += the_protons;
        if (!head->info.bad_frame)
        {
                ++m_good_frames;
                m_good_uamph_sum += the_protons;
        }
        m_frame_period.push_back(m_current_period + 1);
        m_frame_time.push_back(offset_time);
        m_frame_number.push_back(m_current_frame);
        m_frame_proton_charge.push_back(the_protons * PPP_TO_UAMPH);
        m_good_frame.push_back(!head->info.bad_frame);
//      head->info.bad_frame;
//      head->vetos;
//      std::cerr << "Frame " << head->frame_number << " time " << DAEEventList::DAETimeAsString(head->time) << std::endl;
}

void NeXusEventCallback::updateCurrentWriteFrame(int frame, int event_source_id)
{
        Poco::Mutex::ScopedLock _lock(m_event_write_mutex);
        int nvalid = 0;
        for(card_frame_t::const_iterator cit = m_card_frame.begin(); cit != m_card_frame.end(); ++cit)
        {
                if (cit->second.first != INVALID_FRAME_MARKER)
                {
                        ++nvalid;
                        break;  // only interested in nvalid != 0
                }
        }
        card_frame_t::iterator it;
        time_t ref_time = time(NULL);
        m_last_frame_time = ref_time;
        if (nvalid == 0)  // we are the first card to call, other cards should be at this frame number or less so set this to make sure we do not lose them
        {
                for(it = m_card_frame.begin(); it != m_card_frame.end(); ++it)
                {
                        it->second.first = frame;
                        it->second.second = ref_time;
                }
                m_event_write_cond.broadcast();
        }
        else
        {
                it = m_card_frame.find(event_source_id);
                if ( it != m_card_frame.end() )
                {
                        it->second.first = frame;
                        it->second.second = ref_time;
                        m_event_write_cond.broadcast();
                }
                else
                {
                        LOGSTR_WARNING("Data from invalid source id " << event_source_id << " in frame " << frame);
                }
        }
}

// this will get called when we have no data in the card, which when reading from a file means we have finished
void NeXusEventCallback::noFrameCallbackEnd(bool end_present, NoFrameCallbackMode mode, int event_source_id)
{
        card_frame_t::const_iterator it = m_card_frame.find(event_source_id);
        if ( it != m_card_frame.end() )
        {
                if ( it->first != INVALID_FRAME_MARKER )
                {
                        updateCurrentWriteFrame(INVALID_FRAME_MARKER, event_source_id);
                }
        }
        else
        {
                LOGSTR_WARNING("Data from invalid source id " << event_source_id);
        }
}

// this will get called when we have no data in the card, which when reading from the real DAE could be just beam off
// what we haven't covered here is if a detector card failed and stopped returning headers at all 
void NeXusEventCallback::noFrameCallbackNull(bool end_present, NoFrameCallbackMode mode, int event_source_id)
{
        card_frame_t::iterator it = m_card_frame.find(event_source_id);
        if ( it != m_card_frame.end() )
        {
                if ( mode != NoFrameRunning )
                {
                        noFrameCallbackEnd(end_present, mode, event_source_id);
                }
        }
        else
        {
                LOGSTR_WARNING("Data from invalid source id " << event_source_id);
        }
}

int NeXusEventCallback::currentWriteFrame() const
{
        static const int INITIAL_VAL = std::numeric_limits<int>::max();
        int current_frame = INITIAL_VAL;
        for(card_frame_t::const_iterator it = m_card_frame.begin(); it != m_card_frame.end(); ++it)
        {
                if (it->second.first != INVALID_FRAME_MARKER)
                {
                        current_frame = std::min(current_frame, it->second.first);
                }
        }
        if (current_frame == INITIAL_VAL)
        {
                current_frame = 0;
        }
        return current_frame;
        // this should work but doesn't - a std::copy with value_iterator gives values, but std::min_element doesn't work
        // return *(std::min_element(value_iterator<int,int>(m_card_frame.begin()), value_iterator<int,int>(m_card_frame.end())));
}

void NeXusEventCallback::writeEvents(const DAEEventHeader* head, int event_source_id, int nev, const int_map_t& event_id, const float_map_t& event_time_offset)
{
        static bool single_nexus_writer_thread = Poco::Util::Application::instance().config().getBool("isisicp.singlenexuswriterthread", true);
        Poco::Mutex::ScopedLock _lock(m_event_write_mutex);
        updateCurrentWriteFrame(head->frame_number, event_source_id);
        int current_frame = currentWriteFrame();
        while(head->frame_number > current_frame)
        {
                //m_event_write_cond.tryWait(m_event_write_mutex, 10000);     // we should be able to just use wait(), but this covers us if a broadcast is not received 
                m_event_write_cond.wait(m_event_write_mutex); 
                current_frame = currentWriteFrame();
        }
        if (head->frame_number < current_frame)
        {
                LOGSTR_WARNING("Writing data arriving too late: passed frame " << head->frame_number << " when current is " << current_frame << " for source " << event_source_id); 
        }
        if (single_nexus_writer_thread)
        {
                Poco::Event ev;
                NeXusWorkerData d(this, ev, head, event_source_id, nev, event_id, event_time_offset);
                m_queue.enqueueNotification(new GenericWorkNotification<NeXusWorkerData>(&d));
                ev.wait();
        }
        else
        {
                writeEventsImpl(head, event_source_id, nev, event_id, event_time_offset);
        }
}

void NeXusEventCallback::writeEventsImpl(const DAEEventHeader* head, int event_source_id, int nev, const int_map_t& event_id, const float_map_t& event_time_offset)
{
        int current_frame = currentWriteFrame();
        poco_assert(head->frame_number == current_frame);
        m_raw_events_sum += nev;
        std::string current_group;
        det_file_t::iterator det_pos;
        uint64_t *group_total_events = NULL;
        int *group_frame_events = NULL;
        int64_t dims_array[1], slab_start[1], dest_start[1];
        const float_map_t::mapped_type* event_times = NULL;
        for(int_map_t::const_iterator it = event_id.begin(); it != event_id.end(); ++it)
        {
                if (it->second.size() == 0)
                {
                        continue;
                }
                if (current_group != it->first)
                {
                        current_group = it->first;
                        det_pos = m_det_file.find(current_group);
                        if (det_pos == m_det_file.end())
                        {
                                std::vector<IXNeXusFile*> det_tmp(2, NULL);
                                BOOST_FOREACH(IXNeXusFile*& det_nx, det_tmp)
                                {
                                        det_nx = m_file->clone();
                                        det_nx->openGroup(m_entry_name.c_str(), "NXentry");
                                        openDataGroup(det_nx, current_group);
                                        group_total_events = &(m_events_write_total[current_group]);  // this is here as openDataGroup check its to know whether to create group
                                }
                                det_tmp[0]->openData("event_id");
                                det_tmp[1]->openData("event_time_offset");
                                m_det_file[current_group] = det_tmp;
                                det_pos = m_det_file.find(current_group);
                        }
                        group_total_events = &(m_events_write_total[current_group]);
                        group_frame_events = &(m_frame_events_write[current_group][current_frame]);
                        event_times = &(event_time_offset.at(current_group));
                }
                std::vector<IXNeXusFile*>& det_nx_v = det_pos->second;
                (*group_frame_events) += it->second.size();
                // write events
                dims_array[0] = it->second.size();
                slab_start[0] = 0;
                dest_start[0] = *group_total_events;
                det_nx_v[0]->addSlab(it->second.data(), dims_array, 1, slab_start, dims_array, dest_start);
                det_nx_v[1]->addSlab(event_times->data(), dims_array, 1, slab_start, dims_array, dest_start);
                (*group_total_events) += it->second.size();
                if (!head->info.bad_frame)
                {
                        m_good_events_sum += it->second.size();
                }
//              for(int i=0; i<it->second.size(); ++i)
//              {
//                      if ( m_first_data_frame.find(it->second[i]) == m_first_data_frame.end() )
//                      {
//                              m_first_data_frame[it->second[i]] = current_frame;
//                      }
//              }
        }
}

void NeXusEventCallback::openDataGroup(IXNeXusFile* nx, const std::string& name, bool create_zero_size_if_missing)
{
        int dims_array[1];
        int chunks_array[1] = { m_nexus_chunk_length };
        bool is_monitor = (name.find("monitor") != std::string::npos);
        if (is_monitor)
        {
                nx->openGroup(name.c_str(), "NXmonitor");
        }
        else
        {
                nx->openGroup("instrument", "NXinstrument");
                nx->openGroup(name.c_str(), "NXdetector");
        }
        if (m_events_write_total.find(name) == m_events_write_total.end())  // create data items if they do not exist already
        {
                if (create_zero_size_if_missing) // we are writing an empty event structure
                {
                        dims_array[0] = 0;
                    nx->makeData("event_id", NX_UINT32, dims_array, 1);
                    nx->makeData("event_time_offset", NX_FLOAT32, dims_array, 1);
                }
                else
                {
                        dims_array[0] = NX_UNLIMITED;
                    nx->makeDataChunked("event_id", NX_UINT32, dims_array, chunks_array, 1);
                    nx->makeDataChunked("event_time_offset", NX_FLOAT32, dims_array, chunks_array, 1);
                }
                nx->writeAttribute("event_time_offset", "units", "microsecond");
                nx->flush();
        }
}

void NeXusEventCallback::closeDataGroup(IXNeXusFile* nx, const std::string& name)
{
        if (name.size() == 0)
        {
                return;
        }
        bool is_monitor = (name.find("monitor") != std::string::npos);
        nx->closeGroup();
        if (!is_monitor)   // detectors are one level lower than monitors
        {
                nx->closeGroup();
        }
}