Functions
	generate_df (network_endpoints, NUMBER_OF_DATA_PRODUCERS=2, DATA_RATE_SLOWDOWN_FACTOR=1, RUN_NUMBER=333, TRIGGER_RATE_HZ=1.0, DATA_FILE="./frames.bin", OUTPUT_PATH=".", DISABLE_OUTPUT=False, TOKEN_COUNT=10)

	generate_trigemu (network_endpoints, NUMBER_OF_DATA_PRODUCERS=2, DATA_RATE_SLOWDOWN_FACTOR=1, RUN_NUMBER=333, TRIGGER_RATE_HZ=1.0, DATA_FILE="./frames.bin", OUTPUT_PATH=".")

	cli (number_of_data_producers, data_rate_slowdown_factor, run_number, trigger_rate_hz, data_file, output_path, disable_data_storage, token_count, host_ip_df, host_ip_trigemu, json_file_base)

Variables
	default_load_path

int	QUEUE_POP_WAIT_MS = 100;

int	CLOCK_SPEED_HZ = 62500000;

	CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])

Function Documentation

◆ cli()

testapp_noreadout_two_process.cli	(	number_of_data_producers,
		data_rate_slowdown_factor,
		run_number,
		trigger_rate_hz,
		data_file,
		output_path,
		disable_data_storage,
		token_count,
		host_ip_df,
		host_ip_trigemu,
		json_file_base )

  JSON_FILE: Input raw data file.
  JSON_FILE: Output json configuration file.

Definition at line 397 of file testapp_noreadout_two_process.py.

    def cli(number_of_data_producers, data_rate_slowdown_factor, run_number, trigger_rate_hz, data_file, output_path, disable_data_storage, token_count, host_ip_df, host_ip_trigemu, json_file_base):
        """
          JSON_FILE: Input raw data file.
          JSON_FILE: Output json configuration file.
        """
 
        json_file_trigemu=json_file_base+"-trigemu.json"
        json_file_df=json_file_base+"-df.json"
 
        network_endpoints={
            "trigdec" : f"tcp://{host_ip_trigemu}:12345",
            "triginh" : f"tcp://{host_ip_df}:12346",
            "timesync": f"tcp://{host_ip_df}:12347"
        }
 
        with open(json_file_trigemu, 'w') as f:
            f.write(generate_trigemu(
                    network_endpoints,
                    NUMBER_OF_DATA_PRODUCERS = number_of_data_producers,
                    DATA_RATE_SLOWDOWN_FACTOR = data_rate_slowdown_factor,
                    RUN_NUMBER = run_number, 
                    TRIGGER_RATE_HZ = trigger_rate_hz,
                    DATA_FILE = data_file,
                    OUTPUT_PATH = output_path
                ))
 
        with open(json_file_df, 'w') as f:
            f.write(generate_df(
                    network_endpoints,
                    NUMBER_OF_DATA_PRODUCERS = number_of_data_producers,
                    DATA_RATE_SLOWDOWN_FACTOR = data_rate_slowdown_factor,
                    RUN_NUMBER = run_number, 
                    TRIGGER_RATE_HZ = trigger_rate_hz,
                    DATA_FILE = data_file,
                    OUTPUT_PATH = output_path,
                    DISABLE_OUTPUT = disable_data_storage,
                    TOKEN_COUNT = token_count
                ))
 

◆ generate_df()

testapp_noreadout_two_process.generate_df	(	network_endpoints,
		NUMBER_OF_DATA_PRODUCERS = 2,
		DATA_RATE_SLOWDOWN_FACTOR = 1,
		RUN_NUMBER = 333,
		TRIGGER_RATE_HZ = 1.0,
		DATA_FILE = "./frames.bin",
		OUTPUT_PATH = ".",
		DISABLE_OUTPUT = False,
		TOKEN_COUNT = 10 )

Generate the json configuration for the readout and DF process

Definition at line 66 of file testapp_noreadout_two_process.py.

    ):
    """Generate the json configuration for the readout and DF process"""
   
    trigger_interval_ticks = math.floor((1/TRIGGER_RATE_HZ) * CLOCK_SPEED_HZ/DATA_RATE_SLOWDOWN_FACTOR)
 
    # Define modules and queues
    queue_bare_specs = [
            app.QueueSpec(inst="time_sync_to_netq", kind='FollyMPMCQueue', capacity=100),
            app.QueueSpec(inst="token_to_netq", kind='FollySPSCQueue', capacity=20),
            app.QueueSpec(inst="trigger_decision_from_netq", kind='FollySPSCQueue', capacity=20),
            app.QueueSpec(inst="trigger_decision_copy_for_bookkeeping", kind='FollySPSCQueue', capacity=20),
            app.QueueSpec(inst="trigger_record_q", kind='FollySPSCQueue', capacity=20),
            app.QueueSpec(inst="data_fragments_q", kind='FollyMPMCQueue', capacity=100),
        ] + [
            app.QueueSpec(inst=f"data_requests_{idx}", kind='FollySPSCQueue', capacity=20)
                for idx in range(NUMBER_OF_DATA_PRODUCERS)
        ]
    
 
    # Only needed to reproduce the same order as when using jsonnet
    queue_specs = app.QueueSpecs(sorted(queue_bare_specs, key=lambda x: x.inst))
 
 
    mod_specs = [
        mspec("ntoq_trigdec", "NetworkToQueue", [
                        app.QueueInfo(name="output", inst="trigger_decision_from_netq", dir="output")
                    ]),
 
        mspec("qton_token", "QueueToNetwork", [
                        app.QueueInfo(name="input", inst="token_to_netq", dir="input")
                    ]),
 
        mspec("qton_timesync", "QueueToNetwork", [
                        app.QueueInfo(name="input", inst="time_sync_to_netq", dir="input")
                    ]),
 
        mspec("rqg", "RequestGenerator", [
                        app.QueueInfo(name="trigger_decision_input_queue", inst="trigger_decision_from_netq", dir="input"),
                        app.QueueInfo(name="trigger_decision_for_event_building", inst="trigger_decision_copy_for_bookkeeping", dir="output"),
                    ] + [
                        app.QueueInfo(name=f"data_request_{idx}_output_queue", inst=f"data_requests_{idx}", dir="output")
                            for idx in range(NUMBER_OF_DATA_PRODUCERS)
                    ]),
 
        mspec("ffr", "FragmentReceiver", [
                        app.QueueInfo(name="trigger_decision_input_queue", inst="trigger_decision_copy_for_bookkeeping", dir="input"),
                        app.QueueInfo(name="trigger_record_output_queue", inst="trigger_record_q", dir="output"),
                        app.QueueInfo(name="data_fragment_input_queue", inst="data_fragments_q", dir="input"),
                    ]),
 
        mspec("datawriter", "DataWriterModule", [
                        app.QueueInfo(name="trigger_record_input_queue", inst="trigger_record_q", dir="input"),
                        app.QueueInfo(name="token_output_queue", inst="token_to_netq", dir="output"),
                    ]),
 
        mspec("fake_timesync_source", "FakeTimeSyncSource", [
                        app.QueueInfo(name="time_sync_sink", inst="time_sync_to_netq", dir="output"),
                    ]),
 
        ] + [
 
                mspec(f"fakedataprod_{idx}", "FakeDataProdModule", [
                            app.QueueInfo(name="data_request_input_queue", inst=f"data_requests_{idx}", dir="input"),
                            app.QueueInfo(name="data_fragment_output_queue", inst="data_fragments_q", dir="output"),
                            ]) for idx in range(NUMBER_OF_DATA_PRODUCERS)
        ]
 
    init_specs = app.Init(queues=queue_specs, modules=mod_specs)
 
    initcmd = rccmd.RCCommand(
        id=basecmd.CmdId("init"),
        entry_state="NONE",
        exit_state="INITIAL",
        data=init_specs
    )
 
    confcmd = mrccmd("conf", "INITIAL", "CONFIGURED",[
                ("ntoq_trigdec", ntoq.Conf(msg_type="dunedaq::dfmessages::TriggerDecision",
                                           msg_module_name="TriggerDecisionNQ",
                                           receiver_config=nor.Conf(ipm_plugin_type="ZmqReceiver",
                                                                    address=network_endpoints["trigdec"])
                                           )
                 ),
 
                ("qton_token", qton.Conf(msg_type="dunedaq::dfmessages::TriggerDecisionToken",
                                           msg_module_name="TriggerDecisionTokenNQ",
                                           sender_config=nos.Conf(ipm_plugin_type="ZmqSender",
                                                                  address=network_endpoints["triginh"],
                                                                  stype="msgpack")
                                           )
                 ),
 
                ("qton_timesync", qton.Conf(msg_type="dunedaq::dfmessages::TimeSync",
                                            msg_module_name="TimeSyncNQ",
                                            sender_config=nos.Conf(ipm_plugin_type="ZmqSender",
                                                                   address=network_endpoints["timesync"],
                                                                   stype="msgpack")
                                           )
                ),
        
                ("rqg", rqg.ConfParams(
                        map=rqg.mapgeoidqueue([
                                rqg.geoidinst(apa=0, link=idx, queueinstance=f"data_requests_{idx}") for idx in range(NUMBER_OF_DATA_PRODUCERS)
                            ])  
                        )),
                ("ffr", ffr.ConfParams(
                            general_queue_timeout=QUEUE_POP_WAIT_MS
                        )),
                ("datawriter", dw.ConfParams(
                            initial_token_count=TOKEN_COUNT,
                            data_store_parameters=hdf5ds.ConfParams(
                                name="data_store",
                                # type = "HDF5DataStore", # default
                                directory_path = OUTPUT_PATH, # default
                                # mode = "all-per-file", # default
                                max_file_size_bytes = 1073741834,
                                disable_unique_filename_suffix = False,
                                filename_parameters = hdf5ds.HDF5DataStoreFileNameParams(
                                    overall_prefix = "fake_minidaqapp",
                                    # digits_for_run_number = 6, #default
                                    file_index_prefix = "file"
                                ),
                                file_layout_parameters = hdf5ds.HDF5DataStoreFileLayoutParams(
                                    trigger_record_name_prefix= "TriggerRecord",
                                    digits_for_trigger_number = 5,
                                )
                            )
                        )),
                ("fake_timesync_source", ftss.ConfParams(
                    sync_interval_ticks = (CLOCK_SPEED_HZ/DATA_RATE_SLOWDOWN_FACTOR),
                    clock_frequency_hz = (CLOCK_SPEED_HZ/DATA_RATE_SLOWDOWN_FACTOR),
                  )),
            ] + [
                (f"fakedataprod_{idx}", fdp.ConfParams(
                        temporarily_hacked_link_number = idx
                        )) for idx in range(NUMBER_OF_DATA_PRODUCERS)
            ])
 
    startpars = rccmd.StartParams(run=RUN_NUMBER, disable_data_storage=DISABLE_OUTPUT)
    startcmd = mrccmd("start", "CONFIGURED", "RUNNING", [
            ("ntoq_trigdec", startpars),
            ("qton_token", startpars),
            ("qton_timesync", startpars),
            ("datawriter", startpars),
            ("ffr", startpars),
            ("fakedataprod_.*", startpars),
            ("rqg", startpars),
            ("fake_timesync_source", startpars),
        ])
 
    stopcmd = mrccmd("stop", "RUNNING", "CONFIGURED", [
            ("ntoq_trigdec", None),
            ("qton_timesync", None),
            ("qton_token", None),
            ("fake_timesync_source", None),
            ("rqg", None),
            ("fakedataprod_.*", None),
            ("ffr", None),
            ("datawriter", None),
        ])
 
    scrapcmd = mcmd("scrap", [
            ("", None)
        ])
 
    # Create a list of commands
    cmd_seq = [initcmd, confcmd, startcmd, stopcmd, scrapcmd]
 
    # Print them as json (to be improved/moved out)
    jstr = json.dumps([c.pod() for c in cmd_seq], indent=4, sort_keys=True)
    return jstr
 
#===============================================================================

◆ generate_trigemu()

testapp_noreadout_two_process.generate_trigemu	(	network_endpoints,
		NUMBER_OF_DATA_PRODUCERS = 2,
		DATA_RATE_SLOWDOWN_FACTOR = 1,
		RUN_NUMBER = 333,
		TRIGGER_RATE_HZ = 1.0,
		DATA_FILE = "./frames.bin",
		OUTPUT_PATH = "." )

Generate the json config for the TriggerDecisionEmulator process

Definition at line 249 of file testapp_noreadout_two_process.py.

    ):
    """Generate the json config for the TriggerDecisionEmulator process"""
    
    trigger_interval_ticks = math.floor((1/TRIGGER_RATE_HZ) * CLOCK_SPEED_HZ/DATA_RATE_SLOWDOWN_FACTOR)
 
    # Define modules and queues
    queue_bare_specs = [
            app.QueueSpec(inst="time_sync_from_netq", kind='FollySPSCQueue', capacity=100),
            app.QueueSpec(inst="token_from_netq", kind='FollySPSCQueue', capacity=20),
            app.QueueSpec(inst="trigger_decision_to_netq", kind='FollySPSCQueue', capacity=20),
        ]
 
    # Only needed to reproduce the same order as when using jsonnet
    queue_specs = app.QueueSpecs(sorted(queue_bare_specs, key=lambda x: x.inst))
 
 
    mod_specs = [
        mspec("qton_trigdec", "QueueToNetwork", [
                        app.QueueInfo(name="input", inst="trigger_decision_to_netq", dir="input")
                    ]),
 
        mspec("ntoq_token", "NetworkToQueue", [
                        app.QueueInfo(name="output", inst="token_from_netq", dir="output")
                    ]),
 
        mspec("ntoq_timesync", "NetworkToQueue", [
                        app.QueueInfo(name="output", inst="time_sync_from_netq", dir="output")
                    ]),
 
        mspec("tde", "TriggerDecisionEmulator", [
                        app.QueueInfo(name="time_sync_source", inst="time_sync_from_netq", dir="input"),
                        app.QueueInfo(name="token_source", inst="token_from_netq", dir="input"),
                        app.QueueInfo(name="trigger_decision_sink", inst="trigger_decision_to_netq", dir="output"),
                    ]),
        ]
 
    init_specs = app.Init(queues=queue_specs, modules=mod_specs)
 
    initcmd = rccmd.RCCommand(
        id=basecmd.CmdId("init"),
        entry_state="NONE",
        exit_state="INITIAL",
        data=init_specs
    )
 
    confcmd = mrccmd("conf", "INITIAL", "CONFIGURED",[
                ("qton_trigdec", qton.Conf(msg_type="dunedaq::dfmessages::TriggerDecision",
                                           msg_module_name="TriggerDecisionNQ",
                                           sender_config=nos.Conf(ipm_plugin_type="ZmqSender",
                                                                  address=network_endpoints["trigdec"],
                                                                  stype="msgpack")
                                           )
                 ),
 
                 ("ntoq_token", ntoq.Conf(msg_type="dunedaq::dfmessages::TriggerDecisionToken",
                                            msg_module_name="TriggerDecisionTokenNQ",
                                            receiver_config=nor.Conf(ipm_plugin_type="ZmqReceiver",
                                                                     address=network_endpoints["triginh"])
                                            )
                 ),
 
                ("ntoq_timesync", ntoq.Conf(msg_type="dunedaq::dfmessages::TimeSync",
                                           msg_module_name="TimeSyncNQ",
                                           receiver_config=nor.Conf(ipm_plugin_type="ZmqReceiver",
                                                                    address=network_endpoints["timesync"])
                                           )
                ),
 
                ("tde", tde.ConfParams(
                        links=[idx for idx in range(NUMBER_OF_DATA_PRODUCERS)],
                        min_links_in_request=NUMBER_OF_DATA_PRODUCERS,
                        max_links_in_request=NUMBER_OF_DATA_PRODUCERS,
                        min_readout_window_ticks=1200,
                        max_readout_window_ticks=1200,
                        trigger_window_offset=1000,
                        # The delay is set to put the trigger well within the latency buff
                        trigger_delay_ticks=math.floor( 2* CLOCK_SPEED_HZ/DATA_RATE_SLOWDOWN_FACTOR),
                        # We divide the trigger interval by
                        # DATA_RATE_SLOWDOWN_FACTOR so the triggers are still
                        # emitted per (wall-clock) second, rather than being
                        # spaced out further
                        trigger_interval_ticks=trigger_interval_ticks,
                        clock_frequency_hz=CLOCK_SPEED_HZ/DATA_RATE_SLOWDOWN_FACTOR                    
                        )),
            ])
 
    startpars = rccmd.StartParams(run=RUN_NUMBER, disable_data_storage=False)
    startcmd = mrccmd("start", "CONFIGURED", "RUNNING", [
            ("qton_trigdec", startpars),
            ("ntoq_token", startpars),
            ("ntoq_timesync", startpars),
            ("tde", startpars),
        ])
 
    stopcmd = mrccmd("stop", "RUNNING", "CONFIGURED", [
            ("qton_trigdec", None),
            ("ntoq_timesync", None),
            ("ntoq_token", None),
            ("tde", None),
        ])
 
    pausecmd = mrccmd("pause", "RUNNING", "RUNNING", [
            ("", None)
        ])
 
    resumecmd = mrccmd("resume", "RUNNING", "RUNNING", [
            ("tde", tde.ResumeParams(
                            trigger_interval_ticks=trigger_interval_ticks
                        ))
        ])
 
    scrapcmd = mcmd("scrap", [
            ("", None)
        ])
 
    # Create a list of commands
    cmd_seq = [initcmd, confcmd, startcmd, stopcmd, pausecmd, resumecmd, scrapcmd]
 
    # Print them as json (to be improved/moved out)
    jstr = json.dumps([c.pod() for c in cmd_seq], indent=4, sort_keys=True)
    return jstr
 