wfsim package

Submodules

wfsim.core module

class wfsim.core.NestId

Bases: object

Nest ids for refering to different scintilation models, only ER is actually validated

ALPHA = [6]

ER = [7, 8, 11, 12]

LED = [20]

NR = [0]

class wfsim.core.PMT_Afterpulse(config)

Bases: wfsim.core.Pulse

Produce pmt after pulse simulation, using already built cdfs The cdfs follow distribution parameters extracted from data.

photon_afterpulse(signal_pulse)

For pmt afterpulses, gain and dpe generation is a bit different from standard photons

class wfsim.core.PhotoElectric_Electron(config)

Bases: wfsim.core.S2

Produce electron after S2 pulse simulation, using a gaussian distribution

electron_afterpulse(signal_pulse, signal_pulse_instruction)

generate_instruction(signal_pulse, signal_pulse_instruction)

class wfsim.core.PhotoIonization_Electron(config)

Bases: wfsim.core.S2

Produce electron after pulse simulation, using already built cdfs The cdfs follow distribution parameters extracted from data.

electron_afterpulse(signal_pulse, signal_pulse_instruction)

For electron afterpulses we assume a uniform x, y

generate_instruction(signal_pulse, signal_pulse_instruction)

class wfsim.core.Pulse(config)

Bases: object

Pulse building class

static add_current(photon_timings, photon_gains, pulse_left, dt, pmt_current_templates, pulse_current)

clear_pulse_cache()

init_pmt_current_templates()

Create spe templates, for 10ns sample duration and 1ns rounding we have: _pmt_current_templates[i] : photon timing fall between [10*m+i, 10*m+i+1) (i, m are integers)

init_spe_scaling_factor_distributions()

static singlet_triplet_delays(size, singlet_ratio, config, phase)

Given the amount of the eximer, return time between excimer decay and their time of generation. size - amount of eximer self.phase - ‘liquid’ or ‘gas’ singlet_ratio - fraction of excimers that become singlets

(NOT the ratio of singlets/triplets!)

uniform_to_pe_arr(p, channel=0)

class wfsim.core.RawData(config)

Bases: object

ZLE()

Modified software zero lengh encoding, coverting WFs into pulses (XENON definition)

static add_baseline(data, channel_mask, baseline)

static add_noise(data, channel_mask, noise_data, noise_data_length)

Get chunk(s) of noise sample from real noise data

digitize_pulse_cache()

Superimpose pulses (wfsim definition) into WFs w/ dynamic range truncation

static digitizer_saturation(data, channel_mask)

get_truth(instruction, truth_buffer)

Write truth in the first empty row of truth_buffer

Parameters

instruction – Array of instructions that were simulated as a

single cluster, and should thus get one line in the truth info. :param truth_buffer: Truth buffer to write in.

sim_data(instruction, **kwargs)

Simulate a pulse according to instruction, and yield any additional instructions for secondary electron afterpulses.

sim_primary(primary_pulse, instruction, **kwargs)

static sum_signal(adc_wave, left, right, sum_template)

static symtype(ptype)

class wfsim.core.S1(config)

Bases: wfsim.core.Pulse

Given temperal inputs as well as number of photons Random generate photon timing and channel distribution.

static alpha(size, config, phase)

Calculate S1 photon timings for an alpha decay. Neglible recombination time, not validated :param size: 1d array of ints, number of photons :param config: dict wfsim config :parma phase: str “liquid”

return 1d array of photon timings

static er(size, config, phase)

Complex ER model, not validated :param size: 1d array of ints, number of photons :param config: dict wfsim config :parma phase: str “liquid” return 1d array of photon timings

static get_n_photons(n_photons, positions, s1_light_yield_map, config)

Calculates number of detected photons based on number of photons in total and the postions :param n_photons: 1d array of ints with number of photons: :param postions: 2d array with xyz positions of interactions :param s1_light_yield map: interpolator instance of s1 light yield map :param config: dict wfsim config

return array with number photons

static led(size, config, **kwargs)

distribute photons uniformly within the LED pulse length, not validated :param size: 1d array of ints, number of photons :param config: dict wfsim config :parma phase: str “liquid

return 1d array of photon timings

static nr(size, config, phase)

NR model model, not validated :param size: 1d array of ints, number of photons :param config: dict wfsim config :parma phase: str “liquid” return 1d array of photon timings

static photon_channels(positions, n_photons, config, s1_pattern_map)

Calculate photon arrival channels :params positions: 2d array with xy positions of interactions :params n_photons: 1d array of ints with number of photons to simulate :params config: dict wfsim config :params s1_pattern_map: interpolator instance of the s1 pattern map

returns nested array with photon channels

static photon_timings(t, n_photons, recoil_type, config, phase)

Calculate distribution of photon arrival timnigs :param t: 1d array of ints :param n_photons: 1d array of ints :param recoil_type: 1d array of ints :param config: dict wfsim config :param phase: str “liquid”

returns photon timing array

class wfsim.core.S2(config)

Bases: wfsim.core.Pulse

Given temperal inputs as well as number of electrons Random generate photon timing and channel distribution.

static electron_timings(t, n_electron, z, sc_gain, timings, gains, drift_velocity_liquid, drift_time_gate, diffusion_constant_longitudinal, electron_trapping_time)

Calculate arrival times of the electrons. Data is written to the timings and gains arrays :param t: 1d array of ints :param n_electron:1 d array of ints :param z: 1d array of floats :param sc_gain: secondairy scintallation gain :param timings: empty array with length sum(n_electron) :param gains: empty array with length sum(n_electron) :param drift_velocity_liquid, drift_time_gate, diffusion_constant_longitudinal, electron_trapping_time: configuration values

static get_electron_yield(n_electron, z_obs, config)

Drift electrons up to the gas interface and absorb them

Parameters

• n_electron – 1d array with ints as number of electrons

• z_obs – 1d array of floats with the observed z positions

• config – dict with wfsim config

returns 1d array ints with number of electrons

static get_s2_light_yield(positions, config, resource)

Calculate s2 light yield…

Parameters

• positions – 2d array of positions (floats)

• config – dict with wfsim config

• resource – instance of the resource class

returns array of floats (mean expectation)

static inverse_field_distortion(x, y, z, resource)

For 1T the pattern map is a data driven one so we need to reverse engineer field distortion into the simulated positions :param x: 1d array of float :param y: 1d array of float :param z: 1d array of float :param resource: instance of resource class returns z: 1d array, postions 2d array

static luminescence_timings_garfield(xy, shape, resource, config)

Luminescence time distribution computation according to garfield scintillation maps :param xy: 1d array with positions :param shape: tuple with nelectron,nphotons :param config: dict wfsim config :param resource: instance of wfsim resource

returns 1d (nested?) array with ints for photon timings

static luminescence_timings_simple(xy, n_electron, shape, config, resource)

Luminescence time distribution computation according to simple s2 model (many many many single electrons) :param xy: 1d array with positions :param n_electron: 1d array with ints for number f electrons :param shape: tuple with nelectron,nphotons :param config: dict wfsim config :param resource: instance of wfsim resource returns _luminescence_timings_simple

static photon_channels(n_electron, z_obs, positions, _photon_timings, _instruction, config, resource)

Set the _photon_channels property list of length same as _photon_timings

Parameters

• n_electron – a 1d int array

• z_obs – a 1d float array

• positions – a 2d float array of shape [n interaction, 2] for the xy coordinate

• _photon_timings – 1d int array of photon timings,

• _instructions – array of instructions with dtype wfsim.instructions_dtype

• config – dict wfsim config

• resource – instance of resource class

static photon_timings(t, n_electron, z, xy, sc_gain, config, resource, phase)

Generates photon timings for S2s. Returns a list of photon timings and instructions repeated for original electron

Parameters

• t – 1d float array arrival time of the electrons

• n_electron – 1d float array number of electrons to simulate

• z – float array. Z positions of s2

• xy – 1d float array, xy positions of s2

• sc_gain – float, secondairy s2 gain

• config – dict of the wfsim config

• resource – instance of the resource class

• phase – string, “gas”

static s2_pattern_map_diffuse(n_electron, z, xy, config, resource)

Returns an array of pattern of shape [n interaction, n PMTs] pattern of each interaction is an average of n_electron patterns evaluated at diffused position near xy. The diffused positions sample from 2d symmetric gaussian with spread scale with sqrt of drift time.

Parameters

• n_electron – a 1d int array

• z – a 1d float array

• xy – a 2d float array of shape [n interaction, 2]

wfsim.load_resource module

class wfsim.load_resource.DummyMap(const, shape=())

Bases: object

Return constant results the length match the length of input but from the second dimensions the shape is user defined input

reduce_last_dim()

class wfsim.load_resource.Resource(config=None)

Bases: object

Get the configs needed for running WFSim. Configs can be obtained in

two ways: 1. Get it directly from the mongo database. This only needs the

name of the file.

2. Load it with straxen get_resource, this can either:

   Download from a public repository Read from local cache Download from a private repository if credentials are properly setup

wfsim.load_resource.load_config(config)

Create a Resource instance from the configuration

Uses a cache to avoid re-creating instances from the same config

wfsim.load_resource.make_map(map_file, fmt='text')

Fetch and make an instance of InterpolatingMap based on map_file Alternativly map_file can be a list of [“constant dummy”, constant: int, shape: list] return an instance of DummyMap

wfsim.pax_interface module

class wfsim.pax_interface.PaxEventSimulator(config={})

Bases: object

Simulate wf from instruction and stored in wfsim.pax_datastructure.datastructure.Event mimicing pax.datastructure.Event Then pickled, compressed and saved mimicing pax raw data zips.

Call compute to start the simulation process.

class WriteZipped(config)

Bases: object

close_current_file()

Closes the currently open file, if there is one. Also handles temporary file renaming.

file_extension = 'zip'

open_new_file(first_event_number)

Opens a new file, closing any old open ones

write_event(event_proxy)

Write one more event to the folder, opening/closing files as needed

class WriteZippedEncoder(config)

Bases: object

static make_event_proxy(event, data, block_id=None)

transfer_event(event)

compute()

class wfsim.pax_interface.PaxEvents(config)

Bases: object

wfsim.raw_optical module

class wfsim.raw_optical.RawDataOptical(config)

Bases: wfsim.core.RawData

sim_primary(primary_pulse, instruction, channels, timings)

static symtype(ptype)

Little stupid we need this guy twice, but we need the different values for PULSE_TYPE_NAMES

wfsim.strax_interface module

class wfsim.strax_interface.ChunkRawRecords(config)

Bases: object

final_results()

source_finished()

class wfsim.strax_interface.ChunkRawRecordsOptical(config)

Bases: wfsim.strax_interface.ChunkRawRecords

class wfsim.strax_interface.McChainSimulator(strax_context, run_id)

Bases: object

Simulator wrapper class to be used for full chain simulator.

Expected fax_file input is a g4 root file. Does three things:

1. Process root file with epix 1b process neutron veto instructions & synchronize timings between nveto and tpc 2 Run simulation 3 Synchronize metadata of tpc and nveto to have the same run start and stop times

Usage:

simulator = wfsim.McChainSimulator(st,run_id) simulator.run_chain()

instructions_from_epix()

Run epix and save instructions as self.instruction. For the moment we’ll just process the whole file and then throw out all events we won’t use Epix needs to be imported in here to avoid circle imports

instructions_from_nveto()

run_chain()

Main function. Sets instructions needed, runs wfsim and synch metadata

run_strax(run_id)

Runs wfsim up to raw records for tpc and if requisted the nveto. All this setting neutron_veto=True/False is needed to get wfsim to do the right thing

set_configuration()

Set chunking configuration and feeds instructions to wfsim

set_instructions()

Gets instructions from epix and neutron veto if needed. Afterwards sets the correct timings and configuration

set_timing()

Set timing information in such a way to synchronize instructions for the TPC and nVeto

sync_meta(run_id)

Sync metadata outputs from the tpc and the nveto.

class wfsim.strax_interface.RawRecordsFromFax1T

Bases: wfsim.strax_interface.RawRecordsFromFaxNT

config: Dict

deps: Dict

dtype: Union[tuple, numpy.dtype, immutabledict.immutabledict, dict]

input_buffer: Dict[str, strax.chunk.Chunk]

provides: tuple = ('raw_records', 'truth')

run_i: int

run_id: str

class wfsim.strax_interface.RawRecordsFromFaxEpix

Bases: wfsim.strax_interface.RawRecordsFromFaxNT

check_instructions()

config: Dict

deps: Dict

dtype: Union[tuple, numpy.dtype, immutabledict.immutabledict, dict]

get_instructions()

input_buffer: Dict[str, strax.chunk.Chunk]

run_i: int

run_id: str

set_timing()

Set timing information in such a way to synchronize instructions for the TPC and nVeto

takes_config = immutabledict({'optical': <strax.config.Option object>, 'seed': <strax.config.Option object>, 'fax_file': <strax.config.Option object>, 'fax_config_override': <strax.config.Option object>, 'event_rate': <strax.config.Option object>, 'chunk_size': <strax.config.Option object>, 'nchunk': <strax.config.Option object>, 'right_raw_extension': <strax.config.Option object>, 'timeout': <strax.config.Option object>, 'fax_config': <strax.config.Option object>, 'gain_model': <strax.config.Option object>, 'detector': <strax.config.Option object>, 'channel_map': <strax.config.Option object>, 'n_tpc_pmts': <strax.config.Option object>, 'n_top_pmts': <strax.config.Option object>, 'neutron_veto': <strax.config.Option object>, 'wfsim_instructions': <strax.config.Option object>, 'epix_config': <strax.config.Option object>, 'event_start': <strax.config.Option object>, 'event_stop': <strax.config.Option object>})

class wfsim.strax_interface.RawRecordsFromFaxNT

Bases: wfsim.strax_interface.FaxSimulatorPlugin

check_instructions()

compute()

config: Dict

data_kind: Union[str, immutabledict.immutabledict, dict] = immutabledict({'raw_records': 'raw_records', 'raw_records_he': 'raw_records_he', 'raw_records_aqmon': 'raw_records_aqmon', 'truth': 'truth'})

deps: Dict

dtype: Union[tuple, numpy.dtype, immutabledict.immutabledict, dict]

get_instructions()

infer_dtype()

Return dtype of computed data; used only if no dtype attribute defined

input_buffer: Dict[str, strax.chunk.Chunk]

provides: tuple = ('raw_records', 'raw_records_he', 'raw_records_aqmon', 'truth')

run_i: int

run_id: str

class wfsim.strax_interface.RawRecordsFromFaxOptical

Bases: wfsim.strax_interface.RawRecordsFromFaxNT

config: Dict

deps: Dict

dtype: Union[tuple, numpy.dtype, immutabledict.immutabledict, dict]

get_instructions()

input_buffer: Dict[str, strax.chunk.Chunk]

run_i: int

run_id: str

class wfsim.strax_interface.RawRecordsFromFaxnVeto

Bases: wfsim.strax_interface.RawRecordsFromFaxOptical

check_instructions()

compute()

config: Dict

data_kind: Union[str, immutabledict.immutabledict, dict] = immutabledict({'raw_records_nv': 'raw_records_nv', 'truth': 'truth'})

deps: Dict

dtype: Union[tuple, numpy.dtype, immutabledict.immutabledict, dict]

get_instructions()

input_buffer: Dict[str, strax.chunk.Chunk]

provides: tuple = ('raw_records_nv', 'truth')

run_i: int

run_id: str

set_config()

takes_config = immutabledict({'optical': <strax.config.Option object>, 'seed': <strax.config.Option object>, 'fax_file': <strax.config.Option object>, 'fax_config_override': <strax.config.Option object>, 'event_rate': <strax.config.Option object>, 'chunk_size': <strax.config.Option object>, 'nchunk': <strax.config.Option object>, 'right_raw_extension': <strax.config.Option object>, 'timeout': <strax.config.Option object>, 'fax_config': <strax.config.Option object>, 'gain_model': <strax.config.Option object>, 'detector': <strax.config.Option object>, 'channel_map': <strax.config.Option object>, 'n_tpc_pmts': <strax.config.Option object>, 'n_top_pmts': <strax.config.Option object>, 'neutron_veto': <strax.config.Option object>, 'wfsim_nveto_instructions': <strax.config.Option object>, 'wfsim_nveto_channels': <strax.config.Option object>, 'wfsim_nveto_timings': <strax.config.Option object>, 'fax_config_nveto': <strax.config.Option object>})

wfsim.units module

Define unit system for pax (i.e., seconds, etc.) This sets up variables for the various unit abbreviations, ensuring we always have a ‘consistent’ unit system. There are almost no cases that you should change this without talking with a maintainer.

wfsim.utils module

wfsim.utils.find_intervals_below_threshold(w, threshold, holdoff, result_buffer)

Fills result_buffer with l, r bounds of intervals in w < threshold. :param w: Waveform to do hitfinding in :param threshold: Threshold for including an interval :param result_buffer: numpy N*2 array of ints, will be filled by function.

if more than N intervals are found, none past the first N will be processed.

:returns : number of intervals processed Boundary indices are inclusive, i.e. the right boundary is the last index which was < threshold

Module contents