Coverage for cafl4ds/filters/base.py: 100%

1"""The selection-filter interface (the ``A`` factor / independent variable). 

2 

3The filter is the knob the whole study turns. The ``A`` factor has **two composable roles**, 

4because the cheap Phase-0 knobs are not all the same kind of thing: 

5 

6* An **admission** :class:`Filter` *narrows* an incoming batch to a subset — B-floor 

7 (:class:`~cafl4ds.filters.accept_all.AcceptAll`, accept everything), SemDeDup 

8 (near-duplicate removal), the loss-gate (keep high-loss frames). Admission filters are pure 

9 batch reducers, so they **compose freely as a list** (apply one after another). 

10* A **replay** :class:`ReplayBuffer` decides what is *stored and replayed* — it may emit past 

11 items, so it can *grow* the training batch, and it carries state across steps. Reservoir 

12 sampling (:class:`~cafl4ds.filters.reservoir.ReservoirReplay`) is the Phase-0 instance. A 

13 buffer is a single-choice role and always runs **last** (after admission). 

14 

15:class:`~cafl4ds.filters.composite.CompositeSelector` combines an admission list with an 

16optional buffer behind the single :class:`Filter` interface the loop calls, which is what lets 

17the plan's baselines be selected by config — B-floor, B1 (reservoir), dedup, loss-gate, and 

18B1.5 (dedup ∘ reservoir). Later phases add novelty (F-a), coverage (F-b), and the 

19health-steerable filter (F-c) behind these same interfaces. 

20 

21The :class:`FilterContext` deliberately carries the live model and step so that 

22model-in-the-loop filters (the *fast edge*) and, later, health state (the *slow edge*) have 

23what they need without changing the loop's call site. 

24""" 

25 

26from __future__ import annotations 

27 

28from abc import ABC, abstractmethod 

29from dataclasses import dataclass 

30 

31import torch 

32 

33from cafl4ds.data.streams import StreamBatch 

34from cafl4ds.ssl.base import SSLMethod 

35 

36 

37@dataclass 

38class FilterContext: 

39 """Everything a filter may consult when selecting from a batch.""" 

40 

41 method: SSLMethod 

42 """The live SSL method — its encoder scores informativeness against the *current* model 

43 (the fast edge). B-floor ignores it.""" 

44 step: int 

45 """The current global stream step.""" 

46 

47 

48class Filter(ABC): 

49 """Selects which images from an incoming stream batch enter the SSL update.""" 

50 

51 @abstractmethod 

52 def select(self, batch: StreamBatch, ctx: FilterContext) -> torch.Tensor: 

53 """Return the images accepted from ``batch`` for this update. 

54 

55 Args: 

56 batch: The incoming label-free stream batch. 

57 ctx: The live filter context (model, step). 

58 

59 Returns: 

60 The accepted images ``[K, C, H, W]`` with ``0 <= K <= B`` (possibly the whole 

61 batch, possibly empty). 

62 """ 

63 

64 

65class ReplayBuffer(ABC): 

66 """Stores incoming images and emits the batch the SSL update trains on (the replay role). 

67 

68 Unlike an admission :class:`Filter`, a buffer holds state across steps and may return 

69 *more* items than it was given (the incoming batch mixed with replayed past items), so it 

70 is a distinct role that runs after admission. Phase 0 has one instance, 

71 :class:`~cafl4ds.filters.reservoir.ReservoirReplay`. 

72 """ 

73 

74 @abstractmethod 

75 def observe(self, images: torch.Tensor, ctx: FilterContext) -> torch.Tensor: 

76 """Ingest the admitted images and return the batch to train on this step. 

77 

78 Args: 

79 images: The admitted images ``[K, C, H, W]`` (post-admission) for this step. 

80 ctx: The live filter context (model, step). 

81 

82 Returns: 

83 The training batch ``[M, C, H, W]`` — typically the incoming images plus a replay 

84 sample drawn from the buffer's stored history (``M >= K``). 

85 """