Coverage for src/evutils/transforms/functional/_time.py: 93%

1"""Temporal functional transforms (skew, jitter, normalize).""" 

2import numpy as np 

3from evutils.jit import lazy_njit 

4 

5from ._common import apply_kernel 

6 

7@lazy_njit 

8def _time_skew_jit(t, x, y, p, coefficient: float, offset: float): 

9 """Apply the affine timestamp map ``t' = t * coefficient + offset``.""" 

10 new_t = (t.astype(np.float64) * coefficient + offset).astype(np.int64) 

11 return new_t, x, y, p 

12 

13def time_skew(events, coefficient, offset=0.0): 

14 """Rescale (and shift) all timestamps by a linear map. 

15 

16 Parameters 

17 ---------- 

18 events : np.ndarray or EventArray 

19 Events to skew. 

20 coefficient : float 

21 Multiplier applied to every timestamp (e.g. ``2.0`` doubles all gaps). 

22 offset : float, optional 

23 Added after multiplication. Default ``0.0``. 

24 

25 Returns 

26 ------- 

27 np.ndarray or EventArray 

28 Events with rewritten timestamps, in their original container type. 

29 """ 

30 return apply_kernel(events, _time_skew_jit, float(coefficient), float(offset)) 

31 

32@lazy_njit 

33def _time_jitter_jit(t, x, y, p, std: float, clip_negative: bool, 

34 sort_timestamps: bool): 

35 """Add Gaussian noise to timestamps, optionally clipping and re-sorting.""" 

36 shifts = np.random.normal(0.0, std, len(t)) 

37 new_t = (t.astype(np.float64) + shifts).astype(np.int64) 

38 

39 if clip_negative: 

40 keep = new_t >= 0 

41 new_t, x, y, p = new_t[keep], x[keep], y[keep], p[keep] 

42 

43 if sort_timestamps: 

44 order = np.argsort(new_t) 

45 new_t, x, y, p = new_t[order], x[order], y[order], p[order] 

46 

47 return new_t, x, y, p 

48 

49def time_jitter(events, std=1.0, clip_negative=True, sort_timestamps=False): 

50 """Add Gaussian noise to each timestamp. 

51 

52 Parameters 

53 ---------- 

54 events : np.ndarray or EventArray 

55 Events to jitter. 

56 std : float, optional 

57 Standard deviation of the timestamp noise. Default ``1.0``. 

58 clip_negative : bool, optional 

59 Drop events whose jittered timestamp is negative. Default ``True``. 

60 sort_timestamps : bool, optional 

61 Re-sort events by timestamp after jittering. Default ``False``. 

62 

63 Returns 

64 ------- 

65 np.ndarray or EventArray 

66 Jittered events, in their original container type. 

67 """ 

68 return apply_kernel(events, _time_jitter_jit, float(std), 

69 bool(clip_negative), bool(sort_timestamps)) 

70 

71@lazy_njit 

72def _normalize_ts_jit(t, x, y, p, start_ts: int): 

73 """Shift every timestamp so the minimum lands at ``start_ts``.""" 

74 new_t = t - (t.min() - start_ts) 

75 return new_t, x, y, p 

76 

77def normalize_ts(events, start_ts=0): 

78 """Shift timestamps so the earliest event lands at ``start_ts``. 

79 

80 Pure and stateless: each call normalizes the batch it is handed on its own. 

81 For chunk-by-chunk streams use ``EventReader(normalize_ts=True)``, which 

82 latches the offset from the first chunk and applies it across the whole 

83 stream (so the timeline stays continuous instead of resetting per chunk). 

84 

85 Parameters 

86 ---------- 

87 events : np.ndarray or EventArray 

88 Events to normalize. 

89 start_ts : int, optional 

90 Timestamp assigned to the earliest event. Default ``0``. 

91 

92 Returns 

93 ------- 

94 np.ndarray or EventArray 

95 Events with shifted timestamps, in their original container type. The 

96 input is not modified. 

97 

98 Examples 

99 -------- 

100 >>> import numpy as np 

101 >>> from evutils.transforms.functional import normalize_ts 

102 >>> events = np.array( 

103 ... [(0, 0, 100, 1), (1, 1, 200, 1), (2, 2, 300, 0)], 

104 ... dtype=[('x', 'u2'), ('y', 'u2'), ('t', 'i8'), ('p', 'i1')] 

105 ... ) 

106 >>> normalize_ts(events)['t'] 

107 array([ 0, 100, 200]) 

108 """ 

109 return apply_kernel(events, _normalize_ts_jit, int(start_ts))