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