CAiMIRA source code

Subpackages

Submodules

caimira.dataclass_utils module

nested_getattr(obj, name: str)

Get an attribute on a dataclass, much like getattr, except it supports nested attributes definitions. For example:

>>> nested_getattr(obj, 'attr1.sub_attr2.sub_sub_attr3')

nested_replace(obj, new_values: Dict[str, Any])

Replace an attribute on a dataclass, much like dataclasses.replace, except it supports nested replacement definitions. For example:

>>> new_obj = nested_replace(obj, {'attr1.sub_attr2.sub_sub_attr3': 4})
>>> new_obj.attr1.sub_attr2.sub_sub_attr3
4

replace(obj, **changes)

A version of dataclasses.replace that handles ClassVar declarations.

See https://bugs.python.org/issue33796.

walk_dataclass(model, name='')

Recursively walk a dataclass instance, generating (name, obj) pairs for attributes and decending into nested dataclasses.

>>> list(walk_dataclass(obj), 'my_obj')
[('my_obj.attr_a', <dataclass instance>), ('my_obj.attr_a.sub_attr', <dataclass instance>)]

caimira.models module

This module implements the core CAiMIRA models.

The CAiMIRA model is a flexible, object-oriented numerical model. It is designed to allow the user to swap-out and extend its various components. One of the major abstractions of the model is the distinction between virus concentration (ConcentrationModel) and virus exposure (ExposureModel).

The concentration component is a recursive (on model time) model and therefore in order to optimise its execution certain layers of caching are implemented. This caching mandates that the models in this module, once instantiated, are immutable and deterministic (i.e. running the same model twice will result in the same answer).

In order to apply stochastic / non-deterministic analyses therefore you must introduce the randomness before constructing the models themselves; the caimira.monte_carlo module is a good example of doing this - that module uses the models defined here to allow you to construct a ConcentrationModel containing parameters which are expressed as probability distributions. Under the hood the caimira.monte_carlo.ConcentrationModel implementation simply samples all of those probability distributions to produce many instances of the deterministic model.

The models in this module have been designed for flexibility above performance, particularly in the single-model case. By using the natural expressiveness of Python we benefit from a powerful, readable and extendable implementation. A useful feature of the implementation is that we are able to benefit from numpy vectorisation in the case of wanting to run multiple-parameterisations of the model at the same time. In order to benefit from this feature you must construct the models with an array of parameter values. The values must be either scalar, length 1 arrays, or length N arrays, where N is the number of parameterisations to run; N must be the same for all parameters of a single model.

class Activity(inhalation_rate: float | numpy.ndarray, exhalation_rate: float | numpy.ndarray)

Bases: object

exhalation_rate: float | ndarray: Exhalation rate in m^3/h

inhalation_rate: float | ndarray: Inhalation rate in m^3/h

types: ClassVar[Dict[str, Activity]] = {'Heavy exercise': Activity(inhalation_rate=3.3, exhalation_rate=3.3), 'Light activity': Activity(inhalation_rate=1.25, exhalation_rate=1.25), 'Moderate activity': Activity(inhalation_rate=1.78, exhalation_rate=1.78), 'Seated': Activity(inhalation_rate=0.51, exhalation_rate=0.51), 'Standing': Activity(inhalation_rate=0.57, exhalation_rate=0.57)}: Pre-populated examples of Activities.

class AirChange(active: caimira.models.Interval, air_exch: float | numpy.ndarray)

Bases: Ventilation

active: Interval: The interval in which the ventilation is operating.

air_exch: float | ndarray

air_exchange(room: Room, time: float) → float | ndarray

Returns the rate at which air is being exchanged in the given room at a given time (in hours).

Note that whilst the time is known inside this function, it may not be used to vary the result unless the specific time used is declared as part of a state change in the interval (e.g. when air_exchange == 0).

class CO2ConcentrationModel(room: Room, ventilation: _VentilationBase, CO2_emitters: SimplePopulation, CO2_atmosphere_concentration: float = 440.44, CO2_fraction_exhaled: float = 0.042)

Bases: _ConcentrationModelBase

Class used for the computation of the CO2 concentration.

CO2_atmosphere_concentration: float = 440.44: CO2 concentration in the atmosphere (in ppm)

CO2_emitters: SimplePopulation: Population in the room emitting CO2

CO2_fraction_exhaled: float = 0.042: CO2 fraction in the exhaled air

min_background_concentration() → float | ndarray: Background CO2 concentration in the atmosphere (in ppm)

normalization_factor() → float | ndarray: Normalization factor (in the same unit as the concentration). This factor is applied to the normalized concentration only at the very end.

property population: SimplePopulation: Population in the room (the emitters of what we compute the concentration of)

removal_rate(time: float) → float | ndarray: Remove rate of the species considered, in h^-1

class Cases(geographic_population: int = 0, geographic_cases: int = 0, ascertainment_bias: int = 0)

Bases: object

The geographical data to calculate the probability of having at least 1 new infection in a probabilistic exposure.

ascertainment_bias: int = 0: Number of new cases confidence level

geographic_cases: int = 0: Geographic location new cases

geographic_population: int = 0: Geographic location population

probability_meet_infected_person(virus: Virus, n_infected: int, event_population: int) → float | ndarray: Probability to meet n_infected persons in an event. From https://doi.org/10.1038/s41562-020-01000-9.

probability_random_individual(virus: Virus) → float | ndarray: Probability that a randomly selected individual in a focal population is infected.

class ConcentrationModel(room: Room, ventilation: _VentilationBase, infected: InfectedPopulation, evaporation_factor: float = 0.3)

Bases: _ConcentrationModelBase

Class used for the computation of the long-range virus concentration.

evaporation_factor: float = 0.3

infected: InfectedPopulation: Infected population in the room, emitting virions

infectious_virus_removal_rate(time: float) → float | ndarray

normalization_factor() → float | ndarray: Normalization factor (in the same unit as the concentration). This factor is applied to the normalized concentration only at the very end.

property population: InfectedPopulation: Population in the room (the emitters of what we compute the concentration of)

removal_rate(time: float) → float | ndarray: Remove rate of the species considered, in h^-1

property virus: Virus

class EmittingPopulation(number: int | caimira.models.IntPiecewiseConstant, presence: caimira.models.Interval | None, activity: caimira.models.Activity, mask: caimira.models.Mask, host_immunity: float, virus: caimira.models.Virus, known_individual_emission_rate: float)

Bases: _PopulationWithVirus

aerosols(): Total volume of aerosols expired per volume of exhaled air (mL/cm^3). Here arbitrarily set to 1 as the full emission rate is known.

emission_rate_per_aerosol_per_person_when_present() → float | ndarray: The emission rate of virions in the expired air per mL of respiratory fluid, per person, if the infected population is present, in (virion.cm^3)/(mL.h). This method includes only the diameter-independent variables within the emission rate. It should not be a function of time.

known_individual_emission_rate: float: The emission rate of a single individual, in virions / h.

class Expiration(diameter: float | ndarray, cn: float = 1.0)

Bases: _ExpirationBase

Model for the expiration. For a given diameter of aerosol, provides the aerosol volume, weighted by the mask outward efficiency when applicable.

aerosols(mask: Mask): Total volume of aerosols expired per volume of exhaled air. Result is in mL.cm^-3

cn: float = 1.0

diameter: float | ndarray: diameter of the aerosol in microns

jet_origin_concentration(): Concentration of viruses at the jet origin (mL/m3).

property particle: Particle: The Particle object representing the aerosol

class ExposureModel(concentration_model: ConcentrationModel, short_range: Tuple[ShortRangeModel, ...], exposed: Population, geographical_data: Cases, repeats: int = 1)

Bases: object

Represents the exposure to a concentration of virions in the air.

concentration(time: float) → float | ndarray

Virus exposure concentration, as a function of time.

It considers the long-range concentration with the contribution of the short-range concentration.

concentration_model: ConcentrationModel: The virus concentration model which this exposure model should consider.

deposited_exposure() → float | ndarray: The number of virus per m^3 deposited on the respiratory tract.

deposited_exposure_between_bounds(time1: float, time2: float) → float | ndarray

The number of virus per m^3 deposited on the respiratory tract between any two times.

Considers a contribution between the short-range and long-range exposures: It calculates the deposited exposure given a short-range interaction (if any). Then, the deposited exposure given the long-range interactions is added to the initial deposited exposure.

expected_new_cases() → float | ndarray

exposed: Population: The population of non-infected people to be used in the model.

geographical_data: Cases: Geographical data

infection_probability() → float | ndarray

long_range_deposited_exposure_between_bounds(time1: float, time2: float) → float | ndarray

long_range_fraction_deposited() → float | ndarray: The fraction of particles actually deposited in the respiratory tract (over the total number of particles). It depends on the particle diameter.

population_state_change_times() → List[float]: All time dependent population entities on this model must provide information about the times at which their state changes.

repeats: int = 1: The number of times the exposure event is repeated (default 1).

reproduction_number() → float | ndarray: The reproduction number can be thought of as the expected number of cases directly generated by one infected case in a population.

short_range: Tuple[ShortRangeModel, ...]: The list of short-range models which this exposure model should consider.

total_probability_rule() → float | ndarray

class HEPAFilter(active: caimira.models.Interval, q_air_mech: float | numpy.ndarray)

Bases: Ventilation

active: Interval: The interval in which the HEPA filter is operating.

air_exchange(room: Room, time: float) → float | ndarray

Returns the rate at which air is being exchanged in the given room at a given time (in hours).

Note that whilst the time is known inside this function, it may not be used to vary the result unless the specific time used is declared as part of a state change in the interval (e.g. when air_exchange == 0).

q_air_mech: float | ndarray

class HVACMechanical(active: caimira.models.Interval, q_air_mech: float | numpy.ndarray)

Bases: Ventilation

active: Interval: The interval in which the mechanical ventilation (HVAC) is operating.

air_exchange(room: Room, time: float) → float | ndarray

Returns the rate at which air is being exchanged in the given room at a given time (in hours).

Note that whilst the time is known inside this function, it may not be used to vary the result unless the specific time used is declared as part of a state change in the interval (e.g. when air_exchange == 0).

q_air_mech: float | ndarray

class HingedWindow(active: Interval, outside_temp: PiecewiseConstant, window_height: float | ndarray, opening_length: float | ndarray, number_of_windows: int = 1, min_deltaT: float = 0.1, window_width: float | ndarray = 0.0)

Bases: WindowOpening

Top-hung or bottom-hung hinged window (with the hinge parallel to horizontal plane).

property discharge_coefficient: float | ndarray: Simple model to compute discharge coefficient for top or bottom hung hinged windows, in the absence of empirical test results from manufacturers. From an excel spreadsheet calculator (Richard Daniels, Crawford Wright, Benjamin Jones - 2018) from the UK government - see Section 8.3 of BB101 and Section 11.3 of ESFA Output Specification Annex 2F on Ventilation opening areas.

window_width: float | ndarray = 0.0: Window width (m).

class InfectedPopulation(number: int | caimira.models.IntPiecewiseConstant, presence: caimira.models.Interval | None, activity: caimira.models.Activity, mask: caimira.models.Mask, host_immunity: float, virus: caimira.models.Virus, expiration: caimira.models._ExpirationBase)

Bases: _PopulationWithVirus

aerosols(): Total volume of aerosols expired per volume of exhaled air (mL/cm^3).

emission_rate_per_aerosol_per_person_when_present() → float | ndarray: The emission rate of virions in the expired air per mL of respiratory fluid, if the infected population is present, in (virion.cm^3)/(mL.h). This method includes only the diameter-independent variables within the emission rate. It should not be a function of time.

expiration: _ExpirationBase: The type of expiration that is being emitted whilst doing the activity.

fraction_of_infectious_virus() → float | ndarray: The fraction of infectious virus.

property particle: Particle: The Particle object representing the aerosol - here the default one

class IntPiecewiseConstant(transition_times: Tuple[float, ...], values: Tuple[int, ...])

Bases: PiecewiseConstant

value(time) → float | ndarray

values: Tuple[int, ...]: values of the function between transitions

class Interval

Bases: object

Represents a collection of times in which a “thing” happens.

The “thing” may be when an action is taken, such as opening a window, or entering a room.

Note that all intervals are open at the start, and closed at the end. So a simple start, stop interval follows:

start < t <= end

boundaries() → Tuple[Tuple[Time_t, Time_t], ...] | Tuple

transition_times() → Set[float]

triggered(time: float) → bool: Whether the given time falls inside this interval.

class Mask(η_inhale: float | numpy.ndarray, η_exhale: NoneType | float | numpy.ndarray = None, factor_exhale: float = 1.0)

Bases: object

exhale_efficiency(diameter: float | ndarray) → float | ndarray: Overall exhale efficiency, including the effect of the leaks. See CERN-OPEN-2021-004 (doi: 10.17181/CERN.1GDQ.5Y75), and Ref. therein (Asadi 2020). Obtained from measurements of filtration efficiency and of the leakage through the sides. Diameter is in microns.

factor_exhale: float = 1.0: Global factor applied to filtration efficiency of masks when exhaling.

inhale_efficiency() → float | ndarray: Overall inhale efficiency, including the effect of the leaks.

types: ClassVar[Dict[str, Mask]] = {'Cloth': Mask(η_inhale=0.225, η_exhale=0.35, factor_exhale=1.0), 'FFP2': Mask(η_inhale=0.865, η_exhale=None, factor_exhale=1.0), 'No mask': Mask(η_inhale=0, η_exhale=0, factor_exhale=1.0), 'Type I': Mask(η_inhale=0.5, η_exhale=None, factor_exhale=1.0)}: Pre-populated examples of Masks.

η_exhale: None | float | ndarray = None: Filtration efficiency of masks when exhaling.

η_inhale: float | ndarray: Filtration efficiency of masks when inhaling.

class MultipleExpiration(expirations: Tuple[_ExpirationBase, ...], weights: Tuple[float, ...])

Bases: _ExpirationBase

Represents an expiration of aerosols. Group together different modes of expiration, that represent each the main expiration mode for a certain fraction of time (given by the weights). This class can only be used with single diameters defined in each expiration (it cannot be used with diameter distributions).

aerosols(mask: Mask): Total volume of aerosols expired per volume of exhaled air (mL/cm^3).

expirations: Tuple[_ExpirationBase, ...]

weights: Tuple[float, ...]

class MultipleVentilation(ventilations: Tuple[_VentilationBase, ...])

Bases: _VentilationBase

Represents a mechanism by which air can be exchanged (replaced/filtered) in a time dependent manner.

Group together different sources of ventilations.

air_exchange(room: Room, time: float) → float | ndarray: Returns the rate at which air is being exchanged in the given room at a given time (in hours).

transition_times(room: Room) → Set[float]

ventilations: Tuple[_VentilationBase, ...]

class Particle(diameter: None | float | ndarray = None)

Bases: object

Represents an aerosol particle.

diameter: None | float | ndarray = None: diameter of the aerosol in microns

fraction_deposited(evaporation_factor: float = 0.3) → float | ndarray: The fraction of particles actually deposited in the respiratory tract (over the total number of particles). It depends on the particle diameter. From W. C. Hinds, New York, Wiley, 1999 (pp. 233 – 259). evaporation_factor represents the factor applied to the diameter, due to instantaneous evaporation of the particle in the air.

settling_velocity(evaporation_factor: float = 0.3) → float | ndarray: Settling velocity (i.e. speed of deposition on the floor due to gravity), for aerosols, in m/s. Diameter-dependent expression from https://doi.org/10.1101/2021.10.14.21264988 When an aerosol-diameter is not given, returns the default value of 1.88e-4 m/s (corresponds to diameter of 2.5 microns, i.e. geometric average of the breathing expiration distribution, taking evaporation into account, see https://doi.org/10.1101/2021.10.14.21264988) evaporation_factor represents the factor applied to the diameter, due to instantaneous evaporation of the particle in the air.

class PeriodicInterval(period: float, duration: float, start: float = 0.0)

Bases: Interval

boundaries() → Tuple[Tuple[Time_t, Time_t], ...] | Tuple

duration: float: How long does the interval occur for (minutes). A value greater than period signifies the event is permanently occurring, a value of 0 signifies that the event never happens.

period: float: How often does the interval occur (minutes).

start: float = 0.0: Time at which the first person (infected or exposed) arrives at the enclosed space.

class PiecewiseConstant(transition_times: Tuple[float, ...], values: Tuple[float | numpy.ndarray, ...])

Bases: object

interval() → Interval

refine(refine_factor=10) → PiecewiseConstant

transition_times: Tuple[float, ...]: transition times at which the function changes value (hours).

value(time) → float | ndarray

values: Tuple[float | ndarray, ...]: values of the function between transitions

class Population(number: int | IntPiecewiseConstant, presence: None | Interval, activity: Activity, mask: Mask, host_immunity: float)

Bases: SimplePopulation

Represents a group of people all with exactly the same behaviour and situation, considering the usage of mask and a certain host immunity.

host_immunity: float

mask: Mask: The kind of mask being worn by the people.

class Room(volume: float | numpy.ndarray, inside_temp: caimira.models.PiecewiseConstant = PiecewiseConstant(transition_times=(0, 24), values=(293,)), humidity: float | numpy.ndarray = 0.5)

Bases: object

humidity: float | ndarray = 0.5: The humidity in the room (from 0 to 1 - e.g. 0.5 is 50% humidity)

inside_temp: PiecewiseConstant = PiecewiseConstant(transition_times=(0, 24), values=(293,)): The temperature inside the room (Kelvin).

volume: float | ndarray: The total volume of the room

class SARSCoV2(viral_load_in_sputum: float | numpy.ndarray, infectious_dose: float | numpy.ndarray, viable_to_RNA_ratio: float | numpy.ndarray, transmissibility_factor: float, infectiousness_days: int = 14)

Bases: Virus

halflife(humidity: float | ndarray, inside_temp: float | ndarray) → float | ndarray: Half-life changes with humidity level. Here is implemented a simple piecewise constant model (for more details see A. Henriques et al, CERN-OPEN-2021-004, DOI: 10.17181/CERN.1GDQ.5Y75)

infectiousness_days: int = 14: Number of days the infector is contagious

class ShortRangeModel(expiration: _ExpirationBase, activity: Activity, presence: SpecificInterval, distance: float | ndarray)

Bases: object

Based on the two-stage (jet/puff) expiratory jet model by Jia et al (2022) - https://doi.org/10.1016/j.buildenv.2022.109166

activity: Activity: Activity type

dilution_factor() → float | ndarray: The dilution factor for the respective expiratory activity type.

distance: float | ndarray: Interpersonal distances

expiration: _ExpirationBase: Expiration type

extract_between_bounds(time1: float, time2: float) → None | Tuple[float, float]: Extract the bounds of the interval resulting from the intersection of [time1, time2] and the presence interval. If [time1, time2] has nothing common to the presence interval, we return (0, 0). Raise an error if time1 and time2 are not in ascending order.

presence: SpecificInterval: Short-range expiration and respective presence

short_range_concentration(concentration_model: ConcentrationModel, time: float) → float | ndarray: Virus short-range exposure concentration, as a function of time.

class SimplePopulation(number: int | IntPiecewiseConstant, presence: None | Interval, activity: Activity)

Bases: object

Represents a group of people all with exactly the same behaviour and situation.

activity: Activity: The physical activity being carried out by the people.

number: int | IntPiecewiseConstant: How many in the population.

people_present(time: float)

person_present(time: float)

presence: None | Interval: The times in which the people are in the room.

presence_interval()

class SlidingWindow(active: Interval, outside_temp: PiecewiseConstant, window_height: float | ndarray, opening_length: float | ndarray, number_of_windows: int = 1, min_deltaT: float = 0.1)

Bases: WindowOpening

Sliding window, or side-hung window (with the hinge perpendicular to the horizontal plane).

active: Interval: The interval in which the window is open.

property discharge_coefficient: float | ndarray: Average measured value of discharge coefficient for sliding or side-hung windows.

opening_length: float | ndarray: The length of the opening-gap when the window is open (m).

outside_temp: PiecewiseConstant: The temperature outside of the window (Kelvin).

window_height: float | ndarray: The height of the window (m).

class SpecificInterval(present_times: Tuple[Tuple[Time_t, Time_t], ...] | Tuple)

Bases: Interval

boundaries() → Tuple[Tuple[Time_t, Time_t], ...] | Tuple

present_times: Tuple[Tuple[Time_t, Time_t], ...] | Tuple: A sequence of times (start, stop), in hours, that the infected person is present. The flattened list of times must be strictly monotonically increasing.

class Ventilation(active: caimira.models.Interval)

Bases: _VentilationBase

active: Interval: The interval in which the ventilation is active.

transition_times(room: Room) → Set[float]

class Virus(viral_load_in_sputum: float | numpy.ndarray, infectious_dose: float | numpy.ndarray, viable_to_RNA_ratio: float | numpy.ndarray, transmissibility_factor: float, infectiousness_days: int)

Bases: object

decay_constant(humidity: float | ndarray, inside_temp: float | ndarray) → float | ndarray

halflife(humidity: float | ndarray, inside_temp: float | ndarray) → float | ndarray

infectious_dose: float | ndarray: Dose to initiate infection, in RNA copies

infectiousness_days: int: Number of days the infector is contagious

transmissibility_factor: float: Reported increase of transmissibility of a VOC

types: ClassVar[Dict[str, Virus]] = {'SARS_CoV_2': SARSCoV2(viral_load_in_sputum=1000000000.0, infectious_dose=50.0, viable_to_RNA_ratio=0.5, transmissibility_factor=1.0, infectiousness_days=14), 'SARS_CoV_2_ALPHA': SARSCoV2(viral_load_in_sputum=1000000000.0, infectious_dose=50.0, viable_to_RNA_ratio=0.5, transmissibility_factor=0.78, infectiousness_days=14), 'SARS_CoV_2_BETA': SARSCoV2(viral_load_in_sputum=1000000000.0, infectious_dose=50.0, viable_to_RNA_ratio=0.5, transmissibility_factor=0.8, infectiousness_days=14), 'SARS_CoV_2_DELTA': SARSCoV2(viral_load_in_sputum=1000000000.0, infectious_dose=50.0, viable_to_RNA_ratio=0.5, transmissibility_factor=0.51, infectiousness_days=14), 'SARS_CoV_2_GAMMA': SARSCoV2(viral_load_in_sputum=1000000000.0, infectious_dose=50.0, viable_to_RNA_ratio=0.5, transmissibility_factor=0.72, infectiousness_days=14), 'SARS_CoV_2_OMICRON': SARSCoV2(viral_load_in_sputum=1000000000.0, infectious_dose=50.0, viable_to_RNA_ratio=0.5, transmissibility_factor=0.2, infectiousness_days=14)}: Pre-populated examples of Viruses.

viable_to_RNA_ratio: float | ndarray: viable-to-RNA virus ratio as a function of the viral load

viral_load_in_sputum: float | ndarray: RNA copies / mL

class WindowOpening(active: caimira.models.Interval, outside_temp: caimira.models.PiecewiseConstant, window_height: float | numpy.ndarray, opening_length: float | numpy.ndarray, number_of_windows: int = 1, min_deltaT: float = 0.1)

Bases: Ventilation

active: Interval: The interval in which the window is open.

air_exchange(room: Room, time: float) → float | ndarray

Returns the rate at which air is being exchanged in the given room at a given time (in hours).

Note that whilst the time is known inside this function, it may not be used to vary the result unless the specific time used is declared as part of a state change in the interval (e.g. when air_exchange == 0).

property discharge_coefficient: float | ndarray: Discharge coefficient (or cd_b): what portion effective area is used to exchange air (0 <= discharge_coefficient <= 1). To be implemented in subclasses.

min_deltaT: float = 0.1: Minimum difference between inside and outside temperature (K).

number_of_windows: int = 1: The number of windows of the given dimensions.

opening_length: float | ndarray: The length of the opening-gap when the window is open (m).

outside_temp: PiecewiseConstant: The temperature outside of the window (Kelvin).

transition_times(room: Room) → Set[float]

window_height: float | ndarray: The height of the window (m).

caimira.state module

This module is entirely in support of providing a convenient mutable counterpart to frozen dataclasses. Significant effort went into to trying to use traitlets for this purpose, but the need to define class-level attributes proved to be a limitation that meant we could not mutate the state from one subclass to another after the state was instantiated.

This module MUST not import other parts of caimira as this would point at a leaky abstraction.

class DataclassInstanceState(dataclass: ~typing.Type[~caimira.state.Datamodel_T], state_builder=<caimira.state.StateBuilder object>)

Bases: DataclassState[Datamodel_T]

Represents the state of a frozen dataclass. No type checking of the attributes is attempted.

Setting the state can be done with:

setattr(state, attr, value)

Accessing the instance that this state represents can be done with:

state.dcs_instance()

Changing the type to a subclass of the base that this state represents can be done with:

state.dcs_set_instance_type(ASubclassOfBase)

dcs_instance(): Return the instance that this state represents. The instance returned is immutable, so it is advised to call this method each time that you want the instance so that it reflects the most up-to-date state.

dcs_observe(callback: Callable): If any changes are made to the state, call the given callback.

dcs_set_instance_type(instance_dataclass: Type[Any])

Update the current instance of the state to this type.

Note: This currently wipes all downstream observers.

dcs_state_transaction(): For the lifetime of this context manager, do not fire observer notifications. If any notifications would have been fired during the lifetime of this context manager, then an event will be fired once exiting the context.

dcs_update_from(data: Any): Update the state based on the values of the given dataclass instance.

class DataclassState(state_builder=<caimira.state.StateBuilder object>)

Bases: Generic[Datamodel_T]

dcs_instance() → None | Datamodel_T: Return the instance that this state represents. The instance returned is immutable, so it is advised to call this method each time that you want the instance so that it reflects the most up-to-date state.

dcs_observe(callback: Callable): If any changes are made to the state, call the given callback.

dcs_set_instance_type(instance_dataclass: Type[Datamodel_T])

Update the current instance of the state to this type.

Note: This currently wipes all downstream observers.

dcs_state_transaction(): For the lifetime of this context manager, do not fire observer notifications. If any notifications would have been fired during the lifetime of this context manager, then an event will be fired once exiting the context.

dcs_update_from(data: Datamodel_T): Update the state based on the values of the given dataclass instance.

class DataclassStateNamed(states: Dict[str, DataclassState[Datamodel_T]], base_type: str, **kwargs)

Bases: DataclassState[Datamodel_T]

A collection of instances of the given type, switchable by name, but each instance is still mutable.

dcs_instance(): Return the instance that this state represents. The instance returned is immutable, so it is advised to call this method each time that you want the instance so that it reflects the most up-to-date state.

dcs_observe(callback: Callable): If any changes are made to the state, call the given callback.

dcs_select(name: str)

dcs_set_instance_type(instance_dataclass: Type[Datamodel_T])

Update the current instance of the state to this type.

Note: This currently wipes all downstream observers.

dcs_state_transaction(): For the lifetime of this context manager, do not fire observer notifications. If any notifications would have been fired during the lifetime of this context manager, then an event will be fired once exiting the context.

dcs_update_from(data: Datamodel_T): Update the state based on the values of the given dataclass instance.

class DataclassStatePredefined(dataclass: Type[Datamodel_T], choices: Dict[str, Datamodel_T], **kwargs)

Bases: DataclassInstanceState[Datamodel_T]

Only a pre-defined selection of states for the given type are allowed. Selected by name (the keys in the dictionary).

You can change the chosen state with:

state.dcs_select(name)

dcs_instance(): Return the instance that this state represents. The instance returned is immutable, so it is advised to call this method each time that you want the instance so that it reflects the most up-to-date state.

dcs_select(name: str)

class StateBuilder

Bases: object

build_generic(type_to_build: Type) → DataclassInstanceState

resolve_builder(field: Field)

visit(field: Field)

caimira.utils module

method_cache(fn)

A decorator for instance based caching.

Unlike lru_cache / memoization, this allows us to not have to have the instance itself be hashable - only the arguments must be so.

The cache is stored as a dictionary in a private attribute on the instance with the name _cache_{func_name}.

Module contents

Documentation for the CAiMIRA package