Skip to content

particula.particles.distribution_strategies.particle_resolved_speciated_mass

particle_resolved_speciated_mass

Particle resolved speciated mass strategy.

ParticleResolvedSpeciatedMass

Bases: DistributionStrategy

Strategy for particle-resolved masses with multiple species.

Allows each particle to have separate masses for each species, with individualized densities. This strategy provides a more detailed approach when each particle's composition must be modeled explicitly.

Methods: - get_name : Return the type of the distribution strategy. - get_species_mass : Calculate the mass per species. - get_mass : Calculate the mass of the particles or bin. - get_total_mass : Calculate the total mass of particles. - get_radius : Calculate the radius of particles. - add_mass : Add mass to the particle distribution. - add_concentration : Add concentration to the distribution. - collide_pairs : Perform collision logic on specified particle pairs.

add_concentration 

add_concentration(distribution: NDArray[float64], concentration: NDArray[float64], added_distribution: NDArray[float64], added_concentration: NDArray[float64]) -> tuple[NDArray[np.float64], NDArray[np.float64]]

Add new particles to the distribution.

Returns:

  • tuple[NDArray[float64], NDArray[float64]] –

    Updated distribution and concentration arrays.

Source code in particula/particles/distribution_strategies/particle_resolved_speciated_mass.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def add_concentration(
    self,
    distribution: NDArray[np.float64],
    concentration: NDArray[np.float64],
    added_distribution: NDArray[np.float64],
    added_concentration: NDArray[np.float64],
) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Add new particles to the distribution.

    Returns:
        Updated distribution and concentration arrays.
    """
    rescaled = False
    if np.all(added_concentration == 1):
        rescaled = True
    max_concentration = np.max(concentration)
    if np.allclose(
        added_concentration, max_concentration, atol=1e-2
    ) or np.all(concentration == 0):
        if max_concentration > 0:
            added_concentration = added_concentration / max_concentration
        rescaled = True
    if not rescaled:
        message = (
            "When adding concentration to ParticleResolvedSpeciatedMass, "
            "added concentration should be all ones or all the same."
        )
        logger.error(message)
        raise ValueError(message)

    concentration = np.divide(
        concentration,
        concentration,
        out=np.zeros_like(concentration),
        where=concentration != 0,
    )

    empty_bins = np.flatnonzero(concentration == 0)
    empty_bins_count = len(empty_bins)
    added_bins_count = len(added_concentration)
    if empty_bins_count >= added_bins_count:
        distribution[empty_bins] = added_distribution
        concentration[empty_bins] = added_concentration
        return distribution, concentration
    if empty_bins_count > 0:
        distribution[empty_bins] = added_distribution[:empty_bins_count]
        concentration[empty_bins] = added_concentration[:empty_bins_count]
    distribution = np.concatenate(
        (distribution, added_distribution[empty_bins_count:]), axis=0
    )
    concentration = np.concatenate(
        (concentration, added_concentration[empty_bins_count:]), axis=0
    )
    return distribution, concentration

add_mass 

add_mass(distribution: NDArray[float64], concentration: NDArray[float64], density: NDArray[float64], added_mass: NDArray[float64]) -> tuple[NDArray[np.float64], NDArray[np.float64]]

Add mass to individual particles in the distribution.

Returns:

  • tuple[NDArray[float64], NDArray[float64]] –

    Updated distribution and concentration arrays.

Source code in particula/particles/distribution_strategies/particle_resolved_speciated_mass.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def add_mass(  # pylint: disable=R0801
    self,
    distribution: NDArray[np.float64],
    concentration: NDArray[np.float64],
    density: NDArray[np.float64],
    added_mass: NDArray[np.float64],
) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
    """Add mass to individual particles in the distribution.

    Returns:
        Updated distribution and concentration arrays.
    """
    if distribution.ndim == 2:
        concentration_expand = concentration[:, np.newaxis]
    else:
        concentration_expand = concentration
    new_mass = np.divide(
        np.maximum(distribution * concentration_expand + added_mass, 0),
        concentration_expand,
        out=np.zeros_like(distribution),
        where=concentration_expand != 0,
    )
    if new_mass.ndim == 1:
        new_mass_sum = np.sum(new_mass)
    else:
        new_mass_sum = np.sum(new_mass, axis=1)
    concentration = np.where(new_mass_sum > 0, concentration, 0)
    return new_mass, concentration

collide_pairs 

collide_pairs(distribution: NDArray[float64], concentration: NDArray[float64], density: NDArray[float64], indices: NDArray[int64], charge: Optional[NDArray[float64]] = None) -> tuple[NDArray[np.float64], NDArray[np.float64], Optional[NDArray[np.float64]]]

Collide specified particle pairs by merging mass and charge.

Performs coagulation between particle pairs for particle-resolved simulations. The smaller particle's mass is added to the larger particle, and the smaller particle's concentration is set to zero. If a charge array is provided, charges are conserved by summing the charges of the colliding pair.

The charge handling is optimized: charges are only processed when the charge array is provided as a numpy array AND at least one of the colliding particles has a non-zero charge.

Parameters:

  • - distribution –

    The mass distribution array. Shape is (N,) for single species or (N, M) for M species per particle.

  • - concentration –

    The concentration array of shape (N,).

  • - density –

    The density array of shape (M,) for species densities.

  • - indices –

    Collision pair indices array of shape (K, 2) where each row is [small_index, large_index].

  • - charge –

    Optional charge array of shape (N,). If provided and contains non-zero values in colliding pairs, charges will be summed during collisions. If None, charge handling is skipped.

Returns:

  • tuple[NDArray[float64], NDArray[float64], Optional[NDArray[float64]]] –

    A tuple containing: - Updated distribution array with merged masses. - Updated concentration array with zeroed small particles. - Updated charge array (None if input was None).

Source code in particula/particles/distribution_strategies/particle_resolved_speciated_mass.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def collide_pairs(  # pylint: disable=too-many-positional-arguments
    self,
    distribution: NDArray[np.float64],
    concentration: NDArray[np.float64],
    density: NDArray[np.float64],
    indices: NDArray[np.int64],
    charge: Optional[NDArray[np.float64]] = None,
) -> tuple[
    NDArray[np.float64], NDArray[np.float64], Optional[NDArray[np.float64]]
]:
    """Collide specified particle pairs by merging mass and charge.

    Performs coagulation between particle pairs for particle-resolved
    simulations. The smaller particle's mass is added to the larger
    particle, and the smaller particle's concentration is set to zero.
    If a charge array is provided, charges are conserved by summing the
    charges of the colliding pair.

    The charge handling is optimized: charges are only processed when the
    charge array is provided as a numpy array AND at least one of the
    colliding particles has a non-zero charge.

    Arguments:
        - distribution : The mass distribution array. Shape is (N,) for
            single species or (N, M) for M species per particle.
        - concentration : The concentration array of shape (N,).
        - density : The density array of shape (M,) for species densities.
        - indices : Collision pair indices array of shape (K, 2) where
            each row is [small_index, large_index].
        - charge : Optional charge array of shape (N,). If provided and
            contains non-zero values in colliding pairs, charges will be
            summed during collisions. If None, charge handling is skipped.

    Returns:
        A tuple containing:
            - Updated distribution array with merged masses.
            - Updated concentration array with zeroed small particles.
            - Updated charge array (None if input was None).
    """
    small_index = indices[:, 0]
    large_index = indices[:, 1]

    # Handle mass (existing logic)
    if distribution.ndim == 1:
        distribution[large_index] += distribution[small_index]
        distribution[small_index] = 0
    else:
        distribution[large_index, :] += distribution[small_index, :]
        distribution[small_index, :] = 0
    concentration[small_index] = 0

    # Handle charge if present as numpy array and non-zero
    # charge can be None or array - only process if array
    if charge is not None and isinstance(charge, np.ndarray):
        # Check only colliding pairs for non-zero charges (performance opt)
        if np.any(charge[small_index] != 0) or np.any(
            charge[large_index] != 0
        ):
            charge[large_index] += charge[small_index]
            charge[small_index] = 0

    return distribution, concentration, charge

get_radius 

get_radius(distribution: NDArray[float64], density: NDArray[float64]) -> NDArray[np.float64]

Calculate particle radius from multi-species mass and density.

Returns:

  • NDArray[float64] –

    Particle radius in meters for each particle.

Source code in particula/particles/distribution_strategies/particle_resolved_speciated_mass.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_radius(
    self, distribution: NDArray[np.float64], density: NDArray[np.float64]
) -> NDArray[np.float64]:
    """Calculate particle radius from multi-species mass and density.

    Returns:
        Particle radius in meters for each particle.
    """
    if distribution.ndim == 1:
        volumes = distribution / density
    else:
        volumes = np.sum(distribution / density, axis=1)
    return (3 * volumes / (4 * np.pi)) ** (1 / 3)

get_species_mass 

get_species_mass(distribution: NDArray[float64], density: NDArray[float64]) -> NDArray[np.float64]

Calculate the mass per species for each particle.

Returns:

  • NDArray[float64] –

    Mass per species array for each particle.

Source code in particula/particles/distribution_strategies/particle_resolved_speciated_mass.py 
32
33
34
35
36
37
38
39
40
def get_species_mass(
    self, distribution: NDArray[np.float64], density: NDArray[np.float64]
) -> NDArray[np.float64]:
    """Calculate the mass per species for each particle.

    Returns:
        Mass per species array for each particle.
    """
    return distribution