APIdocs: cminject package¶

All the library code for CMInject lives in this module.

Subpackages¶

Submodules¶

cminject.base module¶

The most fundamental of definitions, for deriving other base classes from.

class cminject.base.Action¶

Bases: abc.ABC

An object to update a Particle’s properties based on its current state. Can do arbitrary calculations to determine the values the properties should be set to.

Warning

If the properties that a .__call__() call changes include the .position attribute, the __call__ method MUST return True, and it SHOULD return False otherwise. If your action doesn’t adhere to this, expect these updated positions to have no effect, since the integrator will overwrite them.

Should be used for all calculations of additional quantities beyond calculating an acceleration, so, beyond what a Field returns. For example, TrackTrajectory is a simple Action that just takes the position and appends it to the trajectory property on the particle.

The __call__ method is guaranteed by Experiment to be called exactly once per time step (as long as the particle is still being simulated).

If you need access to properties of fields, detectors, etc., override the __init__ method and store a reference to each relevant object at construction of the Action instance.

Note

Taking care to avoid data clashes (different actions writing on the same property on a particle, overwriting each others’ results) is up to the user / implementer. This cannot be avoided in a general way. If you know that another Action instance is writing to a specific particle property, don’t overwrite it unless you know the order of execution of the Actions and are doing this completely intentionally.

abstract __call__(particle: cminject.base.Particle, time: float) → bool¶

Does some thing with a Particle instance. Must return True if the particle’s phase_space_position is changed, or if its position or velocity is changed (since these two properties are directly derived from phase_space_position). Note that a “change” entails overwriting the whole property as well as changing only parts of it.

Parameters

particle – The Particle instance.
time – The current time.

Returns

True if the phase space position of a particle was changed in any way. False otherwise.

class cminject.base.Boundary¶

Bases: cminject.base.ZBounded, abc.ABC

Can tell whether a Particle is inside of it or not.

Making the total set of Boundary objects in an experiment closed is - for now - the job of the person implementing the experiment setup.

abstract is_particle_inside(position: numpy.array, time: float) → bool¶

Tells whether the given position is inside this boundary at the given time.

Parameters

position – A spatial position to tell this for.
time – The time to tell this for.

Returns

True if the particle is definitely inside this Boundary, False if it is not inside this Boundary (or if this is unknown).

class cminject.base.Detector(identifier: str)¶

Bases: cminject.base.ZBounded, abc.ABC

Can tell whether a Particle has hit it, and _if_ it has hit it, can also tell where this occurred.

As opposed to real detectors like fluorescent screens, these Detectors must never have an effect on particles. This allows us to, for instance, make an image of a particle distribution at different places of the experiment, without terminating the path of any particle.

Detector subclasses can - and should - freely choose how to internally do the calculation of the hit position.

try_to_detect(particle: cminject.base.Particle) → bool¶

Tries to detect a particle.

If the particle is considered having hit the detector (“could be detected”) based on its current state, then a DetectorHit is stored on the particle’s detector_hits property and True is returned. Otherwise, this function has no effect and returns False.

Parameters: particle – A Particle instance.
Returns: True if the particle has reached this detector, False otherwise.

class cminject.base.Device(fields: List[cminject.base.Field], boundary: cminject.base.Boundary, actions: Optional[List[cminject.base.Action]] = None)¶

Bases: cminject.base.ZBounded, cminject.utils.global_config.ConfigSubscriber, abc.ABC

A combination of Fields, Actions and a Boundary. Used to model real-world devices in an experiment setup.

property actions: List[cminject.base.Action]¶: The list of actions (Action instances) that this Device contains.

add_action(action: cminject.base.Action) → None¶

Adds an action (Action instance) to this Device.

Parameters: action – The action to add.

add_field(field: cminject.base.Field) → None¶

Adds a field (Field instance) to this Device.

Parameters: field – The field to add.

property boundary¶

The boundary (Boundary instance) that this device has.

Returns: The current boundary of this device.

calculate_acceleration(particle: cminject.base.Particle, time: float) → numpy.array¶

Calculates and returns the acceleration that this Device exerts on a Particle particle at time time.

Parameters

particle – The Particle to calculate the acceleration for.
time – The time to calculate the acceleration for.

Returns

The acceleration vector.

config_change(key: cminject.utils.global_config.ConfigKey, value: Any)¶

Will be called whenever the value of any subscribed key changes. Will be called once at the time of subscribing, IF the value for the subscribed key(s) is not None.

Parameters

key – The ConfigKey that the change occurred for.
value – The new value of the configuration value stored for the key key.

Returns

Nothing (unused).

property fields: List[cminject.base.Field]¶: The list of fields (Field instances) that this Device contains.

is_particle_inside(particle_position: numpy.array, time: float) → bool¶

Returns whether the passed Particle is inside this Device or not.

Parameters

particle_position – The particle’s spatial position.
time – The current time.

Returns

True if the Particle inside this Device, False if it is not (or if this is unknown).

run_actions(particle: cminject.base.Particle, time: float) → bool¶: Runs all Actions stored on this Device, if the passed Particle is inside this device. Returns True if any of them changed the particle position. :param particle: The Particle that all of this Device’s Actions will be run for. :param time: The current time. :return: True if any of the Actions returned True, False otherwise.

property z_boundary: Tuple[float, float]¶: The Z boundary (z_min, z_max) of this device.

class cminject.base.Field¶

Bases: cminject.base.ZBounded, abc.ABC

A virtual acceleration field. Interacts with particles by exerting an acceleration on them, based on some collection of (local and current) properties of the particle and the field.

abstract calculate_acceleration(particle: cminject.base.Particle, time: float) → numpy.array¶

Calculates an acceleration for one particle based on the particle’s current properties and the current time. This acceleration will be integrated for in each time step and thus “applied” to the particle.

Parameters

particle – The Particle to calculate this Field’s acceleration for.
time – The time to calculate the acceleration for.

Returns

A (n,)-shaped numpy array describing the acceleration exerted on the particle. n is the number of spatial dimensions of the experiment.

class cminject.base.Particle(identifier: int, start_time: float, position: numpy.array, velocity: numpy.array, *args, **kwargs)¶

Bases: abc.ABC

Describes a particle whose trajectory we want to simulate. It is first and foremost a data container, and it and its subclasses should be written and used as such.

It can be read by any part of the code, but should only be written to by instances of cminject.experiment.Experiment and cminject.base.Action (or instances of their subclasses).

This class declares a few basic properties that we expect every particle to have:

identifier (a unique integer id),

lost (a flag storing whether the particle is considered lost)

position (the particle’s position)

velocity (the particle’s velocity)

mass (the particle’s mass)

trajectory (a list describing points in the particle’s path)

detector_hits (a dict mapping detector identifiers to hit lists)

as_array(which) → numpy.array¶

Returns a NumPy array representation of the current state of this particle. Either the tracked properties, or the constant properties, or all (tracked + constant) can be returned as a single NumPy array with a structured NumPy datatype.

Parameters

which –

The part of this particle’s state to return as an array. Must be one of [‘tracked’, ‘constant’, ‘all’]:

’tracked’ will refer to the implementation of tracked_properties,

’constant’ will refer to the implementation of constant_properties,

’all’ will refer to both and concatenate them in order (tracked + constant).

as_dtype(which) → numpy.array¶

Returns the dtype that the array returned by as_array() will have, when called with the same parameters.

Parameters: which – The part of this particle’s state to return. Refer to the docstring about as_array.

constant_properties¶: The definition of the particle’s constant properties, i.e. properties that do NOT change along the trajectory of the particle. Must return a list in NumPy structured dtype format).

property initial_tracked_properties¶: An array representing the state of all tracked properties of this particle, at its time of creation.

mass¶: The mass of the particle.

Warning

Please ensure that you also use the @cached_property decorator when overriding this property, so that the mass calculation is done exactly once and the result is stored, for (time) efficiency.

property position¶: The spatial position of the particle.

property reached_any_detector: bool¶: Whether this particle has ever reached any detector.

tracked_properties¶: The definition of the particle’s tracked properties, i.e. properties that are changing along the trajectory of the particle. Must return a list in NumPy structured dtype format).

property velocity¶: The spatial velocity of the particle.

class cminject.base.ParticleDetectorHit(hit_position: numpy.array, particle: cminject.base.Particle)¶

Bases: object

A small data container class, to store required information about a detection event of a particle by a detector.

Contains hit_state, which is a full description of the particle’s state, with the ‘position’ part overwritten by the position the detector detected the particle at.

property hit_state: numpy.array¶

The state of the particle in the moment where the detector detected the particle. The ‘position’ part of this state refers to the (approximate) position of detection as returned by the detector.

Returns: The position of the hit.

class cminject.base.ResultStorage¶

Bases: abc.ABC

An object to store the results of an experiment in some fashion. MUST implement store_results, and MAY implement convenience methods to read from the storage again (e.g. a method to get all particle trajectories from a file).

abstract get_detectors() → Optional[Dict[str, Iterable]]¶

The collection of all detectors.

Returns: The detectors, in a dictionary mapping their identifying string to their collection of detected hits.

Note

Detectors that did not detect at least one particle are not necessarily part of the output of this method, and that if there were no detectors or no detector detected at least one particle, None is returned.

abstract get_dimensions() → int¶: The number of spatial dimensions that the stored simulation had.

abstract get_properties() → Optional[Iterable]¶

The collection of constant particle properties, i.e., the values of their Particle.constant_properties().

Returns: See above; None if these weren’t stored.

abstract get_tracked_final() → Optional[Iterable]¶

The collection of final particle states, i.e., the final values of their Particle.tracked_properties().

Returns: See above; None if these weren’t stored.

abstract get_tracked_initial() → Optional[Iterable]¶

The collection of initial particle states, i.e., the initial values of their Particle.tracked_properties().

Returns: See above; None if these weren’t stored.

abstract get_trajectories() → Optional[Iterable]¶

The collection of all particles’ trajectories.

Returns: See above; None if these weren’t stored.

abstract store_results(particles: List[cminject.base.Particle])¶

Stores the results of an experiment (which are always a list of modified Particle instances).

Parameters: particles – The list of particles, each in the state of after running a simulation.

class cminject.base.Setup¶

Bases: abc.ABC

A base class for classes that define experiment setups. Should be considered static, i.e. its method are all “@staticmethod”s, and no instances should (need to be) created.

abstract static construct_experiment(main_args: argparse.Namespace, args: argparse.Namespace) → Experiment¶

Constructs an Experiment instance from the arguments to the main program and this specific setup.

Parameters

main_args – The argparse.Namespace parsed by the main program. Contains args that are common to all setups, like the number of particles and timespan.
args – The argparse.Namespace parsed by this class’s parse_args method.

Returns

An Experiment instance that’s ready to be ran.

abstract static get_parser() → cminject.utils.args.SetupArgumentParser¶

Returns a parser for the arguments relevant to this specific setup. This parser will receive only the arguments that the main program (bin/cminject) did not recognise.

Returns: An argparse.ArgumentParser (or subclass) instance that can parse all the args relevant to this setup.

static validate_args(args: argparse.Namespace)¶

Validates the arguments. Useful for validation that needs to check multiple args at once, and not just one specific arg. Overriding this method is optional and only needs to be done if such validation is desired.

Parameters: args – The argparse.Namespace object that the parser constructed by get_parser() returned after being given all arguments that the main program did not recognise.
Raises: argparse.ArgumentError if validation failed

class cminject.base.Source¶

Bases: abc.ABC

A source of particles, to generate an initial position (and property) distribution.

One can implement sources following different distributions and generating different particle types by subclassing this class and implementing generate_particles however they want.

Warning

Any implementation of Source should only generate exactly one type of particle, i.e. its generate_particles method should never return a list of particles where the (most specific) type of any two particles is different.

abstract generate_particles(start_time: float = 0.0) → List[cminject.base.Particle]¶

Generates a list of particles. How this is done is entirely up to the subclass, by (this is mandatory!) implementing this method in some way.

Returns: A list of Particle instances.

class cminject.base.ZBounded¶

Bases: abc.ABC

A mixin for objects in an experiment setup that are bounded in the Z direction. Classes deriving from this mixin will have to implement the get_z_boundary method.

Objects implementing this mixin should be used to determine the minimum/maximum extent of the entire experiment setup in the Z direction, and then allows an Experiment to make a fast preliminary calculation about whether it can consider a particle lost based on this.

abstract property z_boundary: Tuple[float, float]¶

Returns the Z boundary of this Z-bounded object.

Returns: A tuple of floats, the first entry being z_min, the second being z_max.

cminject.experiment module¶

Code for defining and running a virtual Experiment, by using numerical integration, and parallelization via the multiprocessing module.

class cminject.experiment.Experiment(number_of_dimensions: int, time_interval: Tuple[float, float], time_step: float = 1e-05, z_boundary: Optional[Tuple[float, float]] = None, random_seed=None, result_storage: Optional[cminject.base.ResultStorage] = None)¶

Bases: object

A description of a virtual experiment that can be simulated. Consists of:

cminject.base.Source s, generating cminject.base.Particle s

cminject.base.Device s affecting particles via acceleration by their cminject.base.Field s, decisions made about whether to keep simulating particles by each Device’s cminject.base.Boundary, and generic effects occurring after each time step by their cminject.base.Action s.

cminject.base.Detector s detecting particles

cminject.base.Action s that are indiscriminately run for every particle after every integration step

Constants relevant to the simulation logic, like the time-interval and time-step, seed to use for random generation, the boundary in Z direction, …

A cminject.base.ResultStorage that will store the simulation results.

All of the above are optional, but of course running an “empty” experiment will return an empty result, and an experiment containing particles but no cminject.base.Device s, cminject.base.Action s etc. will return ‘boring’ results.

add_action(action: cminject.base.Action)¶

Adds a Action to this experiment.

Parameters: action – The Action instance to add.

add_detector(detector: cminject.base.Detector)¶

Adds a Detector to this experiment.

Parameters: detector – The Detector instance to add.

add_device(device: cminject.base.Device)¶

Adds a Device to this experiment.

Parameters: device – The Device instance to add.

add_source(source: cminject.base.Source)¶

Adds a Source to this experiment.

Parameters: source – The Source instance to add.

run(single_process: bool = False, chunksize: Optional[int] = None, processes: Optional[int] = None, loglevel: str = 'warning', progressbar: bool = False) → List[cminject.base.Particle]¶

Run the Experiment. A list of resulting Particle instances is returned.

Parameters

single_process – Whether to run this experiment on a single process, i.e. without parallelization. False by default: It’s mostly only useful to pass True for developers, as many debuggers, profilers, etc. are not able to deal well with different processes. To run the experiment and get results quickly, leave this False.
chunksize – The chunk size for pool.imap_unordered. None by default, which equates to 1. This parameter has no effect if single_threaded is True.
processes – The number of processes to use for multiprocessing. Handed directly to multiprocessing.Pool(), so check the documentation there for any further info. Equal to the number of CPU cores by default.
loglevel – The loglevel to run the program with. One of {DEBUG, INFO, WARNING, ERROR, CRITICAL} – see Python’s builtin logging module for more explanation regarding these levels.
progressbar – Show simulation progress in a readable progress bar (by tqdm).

Returns

A list of resulting Particle instances. Things like detector hits and trajectories should be stored on them and can be read off each Particle.

set_integrator(integrator: str, **integrator_params) → None¶

Sets the integrator to use for this Experiment. Refer to SciPy’s scipy.integrate.ode for documentation on the available integrators and their options.

Parameters

integrator – The integrator to use (identified by a string).
integrator_params – Additional parameters to pass onto scipy.integrate.ode.set_integrator().

property time_interval¶: The time interval this experiment will be run within (for every particle)

property time_step¶: The time step to use for numerical simulation (in seconds).

property z_boundary¶: The full boundary in the Z dimension of this experiment

exception cminject.experiment.ImpossiblePropagationException¶

Bases: Exception

Can occur when a Particle could not be propagated due to a logical error, e.g. when trying to propagate a particle with a velocity of exactly 0 in all directions.

cminject.experiment.get_particle_simulation_state(particle_position: numpy.array, time: float, z_boundary: Tuple[float, float], devices: List[cminject.base.Device]) → int¶

Decides whether a particle should be considered lost.

Parameters

particle_position – The particle’s position.
time – The current simulation time in seconds
z_boundary – The total Z boundary to consider
devices – The list of devices that the particle might potentially be in

Returns

cminject.experiment.postprocess_integration_step(particle: cminject.base.Particle, integrator: scipy.integrate._ode.ode, position_mismatch: bool, time_mismatch: bool) → None¶

Does post-processing after an integration step, which consists of:

running all actions
running all detectors
IF particle position was changed or new_time was passed: resetting the integrator

Parameters

particle – The particle which experienced an integration step
integrator – The integrator (scipy.integrate.ode) instance.
position_mismatch – True if the position of the particle has been changed relative to the integrator result before this function was called. False if it has not.
time_mismatch – True if the time of the particle has been changed relative to the integrator’s time before this function was called. False if it has not.

cminject.experiment.postprocess_particle(particle: cminject.base.Particle, integrator: scipy.integrate._ode.ode, t_end: float) → None¶

Does post-processing after a full simulation of a single particle’s trajectory, by:

running all actions
running all detectors
logging (for logging.INFO and more verbose loglevels) information about how the simulation ended

Parameters

particle – The fully simulated particle to postprocess.
integrator – The integrator (scipy.integrate.ode) instance.
t_end – The maximum time this particle could have been simulated for. Used only for logging.

cminject.experiment.propagate_field_free(particle: cminject.base.Particle) → None¶

Propagates a particle assuming it is in a field-free region, up until the next valid point in Z direction.

“Z direction” is always the last dimension, assumed to be the main propagation axis.

“the next valid point” is calculated from the z_boundary properties of all devices and detectors.

Propagation is done according to a simple rule-of-three calculation.

Parameters: particle – The Particle instance to propagate.
Raises: ImpossiblePropagationException – if asked to propagate the particle in an impossible way, e.g. propagating until the next point in positive Z when the particle’s v_Z is negative.

cminject.experiment.simulate_particle(particle: cminject.base.Particle, integrator: str = 'lsoda', integrator_params: Dict = {}) → cminject.base.Particle¶

Simulates the flight path of a single particle – the “meat” of the trajectory simulations.

The description of the experiment is given implicitly, by global variables instead of as parameters. This is done for efficiency reasons when using multiprocessing. See the note in this docstring for further information.

Parameters

particle – The particle instance.
integrator – The integrator to use. Passed directly to scipy.integrate.ode.set_integrator() as the name parameter.
integrator_params – Additional parameters to pass on to scipy.integrate.ode.set_integrator() as the integrator_params parameter.

Returns

A modified version of the particle instance after the simulation has ended.

Note

There are implicit parameters, as global variables. This is done to avoid these (unchanging) objects being serialized and deserialized for each particle that is being simulated, see this blog post: https://confluence.desy.de/display/PARTI/2019/08/08/CMInject%3A+Performance+considerations

For more detailed info on this issue in general, see: https://thelaziestprogrammer.com/python/a-multiprocessing-pool-pickle

For the way this is worked around using global variables and a Pool initializer, see: https://stackoverflow.com/a/10118250/3090225

This greatly improves performance, as we’re not doing unnecessary I/O during which we can’t do any actual calculations. Below is the list of global variables that used to be parameters bound via functools.partial, but are now global variables that are used within the function:

DEVICES: The list of devices.
DETECTORS: The list of detectors.
ACTIONS: The list of actions.
TIME_INTERVAL: The time interval of the simulation.
Z_BOUNDARY: The Z boundary of the simulation.
NUMBER_OF_DIMENSIONS: The number of dimensions of the space the particle moves in.
BASE_SEED: The “base seed”, i.e. the random seed the experiment was defined with, to derive a local random seed from.

cminject.experiment.spatial_derivatives(time: float, phase_space_position: numpy.array, particle: cminject.base.Particle, devices: List[cminject.base.Device], number_of_dimensions: int) → numpy.array¶

Calculates the derivatives of position and velocity, which will then be integrated over. “n” as used below is the number of the simulation’s spatial dimensions.

Parameters

time – The current simulation time in seconds
phase_space_position – A (2n,) numpy array containing the phase space position..
particle – The Particle instance.
devices – The list of devices that (might) affect this particle.
number_of_dimensions – The number of dimensions of the space the particle moves in.

Returns

A (2n,) numpy array containing velocity and acceleration of the particle.