|
15 | 15 | EMPTY_RESULT: dict[str, np.ndarray] = {} |
16 | 16 |
|
17 | 17 |
|
| 18 | +class FrameSkipPolicy: |
| 19 | + """ |
| 20 | + Decides whether to skip (drop) a frame based on a cyclic counter. |
| 21 | +
|
| 22 | + Skip pattern with interval N (N >= 1): |
| 23 | + Process frames 1..N-1, drop frame N, repeat. |
| 24 | + Example (N=3): process, process, DROP, process, process, DROP, ... |
| 25 | +
|
| 26 | + If interval is 0, no frames are ever skipped. |
| 27 | +
|
| 28 | + Args: |
| 29 | + interval: Total cycle length (process and skip). 0 disables skipping. |
| 30 | + skip_amount: Number of consecutive frames to skip per cycle. Must be < interval. |
| 31 | +
|
| 32 | + Raises: |
| 33 | + ValueError: If frame_skip_interval is negative. |
| 34 | + """ |
| 35 | + |
| 36 | + def __init__(self, interval: int = 3, skip_amount: int = 1) -> None: |
| 37 | + if interval < 0 or interval == 1: |
| 38 | + raise ValueError(f"frame_skip_interval must be > 1 or 0 for no skipping, got {interval}") |
| 39 | + if interval > 0 and (skip_amount < 0 or skip_amount >= interval): |
| 40 | + raise ValueError(f"skip_amount must be >= 0 and < interval, got {skip_amount} and {interval}") |
| 41 | + self._interval = interval |
| 42 | + self._skip_amount = skip_amount |
| 43 | + self._counter = 0 |
| 44 | + |
| 45 | + @property |
| 46 | + def interval(self) -> int: |
| 47 | + return self._interval |
| 48 | + |
| 49 | + @property |
| 50 | + def skip_amount(self) -> int: |
| 51 | + return self._skip_amount |
| 52 | + |
| 53 | + def should_skip(self) -> bool: |
| 54 | + """Return True if the current frame should be dropped. Advances the internal counter on every call.""" |
| 55 | + if self._interval == 0 or self._skip_amount == 0: |
| 56 | + return False |
| 57 | + |
| 58 | + position = self._counter % self._interval |
| 59 | + self._counter += 1 |
| 60 | + |
| 61 | + # process the first (interval - skip_count) frames, skip the rest |
| 62 | + process_count = self._interval - self._skip_amount |
| 63 | + return position >= process_count |
| 64 | + |
| 65 | + def reset(self) -> None: |
| 66 | + """Reset the internal counter.""" |
| 67 | + self._counter = 0 |
| 68 | + |
| 69 | + |
18 | 70 | class Processor(PipelineComponent): |
19 | 71 | """ |
20 | 72 | A job component responsible for retrieving raw frames from the inbound broadcaster, |
21 | 73 | sending them to a processor for inference, and broadcasting the processed results to subscribed consumers. |
| 74 | +
|
| 75 | + Supports frame skipping to align model throughput with source frame rate. |
22 | 76 | """ |
23 | 77 |
|
24 | 78 | def __init__( |
25 | | - self, |
26 | | - model_handler: ModelHandler, |
27 | | - batch_size: int = 3, |
28 | | - category_id_to_label_id: dict[int, str] | None = None, |
| 79 | + self, model_handler: ModelHandler, batch_size: int = 1, frame_skip_interval: int = 3, frame_skip_amount: int = 1 |
29 | 80 | ) -> None: |
30 | 81 | super().__init__() |
31 | 82 | self._model_handler = model_handler |
32 | 83 | self._batch_size = batch_size |
33 | | - self._category_id_to_label_id = category_id_to_label_id or {} |
| 84 | + self._skip_policy = FrameSkipPolicy(interval=frame_skip_interval, skip_amount=frame_skip_amount) |
| 85 | + self._initialized = False |
34 | 86 |
|
35 | 87 | def setup( |
36 | | - self, |
37 | | - inbound_broadcaster: FrameBroadcaster[InputData], |
38 | | - outbound_broadcaster: FrameBroadcaster[OutputData], |
| 88 | + self, inbound_broadcaster: FrameBroadcaster[InputData], outbound_broadcaster: FrameBroadcaster[OutputData] |
39 | 89 | ) -> None: |
40 | 90 | self._inbound_broadcaster = inbound_broadcaster |
41 | 91 | self._outbound_broadcaster = outbound_broadcaster |
42 | 92 | self._in_queue: Queue[InputData] = inbound_broadcaster.register(self.__class__.__name__) |
43 | 93 | self._initialized = True |
44 | 94 |
|
45 | | - def run(self) -> None: |
| 95 | + def run(self) -> None: # noqa: C901 |
| 96 | + if not self._initialized: |
| 97 | + raise RuntimeError("Processor must be set up before running") |
46 | 98 | logger.debug("Starting a pipeline runner loop") |
| 99 | + |
47 | 100 | self._model_handler.initialise() |
48 | | - logger.info("Pipeline model handler initialized") |
| 101 | + logger.info( |
| 102 | + "Pipeline model handler initialized, batch size: %d, frame skip interval: %d, skip amount: %d", |
| 103 | + self._batch_size, |
| 104 | + self._skip_policy.interval, |
| 105 | + self._skip_policy.skip_amount, |
| 106 | + ) |
49 | 107 |
|
50 | 108 | while not self._stop_event.is_set(): |
51 | 109 | try: |
52 | 110 | batch_data: list[InputData] = [] |
53 | | - for _ in range(self._batch_size): |
| 111 | + while len(batch_data) < self._batch_size and not self._stop_event.is_set(): |
54 | 112 | try: |
55 | | - input_data = self._in_queue.get(timeout=0.1) |
| 113 | + input_data: InputData = self._in_queue.get(timeout=0.1) |
56 | 114 | if input_data.trace: |
57 | 115 | input_data.trace.record_start("processor") |
58 | | - batch_data.append(input_data) |
59 | | - |
60 | | - if input_data.context.get("requires_manual_control", False): |
61 | | - break |
62 | 116 | except Empty: |
| 117 | + if batch_data: # if we have partial batch data, process what we have |
| 118 | + break |
| 119 | + continue |
| 120 | + |
| 121 | + is_manual = input_data.context.get("requires_manual_control", False) |
| 122 | + |
| 123 | + if not is_manual and self._skip_policy.should_skip(): |
| 124 | + logger.debug("Frame skipped (timestamp=%s)", input_data.timestamp) |
| 125 | + continue |
| 126 | + |
| 127 | + batch_data.append(input_data) |
| 128 | + |
| 129 | + if is_manual: |
63 | 130 | break |
64 | 131 |
|
65 | | - if not batch_data: |
| 132 | + if not batch_data or self._stop_event.is_set(): |
66 | 133 | continue |
67 | 134 |
|
68 | | - batch_results = self._model_handler.predict(batch_data) |
| 135 | + results = self._model_handler.predict(batch_data) |
69 | 136 |
|
70 | 137 | for i, data in enumerate(batch_data): |
71 | | - results: dict[str, np.ndarray] = batch_results[i] if i < len(batch_results) else EMPTY_RESULT |
| 138 | + result = results[i] if i < len(results) else EMPTY_RESULT |
72 | 139 | if data.trace: |
73 | 140 | data.trace.record_end("processor") |
74 | | - output_data = OutputData( |
75 | | - frame=data.frame, |
76 | | - results=[results], |
77 | | - trace=data.trace, |
78 | | - ) |
| 141 | + output_data = OutputData(frame=data.frame, results=[result] if result else [], trace=data.trace) |
79 | 142 | self._outbound_broadcaster.broadcast(output_data) |
80 | 143 |
|
81 | 144 | except Exception as e: |
|
0 commit comments