File size: 6,372 Bytes
0122a25
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
"""This module contains utilities for callbacks."""

from __future__ import annotations

import os
from typing import Any

import lightning.pytorch as pl

from mapdet3d.common.distributed import (
    all_gather_object_cpu,
    broadcast,
    rank_zero_only,
    synchronize,
)
from mapdet3d.common.logging import rank_zero_info
from mapdet3d.common.typing import ArgsType, MetricLogs
from mapdet3d.data.typing import DictData
from mapdet3d.eval.base import Evaluator

from .base import Callback


class EvaluatorCallback(Callback):
    """Callback for model evaluation."""

    def __init__(
        self,
        *args: ArgsType,
        evaluator: Evaluator,
        metrics_to_eval: list[str] | None = None,
        save_predictions: bool = False,
        save_prefix: None | str = None,
        output_dir: str | None = None,
        **kwargs: ArgsType,
    ) -> None:
        """Init callback.

        Args:
            evaluator (Evaluator): Evaluator.
            metrics_to_eval (list[str], Optional): Metrics to evaluate. If
                None, all metrics in the evaluator will be evaluated. Defaults
                to None.
            save_predictions (bool): If the predictions should be saved.
                Defaults to False.
            save_prefix (str, Optional): Output directory for saving the
                evaluation results. Defaults to None.
            output_dir (str, Optional): Output directory for saving the
                evaluation results.
        """
        super().__init__(*args, **kwargs)
        self.evaluator = evaluator
        self.save_predictions = save_predictions
        self.metrics_to_eval = metrics_to_eval or self.evaluator.metrics

        if self.save_predictions:
            assert (
                output_dir is not None
            ), "If save_predictions is True, save_prefix must be provided."

            output_dir = os.path.join(output_dir, "eval")

            self.output_dir = output_dir
            self.save_prefix = save_prefix

    def setup(
        self, trainer: pl.Trainer, pl_module: pl.LightningModule, stage: str
    ) -> None:  # pragma: no cover
        """Setup callback."""
        if self.save_predictions:
            self.output_dir = broadcast(self.output_dir)

            if self.save_prefix is not None:
                self.output_dir = os.path.join(
                    self.output_dir, self.save_prefix
                )

            for metric in self.metrics_to_eval:
                output_dir = os.path.join(self.output_dir, metric)
                os.makedirs(output_dir, exist_ok=True)
        self.evaluator.reset()

    def on_validation_batch_end(  # type: ignore
        self,
        trainer: pl.Trainer,
        pl_module: pl.LightningModule,
        outputs: Any,
        batch: Any,
        batch_idx: int,
        dataloader_idx: int = 0,
    ) -> None:
        """Hook to run at the end of a validation batch."""
        self.on_test_batch_end(
            trainer=trainer,
            pl_module=pl_module,
            outputs=outputs,
            batch=batch,
            batch_idx=batch_idx,
            dataloader_idx=dataloader_idx,
        )

    def on_validation_epoch_end(
        self, trainer: pl.Trainer, pl_module: pl.LightningModule
    ) -> None:
        """Wait for on_validation_epoch_end PL hook to call 'evaluate'."""
        log_dict = self.run_eval()

        for k, v in log_dict.items():
            pl_module.log(f"val/{k}", v, sync_dist=True, rank_zero_only=True)

    def on_test_batch_end(  # type: ignore
        self,
        trainer: pl.Trainer,
        pl_module: pl.LightningModule,
        outputs: DictData,
        batch: DictData,
        batch_idx: int,
        dataloader_idx: int = 0,
    ) -> None:
        """Hook to run at the end of a testing batch."""
        self.evaluator.process_batch(
            **self.get_test_callback_inputs(outputs, batch)
        )
        for metric in self.metrics_to_eval:
            # Save output predictions in current batch.
            if self.save_predictions:
                output_dir = os.path.join(self.output_dir, metric)
                self.evaluator.save_batch(metric, output_dir)

    def on_test_epoch_end(
        self, trainer: pl.Trainer, pl_module: pl.LightningModule
    ) -> None:
        """Hook to run at the end of a testing epoch."""
        log_dict = self.run_eval()

        for k, v in log_dict.items():
            pl_module.log(f"test/{k}", v, sync_dist=True, rank_zero_only=True)

    def run_eval(self) -> MetricLogs:
        """Run evaluation for the given evaluator."""
        self.evaluator.gather(all_gather_object_cpu)

        synchronize()
        self.process()

        log_dict: MetricLogs = {}
        for metric in self.metrics_to_eval:
            metric_dict = self.evaluate(metric)
            metric_dict = broadcast(metric_dict)
            assert isinstance(metric_dict, dict)
            log_dict.update(metric_dict)

        self.evaluator.reset()

        return log_dict

    @rank_zero_only
    def process(self) -> None:
        """Process the evaluator."""
        self.evaluator.process()

    @rank_zero_only
    def evaluate(self, metric: str) -> MetricLogs:
        """Evaluate the performance after processing all input/output pairs.

        Returns:
            MetricLogs: A dictionary containing the evaluation results. The
                keys are formatted as {metric_name}/{key_name}, and the
                values are the corresponding evaluated values.
        """
        rank_zero_info(
            f"Running evaluator {str(self.evaluator)} with {metric} metric... "
        )
        log_dict = {}

        # Save output predictions. This is done here instead of
        # on_test_batch_end because the evaluator may not have processed
        # all batches yet.
        if self.save_predictions:
            output_dir = os.path.join(self.output_dir, metric)
            self.evaluator.save(metric, output_dir)

        # Evaluate metric
        metric_dict, metric_str = self.evaluator.evaluate(metric)
        for k, v in metric_dict.items():
            log_k = metric + "/" + k
            rank_zero_info("%s: %.4f", log_k, v)
            log_dict[f"{metric}/{k}"] = v

        rank_zero_info("Showing results for metric: %s", metric)
        rank_zero_info(metric_str)

        return log_dict