DFApplication.cpp

/**
 * @file DFApplication.cpp
 *
 * Implementation of DFApplication's generate_modules dal method
 *
 * This is part of the DUNE DAQ Software Suite, copyright 2023.
 * Licensing/copyright details are in the COPYING file that you should have
 * received with this code.
 */


#include "ConfigObjectFactory.hpp"
#include "appmodel/DFApplication.hpp"
#include "appmodel/CTBApplication.hpp"
#include "appmodel/DataStoreConf.hpp"
#include "appmodel/DataWriterConf.hpp"
#include "appmodel/DataWriterModule.hpp"
#include "appmodel/FakeDataApplication.hpp"
#include "appmodel/FakeDataProdConf.hpp"
#include "appmodel/FilenameParams.hpp"
#include "appmodel/NetworkConnectionDescriptor.hpp"
#include "appmodel/NetworkConnectionRule.hpp"
#include "appmodel/QueueConnectionRule.hpp"
#include "appmodel/QueueDescriptor.hpp"
#include "appmodel/ReadoutApplication.hpp"
#include "appmodel/SourceIDConf.hpp"
#include "appmodel/TPStreamConf.hpp"
#include "appmodel/TRBConf.hpp"
#include "appmodel/TRBModule.hpp"
#include "appmodel/TPReplayModuleConf.hpp"
#include "appmodel/TPReplayApplication.hpp"
#include "appmodel/appmodelIssues.hpp"

#include "conffwk/Configuration.hpp"

#include "confmodel/Connection.hpp"
#include "confmodel/DetectorStream.hpp"
#include "confmodel/DetectorToDaqConnection.hpp"
#include "confmodel/NetworkConnection.hpp"
#include "confmodel/Service.hpp"

#include "logging/Logging.hpp"
#include "oks/kernel.hpp"

#include <fmt/core.h>
#include <string>
#include <vector>

namespace dunedaq {
namespace appmodel {


inline void
fill_sourceid_object_from_app(const SmartDaqApplication* smartapp,
                              const conffwk::ConfigObject* netConn,
                              conffwk::ConfigObject& sidNetObj)
{
  sidNetObj.set_obj("netconn", netConn);
  sidNetObj.set_objs("source_ids", { &smartapp->get_source_id()->config_object() });
}


inline void
fill_sourceid_object_from_app(const ConfigObjectFactory& obj_fac,
                              const TPReplayApplication* rapp,
                              std::vector<conffwk::ConfigObject>* netConn,
                              std::vector<conffwk::ConfigObject>* sidNetObj,
			      std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs,
			      const NetworkConnectionDescriptor* descriptor,
                              std::string smartapp_uid)
{
  std::vector<const conffwk::ConfigObject*> source_id_objs;

  for (auto tp_sid : rapp->get_tp_source_ids()) {
    // get name extension
    std::string name = tp_sid->UID();
    size_t pos = name.find_last_of('-');
    std::string ext;
    if (pos != std::string::npos) {
      ext = name.substr(pos);
    }

    // set Network connections
    std::string dreqNetUid(smartapp_uid + ext);
    netConn->emplace_back(
      obj_fac.create_net_obj(descriptor, dreqNetUid));
    netConn->back().set_by_val<std::string>("data_type", descriptor->get_data_type());
    netConn->back().set_by_val<std::string>("connection_type", descriptor->get_connection_type());
    auto serviceObj = descriptor->get_associated_service()->config_object();
    netConn->back().set_obj("associated_service", &serviceObj);

    // set SourceID to Network connections
    std::string sidToNetUid(smartapp_uid + ext + "-sids");
    sidNetObj->emplace_back(
      obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
    sidNetObj->back().set_obj("netconn", &netConn->back());

    // set SourceID objs
    sidObjs.push_back(std::make_shared<conffwk::ConfigObject>(tp_sid->config_object()));
    sidNetObj->back().set_objs("source_ids", { sidObjs.back().get() });
  }
}
    
inline void
fill_sourceid_object_from_app(const ConfigObjectFactory& obj_fac,
                              const ReadoutApplication* roapp,
                              const conffwk::ConfigObject* netConn,
                              conffwk::ConfigObject& sidNetObj,
                              std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs)
{
  sidNetObj.set_obj("netconn", netConn);

  std::vector<const conffwk::ConfigObject*> source_id_objs;
  std::vector<uint32_t> app_source_ids;

  for (auto d2d_conn_res : roapp->get_detector_connections()) {

    // get the readout groups and the interfaces and streams therein; 1 reaout group corresponds to 1 data reader
    // module
    auto d2d_conn = d2d_conn_res->cast<confmodel::DetectorToDaqConnection>();

    if (!d2d_conn) {
      continue;
    }

    // Loop over senders
    for (auto dros : d2d_conn->streams()) {

      auto stream = dros->cast<confmodel::DetectorStream>();
      if (!stream)
        continue;
      app_source_ids.push_back(stream->get_source_id());
    }
  }

  for (auto& source_id : app_source_ids) {
    std::string streamSidUid(roapp->UID() + "SourceIDConf" + std::to_string(source_id));
    auto stream_sid_obj = std::make_shared<conffwk::ConfigObject>(obj_fac.create("SourceIDConf", streamSidUid));
    stream_sid_obj->set_by_val<uint32_t>("sid", source_id);
    stream_sid_obj->set_by_val<std::string>("subsystem", "Detector_Readout");
    sidObjs.push_back(stream_sid_obj);
    source_id_objs.push_back(sidObjs.back().get());
  }

  for (auto tp_sid : roapp->get_tp_source_ids()) {
    sidObjs.push_back(std::make_shared<conffwk::ConfigObject>(tp_sid->config_object()));
    source_id_objs.push_back(sidObjs.back().get());
  }
  /*
  std::string trgSidUid(roapp->UID() + "TRGSourceIDConf" + std::to_string(roapp->get_tp_source_id()));
  auto trig_sid_obj = std::make_shared<conffwk::ConfigObject>(obj_fac.create("SourceIDConf", trgSidUid));
  trig_sid_obj->set_by_val<uint32_t>("sid", roapp->get_tp_source_id());
  trig_sid_obj->set_by_val<std::string>("subsystem", "Trigger");
  source_id_objs.push_back(sidObjs.back().get());
  */

  sidNetObj.set_objs("source_ids", source_id_objs);
}

inline void
fill_sourceid_object_from_app(const ConfigObjectFactory& obj_fac,
                              const FakeDataApplication* fdapp,
                              const conffwk::ConfigObject* netConn,
                              conffwk::ConfigObject& sidNetObj,
                              std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs)
{
  sidNetObj.set_obj("netconn", netConn);

  std::vector<const conffwk::ConfigObject*> source_id_objs;
  std::vector<uint32_t> app_source_ids;

  for (auto fdpc : fdapp->get_producers()) {

    // get the readout groups and the interfaces and streams therein; 1 reaout group corresponds to 1 data reader
    // module

    app_source_ids.push_back(fdpc->get_source_id());
  }

  for (auto& source_id : app_source_ids) {
    std::string streamSidUid(fdapp->UID() + "SourceIDConf" + std::to_string(source_id));
    auto stream_sid_obj = std::make_shared<conffwk::ConfigObject>(obj_fac.create("SourceIDConf", streamSidUid));
    stream_sid_obj->set_by_val<uint32_t>("sid", source_id);
    stream_sid_obj->set_by_val<std::string>("subsystem", "Detector_Readout");
    sidObjs.push_back(stream_sid_obj);
    source_id_objs.push_back(sidObjs.back().get());
  }

  sidNetObj.set_objs("source_ids", source_id_objs);
}

std::set<std::string>
DFApplication::object_tags() const {
  std::set<std::string> tags;
  auto host = get_runs_on()->get_runs_on()->UID();
  for (auto writer : get_data_writers()) {
    auto path = writer->get_data_store_params()->get_directory_path();
    tags.insert("storage:"+host+":"+path);
  }
  return tags;
}

void
DFApplication::generate_modules(const confmodel::Session* session) const
{

  ConfigObjectFactory obj_fac(this);

  std::vector<const confmodel::DaqModule*> modules;

  // Containers for module specific config objects for output/input
  // Prepare TRB output objects
  std::vector<const conffwk::ConfigObject*> trbInputObjs;
  std::vector<const conffwk::ConfigObject*> trbOutputObjs;
  std::vector<const conffwk::ConfigObject*> trbSidNetObjs;

  // -- First, we process expected Queue and Network connections and create their objects.

  // Process the queue rules looking for the TriggerRecord queue between TRB and DataWriterModule
  const QueueDescriptor* trQDesc = nullptr;
  for (auto rule : get_queue_rules()) {
    auto destination_class = rule->get_destination_class();
    if (destination_class == "DataWriterModule") {
      trQDesc = rule->get_descriptor();
    }
  }
  if (trQDesc == nullptr) { // BadConf if no descriptor between TRB and DataWriterModule
    throw(BadConf(ERS_HERE, "Could not find queue descriptor rule for TriggerRecords!"));
  }
  // Create queue connection config object
  auto trQueueObj = obj_fac.create_queue_obj(trQDesc, UID());

  // Place trigger record queue object into vector of output objs of TRB module
  trbOutputObjs.push_back(&trQueueObj);

  // Process the network rules looking for the Fragments and TriggerDecision inputs for TRB
  const NetworkConnectionDescriptor* fragNetDesc = nullptr;
  const NetworkConnectionDescriptor* trigdecNetDesc = nullptr;
  const NetworkConnectionDescriptor* tokenNetDesc = nullptr;
  const NetworkConnectionDescriptor* trmonReqNetDesc = nullptr;
  const NetworkConnectionDescriptor* trmonTRNetDesc = nullptr;
  for (auto rule : get_network_rules()) {
    auto descriptor = rule->get_descriptor();
    auto data_type = descriptor->get_data_type();
    if (data_type == "Fragment") {
      fragNetDesc = rule->get_descriptor();
    } else if (data_type == "TriggerDecision") {
      trigdecNetDesc = rule->get_descriptor();
    } else if (data_type == "TriggerDecisionToken") {
      tokenNetDesc = rule->get_descriptor();
    } else if (data_type == "TRMonRequest") {
      trmonReqNetDesc = rule->get_descriptor();
    } else if (data_type == "TriggerRecord") {
      trmonTRNetDesc = rule->get_descriptor();
    }
  }
  if (fragNetDesc == nullptr) { // BadConf if no descriptor for Fragments into TRB
    throw(BadConf(ERS_HERE, "Could not find network descriptor rule for input Fragments!"));
  }
  if (trigdecNetDesc == nullptr) { // BadCond if no descriptor for TriggerDecisions into TRB
    throw(BadConf(ERS_HERE, "Could not find network descriptor rule for input TriggerDecisions!"));
  }
  if (tokenNetDesc == nullptr) { // BadCond if no descriptor for Tokens out of DataWriterModule
    throw(BadConf(ERS_HERE, "Could not find network descriptor rule for output TriggerDecisionTokens!"));
  }
  if (get_source_id() == nullptr) {
    throw(BadConf(ERS_HERE, "Could not retrieve SourceIDConf"));
  }
  // Create network connection config object
  auto fragNetObj = obj_fac.create_net_obj(fragNetDesc, UID());
  auto trigdecNetObj =  obj_fac.create_net_obj(trigdecNetDesc, UID());
  auto tokenNetObj = obj_fac.create_net_obj(tokenNetDesc, "");
  conffwk::ConfigObject trmonReqNetObj;
  conffwk::ConfigObject trmonTRNetObj;
  if (trmonReqNetDesc != nullptr) {
    trmonReqNetObj = obj_fac.create_net_obj(trmonReqNetDesc, UID());
  }
  if (trmonTRNetDesc != nullptr) {
    trmonTRNetObj = obj_fac.create_net_obj(trmonTRNetDesc, "");
  }

  // Process special Network rules!
  // Looking for DataRequest rules from ReadoutAppplications in current Session
  auto sessionApps = session->enabled_applications();
  std::vector<conffwk::ConfigObject> dreqNetObjs;
  std::vector<conffwk::ConfigObject> sidNetObjs;
  std::vector<std::shared_ptr<conffwk::ConfigObject>> sidObjs;
  for (auto app : sessionApps) {
    auto smartapp = app->cast<appmodel::SmartDaqApplication>();
    auto roapp = app->cast<appmodel::ReadoutApplication>();
    auto fdapp = app->cast<appmodel::FakeDataApplication>();
    auto ctbapp = app->cast<appmodel::CTBApplication>();
    auto dfapp = app->cast<appmodel::DFApplication>();
    auto rapp = app->cast<appmodel::TPReplayApplication>();
    if (smartapp == nullptr || dfapp != nullptr) {
      continue;
    }
    auto src_id_check = smartapp->get_source_id();
    if (roapp == nullptr
	&& fdapp == nullptr
	&& src_id_check == nullptr
	&& rapp == nullptr
	&& ctbapp == nullptr) {
      continue;
    }

    auto roQRules = smartapp->get_network_rules();
    for (auto rule : roQRules) {
      auto descriptor = rule->get_descriptor();
      auto data_type = descriptor->get_data_type();
      if (data_type == "DataRequest") {
	if (rapp != nullptr) {
          fill_sourceid_object_from_app(obj_fac, rapp, &dreqNetObjs, &sidNetObjs, sidObjs, descriptor, smartapp->UID());
	}
	// contrary to all the other applications
	// the CTB has 2 network connections, so this logic has to be splitted
	else if (ctbapp) {
	  auto sources = ctbapp->get_sources();
	  for ( const auto & s : sources ) {
	    std::string dreqNetUid(descriptor->get_uid_base() + smartapp->UID()+ '_' + s.first);
	    dreqNetObjs.emplace_back( obj_fac.create_net_obj(descriptor, dreqNetUid) );

	    std::string sidToNetUid(descriptor->get_uid_base() + smartapp->UID() + "_" + s.first);
	    sidNetObjs.emplace_back( obj_fac.create("SourceIDToNetworkConnection", sidToNetUid ) );
	    sidNetObjs.back().set_obj("netconn", & dreqNetObjs.back());
	    sidNetObjs.back().set_objs("source_ids", { & s.second->config_object() });
	  } // loop over CTB sources
	} else {

	  dreqNetObjs.emplace_back(obj_fac.create_net_obj(descriptor, smartapp->UID()));

	  std::string sidToNetUid(descriptor->get_uid_base() + smartapp->UID() + "-sids");
	  sidNetObjs.emplace_back(
				    obj_fac.create("SourceIDToNetworkConnection", sidToNetUid));
	  if (roapp != nullptr) {
	    fill_sourceid_object_from_app(obj_fac, roapp, &dreqNetObjs.back(), sidNetObjs.back(), sidObjs);
	  } else if (fdapp != nullptr) {
	    fill_sourceid_object_from_app(obj_fac, fdapp, &dreqNetObjs.back(), sidNetObjs.back(), sidObjs);
	  } else {
	    fill_sourceid_object_from_app(smartapp, &dreqNetObjs.back(), sidNetObjs.back());
	  }
	} // else from if (ctbapp)
      } // If network rule has DataRequest type of data
    }   // Loop over Apps network rules
  }     // loop over Session specific Apps

  // Get pointers to objects here, after vector has been filled so they don't move on us
  for (auto& obj : dreqNetObjs) {
    trbOutputObjs.push_back(&obj);
  }
  for (auto& obj : sidNetObjs) {
    trbSidNetObjs.push_back(&obj);
  }

  // -- Second, we create the Module objects and assign their configs, with the precreated
  // -- connection config objects above.

  // Get TRB Config Object
  auto trbConf = get_trb();
  if (trbConf == nullptr) {
    throw(BadConf(ERS_HERE, "No DataWriterModule or TRB configuration given"));
  }
  auto trbConfObj = trbConf->config_object();
  trbConfObj.set_by_val<uint32_t>("source_id", get_source_id()->get_sid());
  trbInputObjs = { &trigdecNetObj, &fragNetObj };
  if (trmonReqNetDesc != nullptr) {
    trbInputObjs.push_back(&trmonReqNetObj);
  }
  if (trmonTRNetDesc != nullptr) {
    trbOutputObjs.push_back(&trmonTRNetObj);
  }
  // Prepare TRB Module Object and assign its Config Object.
  std::string trbUid(UID() + "-trb");
  conffwk::ConfigObject trbObj = obj_fac.create("TRBModule", trbUid);
  trbObj.set_obj("configuration", &trbConfObj);
  trbObj.set_objs("inputs", trbInputObjs);
  trbObj.set_objs("outputs", trbOutputObjs);
  trbObj.set_obj("trigger_record_output", &trQueueObj);
  trbObj.set_objs("request_connections", trbSidNetObjs);
  // Push TRB Module Object from confdb
  modules.push_back(obj_fac.get_dal<TRBModule>(trbUid));

  // Get DataWriterModule Config Object (only one for now, maybe more later?)
  auto dwrConfs = get_data_writers();
  if (dwrConfs.size() == 0) {
    throw(BadConf(ERS_HERE, "No DataWriterModule or TRB configuration given"));
  }
  uint dw_idx = 0;
  for (auto dwrConf : dwrConfs) {
    // auto fnParamsObj = dwrConf->get_data_store_params()->get_filename_params()->config_object();
    // fnParamsObj.set_by_val<std::string>("writer_identifier", fmt::format("{}_datawriter-{}", UID(), dw_idx));
    auto dwrConfObj = dwrConf->config_object();

    // Prepare DataWriterModule Module Object and assign its Config Object.
    std::string dwrUid(fmt::format("{}-dw-{}", UID(), dw_idx));
    conffwk::ConfigObject dwrObj = obj_fac.create("DataWriterModule", dwrUid);
    dwrObj.set_by_val("writer_identifier", fmt::format("{}_dw_{}", UID(), dw_idx));
    dwrObj.set_obj("configuration", &dwrConfObj);
    dwrObj.set_objs("inputs", { &trQueueObj });
    dwrObj.set_objs("outputs", { &tokenNetObj });
    // Push DataWriterModule Module Object from confdb
    modules.push_back(obj_fac.get_dal<DataWriterModule>(dwrUid));
    ++dw_idx;
  }

  obj_fac.update_modules(modules);
}

} // namespace appmodel  
} // namespace dunedaq