| 22 | |
| 23 | |
| 24 | class Loss(nn.Module, ABC, Generic[T]): |
| 25 | cfg: T |
| 26 | |
| 27 | def __init__(self, cfg: T) -> None: |
| 28 | super().__init__() |
| 29 | self.cfg = cfg |
| 30 | |
| 31 | def forward( |
| 32 | self, |
| 33 | batch: Batch, |
| 34 | flows: Flows, |
| 35 | tracks: list[Tracks] | None, |
| 36 | model_output: ModelOutput, |
| 37 | global_step: int, |
| 38 | ) -> Float[Tensor, ""]: |
| 39 | # Before the loss is enabled, don't compute the loss. |
| 40 | if global_step < self.cfg.enable_after: |
| 41 | return torch.tensor(0, dtype=torch.float32, device=batch.videos.device) |
| 42 | |
| 43 | # Multiply the computed loss value by the weight. |
| 44 | loss = self.compute_unweighted_loss( |
| 45 | batch, flows, tracks, model_output, global_step |
| 46 | ) |
| 47 | return self.cfg.weight * loss |
| 48 | |
| 49 | @abstractmethod |
| 50 | def compute_unweighted_loss( |
| 51 | self, |
| 52 | batch: Batch, |
| 53 | flows: Flows, |
| 54 | tracks: list[Tracks] | None, |
| 55 | model_output: ModelOutput, |
| 56 | global_step: int, |
| 57 | ) -> Float[Tensor, ""]: |
| 58 | pass |
nothing calls this directly
no outgoing calls
no test coverage detected