File indexing completed on 2026-09-14 08:23:24
0001
0002
0003
0004
0005
0006
0007
0008
0009
0010
0011 import os
0012 import dddigi
0013 import logging
0014 from dd4hep import units
0015 from dddigi import DEBUG, INFO, WARNING, ERROR
0016
0017 logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
0018 logger = logging.getLogger(__name__)
0019
0020 global attenuation
0021 attenuation = {'Minitel1Hits': 50 * units.ns,
0022 'Minitel2Hits': 50 * units.ns,
0023 'Minitel3Hits': 50 * units.ns,
0024 }
0025
0026
0027
0028 class Test(dddigi.Digitize):
0029
0030 def __init__(self, geometry=None, process_data=True):
0031 dddigi.Digitize.__init__(self, dddigi.Kernel())
0032 dddigi.setPrintFormat(str('%-32s %5s %s'))
0033 dddigi.setPrintLevel(INFO)
0034 self.kernel().printProperties()
0035 self.geometry = geometry
0036 self.input = None
0037 self.main_sequencer()
0038 self.attenuation = attenuation
0039 self.used_inputs = []
0040 self.inputs = ['MiniTel.run00000000.root',
0041 'MiniTel.run00000001.root',
0042 'MiniTel.run00000002.root',
0043 'MiniTel.run00000003.root',
0044 'MiniTel.run00000004.root',
0045 'MiniTel.run00000005.root',
0046 'MiniTel.run00000006.root',
0047 'MiniTel.run00000007.root',
0048 'MiniTel.run00000008.root']
0049 if not os.path.exists(self.inputs[0]):
0050 if os.path.exists('DDDigi'):
0051 os.chdir('DDDigi')
0052 if process_data and not os.path.exists(self.inputs[0]):
0053
0054 open(self.inputs[0])
0055
0056 def segment_action(self, nam, **options):
0057 obj = dddigi.Interface.createSegmentAction(self.kernel(), str(nam))
0058 return obj
0059
0060 def load_geo(self, volume_manager=None):
0061 fname = 'file:' + os.environ['DD4hepExamplesINSTALL'] + '/examples/ClientTests/compact/MiniTelGenerate.xml'
0062 self.kernel().loadGeometry(str(fname))
0063 self.printDetectors()
0064 if volume_manager:
0065 vm = self.description.volumeManager()
0066 if not vm.isValid():
0067 self.description.processXMLString(str("""<plugins>
0068 <plugin name="DD4hep_VolumeManager"/>
0069 </plugins>"""))
0070 self.volumeManager = self.description.volumeManager()
0071 if self.volumeManager.isValid():
0072 self.info('+++ Successfully created DD4hep VolumeManager')
0073 return self
0074
0075 def data_containers(self):
0076 return list(self.attenuation.keys())
0077
0078 def containers(self, count=None):
0079 if count:
0080 conts = []
0081 result = []
0082 for key in list(self.attenuation.keys()):
0083 conts.append(key)
0084 if len(conts) == count:
0085 result.append(conts)
0086 conts = []
0087 if len(conts) > 0:
0088 result.append(conts)
0089 else:
0090 result = list(self.attenuation.keys())
0091 return result
0092
0093 def check_creation(self, objs):
0094 for o in objs:
0095 if o is None:
0096 self.error('FAILED Failed to create object')
0097
0098 def declare_input(self, name, input, parallel=True):
0099 if not self.input:
0100 self.input = dddigi.Synchronize(self.kernel(), 'DigiParallelActionSequence/READER')
0101 self.input.parallel = True
0102
0103 def next_input(self):
0104 if len(self.used_inputs) == len(self.inputs):
0105 self.used_inputs = []
0106 next_source = self.inputs[len(self.used_inputs)]
0107 self.used_inputs.append(next_source)
0108 self.info('Preparing next input file: ' + str(next_source))
0109 return next_source
0110
0111 def run_checked(self, num_events=5, num_threads=5, parallel=3):
0112 result = 'FAILED'
0113 if self.num_events:
0114 num_events = int(self.num_events)
0115 if self.num_threads:
0116 num_threads = int(self.num_threads)
0117 if self.events_parallel:
0118 parallel = int(self.events_parallel)
0119 evt_done = self.run(num_events=num_events, num_threads=num_threads, parallel=parallel)
0120 if evt_done == num_events:
0121 result = 'PASSED'
0122 self.always('%s Test finished after processing %d events. [%d parallel threads, %d parallel events]'
0123 % (result, evt_done, num_threads, parallel, ))
0124 self.kernel().terminate()
0125 return evt_done
0126
0127
0128
0129 def test_setup_1(digi, print_level=WARNING, parallel=True):
0130 """
0131 Create default setup for tests. Simply too bad to repeat the same lines over and over again.
0132
0133 \author M.Frank
0134 \version 1.0
0135 """
0136
0137 digi.info('Created SIGNAL input')
0138 input = digi.input_action('DigiParallelActionSequence/READER')
0139 input.adopt_action('DigiDDG4ROOT/SignalReader',
0140 mask=0xCBAA,
0141 input=[digi.next_input()],
0142 OutputLevel=print_level, keep_raw=False)
0143
0144 digi.info('Creating collision overlay....')
0145
0146 overlay = input.adopt_action('DigiSequentialActionSequence/Overlay-1')
0147 overlay.adopt_action('DigiDDG4ROOT/Read-1',
0148 mask=0xCBEE,
0149 input=[digi.next_input()],
0150 OutputLevel=print_level,
0151 keep_raw=False)
0152 digi.info('Created input.overlay-1')
0153
0154 event = digi.event_action('DigiSequentialActionSequence/EventAction')
0155 combine = event.adopt_action('DigiContainerCombine/Combine',
0156 OutputLevel=print_level,
0157 parallel=parallel,
0158 input_masks=[0xCBAA, 0xCBEE],
0159 output_mask=0xAAA0,
0160 output_segment='deposits')
0161 combine.erase_combined = True
0162 proc = event.adopt_action('DigiContainerSequenceAction/HitP1',
0163 parallel=parallel,
0164 input_mask=0xAAA0,
0165 input_segment='deposits',
0166 output_mask=0xEEE5,
0167 output_segment='deposits')
0168 combine = digi.create_action('DigiDepositWeightedPosition/WeightedPosition', OutputLevel=print_level)
0169 proc.adopt_container_processor(combine, digi.containers())
0170 conts = [c for c in digi.containers()]
0171 event.adopt_action('DigiContainerDrop/Drop',
0172 containers=conts,
0173 input_segment='deposits',
0174 input_masks=[0xAAA0])
0175
0176 return event
