studies/upgrade_noise/deployment/process_i3_file.py
def main(input_files, output_file, key, pulsemaps, gcd_file, events_max):
    """Run minimal icetray chain with GraphNeT module."""

    # Make sure output directory exists
    makedirs(dirname(output_file), exist_ok=True)

    # Get GCD file
    if gcd_file is None:
        gcd_candidates = [p for p in input_files if is_gcd_file(p)]
        assert (
            len(gcd_candidates) == 1
        ), f"Did not get exactly one GCD-file candidate in `{dirname(input_files[0])}: {gcd_candidates}"
        gcd_file = gcd_candidates[0]

    # Get all input I3-files
    input_files = [p for p in input_files if not is_gcd_file(p)]

    # Run graphnet module in tray
    min_hit_oms = 10

    tray = I3Tray()
    tray.Add("I3Reader", filenamelist=input_files)
    tray.Add(
        lambda frame: len(
            dataclasses.I3RecoPulseSeriesMap.from_frame(frame, pulsemaps[0])
        )
        >= min_hit_oms,
    )
    tray.Add(
        GraphNeTModuleIceCubeUpgrade,
        keys=key,
        model=MODEL_PATH,
        pulsemaps=pulsemaps,
        gcd_file=gcd_file,
    )
    tray.Add("I3Writer", filename=output_file)

    if events_max > 0:
        tray.Execute(events_max)
    else:
        tray.Execute()
Run minimal icetray chain with GraphNeT module.
Run minimal icetray chain with GraphNeT module.
Run minimal icetray chain with GraphNeT module .
def main(input_files, output_file, key, pulsemaps, gcd_file, events_max):
    makedirs(dirname(output_file), exist_ok=True)
    if gcd_file is None:
        gcd_candidates = [p for p in input_files if is_gcd_file(p)]
        assert (
            len(gcd_candidates) == 1
        ), f"Did not get exactly one GCD-file candidate in `{dirname(input_files[0])}: {gcd_candidates}"
        gcd_file = gcd_candidates[0]
    input_files = [p for p in input_files if not is_gcd_file(p)]
    min_hit_oms = 10
    tray = I3Tray()
    tray.Add("I3Reader", filenamelist=input_files)
    tray.Add(
        lambda frame: len(
            dataclasses.I3RecoPulseSeriesMap.from_frame(frame, pulsemaps[0])
        )
        >= min_hit_oms,
    )
    tray.Add(
        GraphNeTModuleIceCubeUpgrade,
        keys=key,
        model=MODEL_PATH,
        pulsemaps=pulsemaps,
        gcd_file=gcd_file,
    )
    tray.Add("I3Writer", filename=output_file)
    if events_max > 0:
        tray.Execute(events_max)
    else:
        tray.Execute()
def main ( input_files , output_file , key , pulsemaps , gcd_file , events_max ) : makedirs ( dirname ( output_file ) , exist_ok = True ) if gcd_file is None : gcd_candidates = [ p for p in input_files if is_gcd_file ( p ) ] assert ( len ( gcd_candidates ) == 1 ) , f"Did not get exactly one GCD-file candidate in `{dirname(input_files[0])}: {gcd_candidates}" gcd_file = gcd_candidates [ 0 ] input_files = [ p for p in input_files if not is_gcd_file ( p ) ] min_hit_oms = 10 tray = I3Tray ( ) tray . Add ( "I3Reader" , filenamelist = input_files ) tray . Add ( lambda frame : len ( dataclasses . I3RecoPulseSeriesMap . from_frame ( frame , pulsemaps [ 0 ] ) ) >= min_hit_oms , ) tray . Add ( GraphNeTModuleIceCubeUpgrade , keys = key , model = MODEL_PATH , pulsemaps = pulsemaps , gcd_file = gcd_file , ) tray . Add ( "I3Writer" , filename = output_file ) if events_max > 0 : tray . Execute ( events_max ) else : tray . Execute ( )
Run minimal icetray chain with GraphNeT module.
Run minimal icetray chain with GraphNeT module .
"""Run minimal icetray chain with GraphNeT module.""" # Make sure output directory exists # Get GCD file # Get all input I3-files # Run graphnet module in tray
input_files
output_file
key
pulsemaps
gcd_file
events_max