Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@ To release a new version (e.g. from `1.0.0` -> `2.0.0`):

## [Unreleased]

* Add `lognormal_dist_from_mean_std` helper function.
* Add `lognormal_dist_from_mean_std` and `lognormal_dist_from_ci` helper
functions.

## [1.2.0] - 2025-09-04

Expand Down
61 changes: 58 additions & 3 deletions meridian/model/prior_distribution.py
Original file line number Diff line number Diff line change
Expand Up @@ -1169,9 +1169,10 @@ def lognormal_dist_from_mean_std(
standard deviation.

Args:
mean: A positive float or array-like object defining the distribution mean.
std: A non-negative float or array-like object defining the distribution
standard deviation.
mean: A float or array-like object defining the distribution mean. Must be
positive.
std: A float or array-like object defining the distribution standard
deviation. Must be non-negative.

Returns:
A `backend.tfd.LogNormal` object with the input mean and standard deviation.
Expand All @@ -1186,6 +1187,60 @@ def lognormal_dist_from_mean_std(
return backend.tfd.LogNormal(mu, sigma)


def lognormal_dist_from_range(
low: float | Sequence[float],
high: float | Sequence[float],
mass_percent: float | Sequence[float] = 0.95,
) -> backend.tfd.LogNormal:
"""Define a LogNormal distribution from a specified range.

This function parameterizes lognormal distributions by the bounds of a range,
so that the specificed probability mass falls within the bounds defined by
`low` and `high`. The probability mass is symmetric about the median. For
example, to define a lognormal distribution with a 95% probability mass of
(1, 10), use:

```python
lognormal = lognormal_dist_from_range(1.0, 10.0, mass_percent=0.95)
```

Args:
low: Float or array-like denoting the lower bound of the range. Values must
be non-negative.
high: Float or array-like denoting the upper bound of range. Values must be
non-negative.
mass_percent: Float or array-like denoting the probability mass. Values must
be between 0 and 1 (exlusive). Default: 0.95.

Returns:
A `backend.tfd.LogNormal` object with the input percentage mass falling
within the given range.
"""
low = np.asarray(low)
high = np.asarray(high)
mass_percent = np.asarray(mass_percent)

if not ((0.0 < low).all() and (low < high).all()): # pytype: disable=attribute-error
raise ValueError("'low' and 'high' values must be non-negative and satisfy "
"high > low.")

if not ((0.0 < mass_percent).all() and (mass_percent < 1.0).all()): # pytype: disable=attribute-error
raise ValueError(
"'mass_percent' values must be between 0 and 1, exclusive."
)

normal = backend.tfd.Normal(0, 1)
mass_lower = 0.5 - (mass_percent / 2)
mass_upper = 0.5 + (mass_percent / 2)

sigma = np.log(high / low) / (
normal.quantile(mass_upper) - normal.quantile(mass_lower)
)
mu = np.log(high) - normal.quantile(mass_upper) * sigma

return backend.tfd.LogNormal(mu, sigma)


def _convert_to_deterministic_0_distribution(
distribution: backend.tfd.Distribution,
) -> backend.tfd.Distribution:
Expand Down
152 changes: 151 additions & 1 deletion meridian/model/prior_distribution_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from collections.abc import Callable, MutableMapping
from collections.abc import Callable, Iterable, MutableMapping
import copy
from typing import Any
import warnings
Expand Down Expand Up @@ -1974,5 +1974,155 @@ def test_correct_mean_std_array(self, mean, std, input_type):
np.testing.assert_allclose(dist.stddev(), expected_std, rtol=1e-5)


class TestLognormalDistFromRange(parameterized.TestCase):

@parameterized.product(
(
dict(low=0.1, high=1.0),
dict(low=1.0, high=10.0),
),
mass_percent=(0.1, 0.8, 0.9, 0.95),
input_type=(float, np.float32, backend.to_tensor),
)
def test_correct_quantile_scalar(
self,
low,
high,
mass_percent,
input_type,
):
low = input_type(low)
high = input_type(high)

dist = prior_distribution.lognormal_dist_from_range(low, high, mass_percent)

mass_lower = 0.5 - mass_percent / 2
mass_upper = 0.5 + mass_percent / 2

np.testing.assert_allclose(dist.quantile(mass_lower), low, rtol=1e-5)
np.testing.assert_allclose(dist.quantile(mass_upper), high, rtol=1e-5)

@parameterized.product(
(
dict(low=0.1, high=(1.0,)),
dict(low=0.1, high=(1.0, 2.0, 3.0)),
dict(low=(0.1,), high=(1.0,)),
dict(low=(0.1,), high=(1.0, 2.0, 3.0)),
dict(low=(0.1,), high=1.0),
dict(low=(0.1, 0.2, 0.3), high=1.0),
dict(low=(0.1, 0.2, 0.3), high=(1.0,)),
dict(low=(0.1, 0.2, 0.3), high=(1.0, 2.0, 3.0)),
),
mass_percent=(0.1, 0.8, 0.9, (0.95,), (0.8, 0.9, 0.95)),
input_type=(tuple, list, np.array, backend.to_tensor)
)
def test_correct_quantile_array(
self,
low,
high,
mass_percent,
input_type,
):

low = input_type(low) if isinstance(low, Iterable) else low
high = input_type(high) if isinstance(high, Iterable) else high
mass_percent = (
input_type(mass_percent)
if isinstance(mass_percent, Iterable)
else mass_percent
)

expected_len = 1

for var in (low, high, mass_percent):
try:
expected_len = max(expected_len, len(var))
except TypeError:
pass

expected_low = np.broadcast_to(low, expected_len)
expected_high = np.broadcast_to(high, expected_len)

dist = prior_distribution.lognormal_dist_from_range(
low, high, mass_percent=mass_percent
)

mass_lower = 0.5 - (np.asarray(mass_percent) / 2)
mass_upper = 0.5 + (np.asarray(mass_percent) / 2)

np.testing.assert_allclose(
dist.quantile(mass_lower), expected_low, rtol=1e-5
)
np.testing.assert_allclose(
dist.quantile(mass_upper), expected_high, rtol=1e-5
)

@parameterized.named_parameters(
dict(
testcase_name='low_negative',
low=-0.1,
high=0.5,
),
dict(
testcase_name='low_zero',
low=0.0,
high=0.5,
),
dict(
testcase_name='high_less_than_low',
low=1.0,
high=0.5,
),
dict(
testcase_name='array_low_negative',
low=(-0.1, 0.1),
high=0.5,
),
dict(
testcase_name='array_low_zero',
low=(0.0, 0.1),
high=0.5,
),
dict(
testcase_name='array_high_less_than_low',
low=(0.5, 1.0),
high=(1.0, 0.75),
),
)
def test_out_of_bounds_low_high_raises_error(self, low, high):
with self.assertRaisesWithLiteralMatch(
ValueError,
"'low' and 'high' values must be non-negative and satisfy high > low."
):
_ = prior_distribution.lognormal_dist_from_range(low, high)

@parameterized.named_parameters(
dict(
testcase_name='negative',
mass_percent=-0.1,
),
dict(
testcase_name='greater_than_1',
mass_percent=1.1,
),
dict(
testcase_name='array_negative',
mass_percent=(-0.1, 0.5),
),
dict(
testcase_name='array_greater_than_1',
mass_percent=(1.1, 0.5),
),
)
def test_out_of_bounds_mass_percent_raises_error(self, mass_percent):
with self.assertRaisesWithLiteralMatch(
ValueError,
"'mass_percent' values must be between 0 and 1, exclusive."
):
_ = prior_distribution.lognormal_dist_from_range(
1.0, 2.0, mass_percent=mass_percent
)


if __name__ == '__main__':
absltest.main()