Coverage for src/cell_abm_pipeline/tasks/calculate_data_bins.py: 0%
13 statements
« prev ^ index » next coverage.py v7.1.0, created at 2024-06-05 19:14 +0000
« prev ^ index » next coverage.py v7.1.0, created at 2024-06-05 19:14 +0000
1import numpy as np
2from prefect import task
5@task
6def calculate_data_bins(
7 data: np.ndarray, bounds: tuple[float, float], bandwidth: float
8) -> list[dict]:
9 if len(data) == 0:
10 return []
12 total = len(data) * bandwidth
13 num_bins = int((bounds[1] - bounds[0]) / bandwidth)
14 lower = bounds[0] - 3 * bandwidth / 2
15 upper = bounds[1] + 3 * bandwidth / 2
17 bins = np.linspace(lower, upper, num_bins + 4).tolist()
18 counts, _ = np.histogram(data, bins)
20 return [
21 {"n": count, "x": x0, "y": count / total, "m": (x0 + x1) / 2}
22 for count, x0, x1 in zip(counts.tolist(), bins[:-1], bins[1:])
23 ]