DDDigi/src/DigiOutputAction.cpp

0001 //==========================================================================
0002 //  AIDA Detector description implementation
0003 //--------------------------------------------------------------------------
0004 // Copyright (C) Organisation europeenne pour la Recherche nucleaire (CERN)
0005 // All rights reserved.
0006 //
0007 // For the licensing terms see $DD4hepINSTALL/LICENSE.
0008 // For the list of contributors see $DD4hepINSTALL/doc/CREDITS.
0009 //
0010 // Author     : M.Frank
0011 //
0012 //==========================================================================
0013
0014 // Framework include files
0015 #include <DD4hep/InstanceCount.h>
0016 #include <DDDigi/DigiOutputAction.h>
0017 #include <DDDigi/DigiContext.h>
0018 #include <DDDigi/DigiPlugins.h>
0019 #include <DDDigi/DigiKernel.h>
0020
0021 // C/C++ include files
0022 #include <stdexcept>
0023
0024 using namespace dd4hep::digi;
0025
0026 /// Standard constructor
0027 DigiOutputAction::DigiOutputAction(const DigiKernel& kernel, const std::string& nam)
0028   : DigiContainerSequenceAction(kernel, nam)
0029 {
0030   declareProperty("sequence_streams", m_sequence_streams);
0031   declareProperty("processor_type", m_processor_type);
0032   declareProperty("containers",     m_containers);
0033   declareProperty("output",         m_output);
0034   InstanceCount::increment(this);
0035 }
0036
0037 /// Default destructor
0038 DigiOutputAction::~DigiOutputAction()   {
0039   InstanceCount::decrement(this);
0040 }
0041
0042 /// Create new output stream name
0043 std::string DigiOutputAction::next_stream_name()   {
0044   if ( m_sequence_streams )   {
0045     std::size_t idx = m_output.rfind(".");
0046     std::stringstream str;
0047     if ( idx != std::string::npos )  {
0048       std::string fname = m_output.substr(0, idx);
0049       std::string ftype = m_output.substr(idx+1);
0050       str << fname << '_' << std::dec << std::setfill('0') << std::setw(8) << fseq_count << "." << ftype;
0051       return str.str();
0052     }
0053     str << m_output << '_' << std::dec << std::setfill('0') << std::setw(8) << fseq_count;
0054     return str.str();
0055   }
0056   return m_output;
0057 }
0058 /// Initialization callback
0059 void DigiOutputAction::initialize()   {
0060   if ( m_containers.empty() )   {
0061     warning("+++ No input containers given for attenuation action -- no action taken");
0062     return;
0063   }
0064   num_events = m_kernel.property("numEvents").value<long>();
0065   for ( const auto& c : m_containers )   {
0066     Key key(c.first, 0, 0);
0067     auto it = m_registered_processors.find(key);
0068     if ( it == m_registered_processors.end() )   {
0069       std::string nam = name() + ".E4H." + c.first;
0070       auto* conv = createAction<DigiContainerProcessor>(m_processor_type, m_kernel, nam);
0071       if ( !conv )   {
0072     except("+++ Failed to create edm4hep processor: %s of type: %s",
0073            nam.c_str(), m_processor_type.c_str());
0074       }
0075       conv->property("OutputLevel").set(int(outputLevel()));
0076       adopt_processor(conv, c.first);
0077       conv->release(); // Release processor **after** adoption.
0078     }
0079   }
0080   std::lock_guard<std::mutex> lock(m_kernel.global_io_lock());
0081   this->DigiContainerSequenceAction::initialize();
0082 }
0083
0084 /// Finalization callback
0085 void DigiOutputAction::finalize()   {
0086   close_output();
0087   this->DigiContainerSequenceAction::finalize();
0088 }
0089
0090 /// Adopt new parallel worker
0091 void DigiOutputAction::adopt_processor(DigiContainerProcessor* action,
0092                        const std::string& container)
0093 {
0094   std::size_t idx = container.find('/');
0095   if ( idx != std::string::npos )   {
0096     std::string nam = container.substr(0, idx);
0097     std::string typ = container.substr(idx+1);
0098     this->DigiContainerSequenceAction::adopt_processor(action, nam);
0099     m_containers.emplace(nam, typ);
0100     return;
0101   }
0102   except("+++ Invalid container specification: %s. %s",
0103      container.c_str(), "Specify container as tuple: \"<name>/<type>\" !");
0104 }
0105
0106 /// Adopt new parallel worker acting on multiple containers
0107 void DigiOutputAction::adopt_processor(DigiContainerProcessor* action,
0108                        const std::vector<std::string>& containers)
0109 {
0110   DigiContainerSequenceAction::adopt_processor(action, containers);
0111 }
0112
0113 /// Pre-track action callback
0114 void DigiOutputAction::execute(DigiContext& context)  const   {
0115   std::lock_guard<std::mutex> lock(context.global_io_lock());
0116   /// Check for valid output stream. If not: open new stream
0117   if ( !have_output() )   {
0118     open_output();
0119   }
0120   this->DigiContainerSequenceAction::execute(context);
0121   /// Commit data. Close stream if necessary
0122   commit_output();
0123 }