Coverage for src/cell_abm_pipeline/tasks/bin_to_hex.py: 0%

1from math import sqrt 

2from typing import Optional, Union 

3 

4import numpy as np 

5from prefect import task 

6 

7 

8@task 

9def bin_to_hex( 

10 x_coordinates: np.ndarray, 

11 y_coordinates: np.ndarray, 

12 values: np.ndarray, 

13 scale: float = 1, 

14 limits: Optional[tuple[float, float, float, float]] = None, 

15) -> dict[tuple[float, float], list[Union[int, float]]]: 

16 bins: dict[tuple[float, float], list[Union[int, float]]] = {} 

17 

18 if limits is not None: 

19 x_min, x_max, y_min, y_max = limits 

20 x = (x_coordinates - x_min) / (x_max - x_min) 

21 y = (y_coordinates - y_min) / (y_max - y_min) 

22 else: 

23 x = x_coordinates 

24 y = y_coordinates 

25 

26 for xi, yi, vi in zip(x, y, values): 

27 sxi = xi / scale 

28 syi = yi / scale 

29 

30 cx1 = scale * round(sxi / sqrt(3)) * sqrt(3) 

31 cy1 = scale * round(syi) 

32 dist1 = sqrt((xi - cx1) ** 2 + (yi - cy1) ** 2) 

33 

34 cx2 = scale * (round(sxi / sqrt(3) - 0.4999) + 0.5) * sqrt(3) 

35 cy2 = scale * (round(syi - 0.49999) + 0.5) 

36 dist2 = sqrt((xi - cx2) ** 2 + (yi - cy2) ** 2) 

37 

38 if dist1 < dist2: 

39 cx, cy = (cx1, cy1) 

40 else: 

41 cx, cy = (cx2, cy2) 

42 

43 if limits is not None: 

44 cx = (cx * (x_max - x_min)) + x_min 

45 cy = (cy * (y_max - y_min)) + y_min 

46 

47 if (cx, cy) not in bins: 

48 bins[(cx, cy)] = [] 

49 

50 bins[(cx, cy)].append(vi) 

51 

52 return bins