Skip to content

rslearn.dataset.handler_summaries

handler_summaries

This module contains dataclasses for summarizing the results of dataset operations.

They can be used by callers to emit telemetry / logs, or discarded.

LayerPrepareSummary dataclass

Results for preparing a single layer.

Source code in rslearn/dataset/handler_summaries.py 
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
@dataclass
class LayerPrepareSummary:
    """Results for preparing a single layer."""

    # Identity
    layer_name: str
    data_source_name: str

    # Timing
    duration_seconds: float

    # Counts
    windows_prepared: int
    windows_skipped: int
    windows_rejected: int
    get_items_attempts: int
    windows_failed: int = 0
    error_messages: list[str] = field(default_factory=list)

    def merge(self, other: LayerPrepareSummary) -> LayerPrepareSummary:
        """Combine two summaries for the same layer."""
        return LayerPrepareSummary(
            layer_name=self.layer_name,
            data_source_name=self.data_source_name,
            duration_seconds=self.duration_seconds + other.duration_seconds,
            windows_prepared=self.windows_prepared + other.windows_prepared,
            windows_skipped=self.windows_skipped + other.windows_skipped,
            windows_rejected=self.windows_rejected + other.windows_rejected,
            get_items_attempts=self.get_items_attempts + other.get_items_attempts,
            windows_failed=self.windows_failed + other.windows_failed,
            error_messages=self.error_messages + other.error_messages,
        )

merge 

merge(other: LayerPrepareSummary) -> LayerPrepareSummary

Combine two summaries for the same layer.

Source code in rslearn/dataset/handler_summaries.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def merge(self, other: LayerPrepareSummary) -> LayerPrepareSummary:
    """Combine two summaries for the same layer."""
    return LayerPrepareSummary(
        layer_name=self.layer_name,
        data_source_name=self.data_source_name,
        duration_seconds=self.duration_seconds + other.duration_seconds,
        windows_prepared=self.windows_prepared + other.windows_prepared,
        windows_skipped=self.windows_skipped + other.windows_skipped,
        windows_rejected=self.windows_rejected + other.windows_rejected,
        get_items_attempts=self.get_items_attempts + other.get_items_attempts,
        windows_failed=self.windows_failed + other.windows_failed,
        error_messages=self.error_messages + other.error_messages,
    )

PrepareDatasetWindowsSummary dataclass

Results from prepare_dataset_windows operation for telemetry purposes.

Source code in rslearn/dataset/handler_summaries.py 
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
@dataclass
class PrepareDatasetWindowsSummary:
    """Results from prepare_dataset_windows operation for telemetry purposes."""

    # Timing
    duration_seconds: float

    # Counts
    total_windows_requested: int

    # Per-layer summaries
    layer_summaries: dict[str, LayerPrepareSummary]

    def merge(
        self, other: PrepareDatasetWindowsSummary
    ) -> PrepareDatasetWindowsSummary:
        """Combine two summaries (e.g. from separate batches)."""
        merged: dict[str, LayerPrepareSummary] = dict(self.layer_summaries)
        for name, summary in other.layer_summaries.items():
            if name in merged:
                merged[name] = merged[name].merge(summary)
            else:
                merged[name] = summary
        return PrepareDatasetWindowsSummary(
            duration_seconds=self.duration_seconds + other.duration_seconds,
            total_windows_requested=self.total_windows_requested
            + other.total_windows_requested,
            layer_summaries=merged,
        )

merge 

merge(other: PrepareDatasetWindowsSummary) -> PrepareDatasetWindowsSummary

Combine two summaries (e.g. from separate batches).

Source code in rslearn/dataset/handler_summaries.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def merge(
    self, other: PrepareDatasetWindowsSummary
) -> PrepareDatasetWindowsSummary:
    """Combine two summaries (e.g. from separate batches)."""
    merged: dict[str, LayerPrepareSummary] = dict(self.layer_summaries)
    for name, summary in other.layer_summaries.items():
        if name in merged:
            merged[name] = merged[name].merge(summary)
        else:
            merged[name] = summary
    return PrepareDatasetWindowsSummary(
        duration_seconds=self.duration_seconds + other.duration_seconds,
        total_windows_requested=self.total_windows_requested
        + other.total_windows_requested,
        layer_summaries=merged,
    )

IngestCounts dataclass

Ingestion counts for a layer.

Source code in rslearn/dataset/handler_summaries.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
@dataclass
class IngestCounts:
    """Ingestion counts for a layer."""

    items: int
    geometries: int

    def merge(self, other: IngestCounts) -> IngestCounts:
        """Combine two ingest counts."""
        return IngestCounts(
            items=self.items + other.items,
            geometries=self.geometries + other.geometries,
        )

merge 

merge(other: IngestCounts) -> IngestCounts

Combine two ingest counts.

Source code in rslearn/dataset/handler_summaries.py 
91
92
93
94
95
96
def merge(self, other: IngestCounts) -> IngestCounts:
    """Combine two ingest counts."""
    return IngestCounts(
        items=self.items + other.items,
        geometries=self.geometries + other.geometries,
    )

LayerIngestSummary dataclass

Results for ingesting a single layer.

Source code in rslearn/dataset/handler_summaries.py 
 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
@dataclass
class LayerIngestSummary:
    """Results for ingesting a single layer."""

    # Identity
    layer_name: str
    data_source_name: str

    # Timing
    duration_seconds: float

    # Counts
    ingest_counts: IngestCounts
    ingest_attempts: int
    error_messages: list[str] = field(default_factory=list)

    def merge(self, other: LayerIngestSummary) -> LayerIngestSummary:
        """Combine two summaries for the same layer."""
        return LayerIngestSummary(
            layer_name=self.layer_name,
            data_source_name=self.data_source_name,
            duration_seconds=self.duration_seconds + other.duration_seconds,
            ingest_counts=self.ingest_counts.merge(other.ingest_counts),
            ingest_attempts=self.ingest_attempts + other.ingest_attempts,
            error_messages=self.error_messages + other.error_messages,
        )

merge 

merge(other: LayerIngestSummary) -> LayerIngestSummary

Combine two summaries for the same layer.

Source code in rslearn/dataset/handler_summaries.py 
115
116
117
118
119
120
121
122
123
124
def merge(self, other: LayerIngestSummary) -> LayerIngestSummary:
    """Combine two summaries for the same layer."""
    return LayerIngestSummary(
        layer_name=self.layer_name,
        data_source_name=self.data_source_name,
        duration_seconds=self.duration_seconds + other.duration_seconds,
        ingest_counts=self.ingest_counts.merge(other.ingest_counts),
        ingest_attempts=self.ingest_attempts + other.ingest_attempts,
        error_messages=self.error_messages + other.error_messages,
    )

IngestDatasetJobsSummary dataclass

Results from ingesting a set of jobs; for telemetry purposes.

Source code in rslearn/dataset/handler_summaries.py 
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
@dataclass
class IngestDatasetJobsSummary:
    """Results from ingesting a set of jobs; for telemetry purposes."""

    # Timing
    duration_seconds: float

    # Counts
    num_jobs: int

    # Per-layer summaries
    layer_summaries: dict[str, LayerIngestSummary]

    def merge(self, other: IngestDatasetJobsSummary) -> IngestDatasetJobsSummary:
        """Combine two summaries (e.g. from separate batches)."""
        merged: dict[str, LayerIngestSummary] = dict(self.layer_summaries)
        for name, summary in other.layer_summaries.items():
            if name in merged:
                merged[name] = merged[name].merge(summary)
            else:
                merged[name] = summary
        return IngestDatasetJobsSummary(
            duration_seconds=self.duration_seconds + other.duration_seconds,
            num_jobs=self.num_jobs + other.num_jobs,
            layer_summaries=merged,
        )

merge 

merge(other: IngestDatasetJobsSummary) -> IngestDatasetJobsSummary

Combine two summaries (e.g. from separate batches).

Source code in rslearn/dataset/handler_summaries.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
def merge(self, other: IngestDatasetJobsSummary) -> IngestDatasetJobsSummary:
    """Combine two summaries (e.g. from separate batches)."""
    merged: dict[str, LayerIngestSummary] = dict(self.layer_summaries)
    for name, summary in other.layer_summaries.items():
        if name in merged:
            merged[name] = merged[name].merge(summary)
        else:
            merged[name] = summary
    return IngestDatasetJobsSummary(
        duration_seconds=self.duration_seconds + other.duration_seconds,
        num_jobs=self.num_jobs + other.num_jobs,
        layer_summaries=merged,
    )

MaterializeWindowLayerSummary dataclass

Results for materializing a single window layer.

Source code in rslearn/dataset/handler_summaries.py 
155
156
157
158
159
160
@dataclass
class MaterializeWindowLayerSummary:
    """Results for materializing a single window layer."""

    skipped: bool
    materialize_attempts: int

MaterializeWindowLayersSummary dataclass

Results for materialize a given layer for all windows in a materialize call.

Source code in rslearn/dataset/handler_summaries.py 
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
195
196
@dataclass
class MaterializeWindowLayersSummary:
    """Results for materialize a given layer for all windows in a materialize call."""

    # Identity
    layer_name: str
    data_source_name: str

    # Timing
    duration_seconds: float

    # Counts
    total_windows_requested: int
    num_windows_materialized: int
    materialize_attempts: int
    windows_failed: int = 0
    error_messages: list[str] = field(default_factory=list)

    def merge(
        self, other: MaterializeWindowLayersSummary
    ) -> MaterializeWindowLayersSummary:
        """Combine two summaries for the same layer."""
        return MaterializeWindowLayersSummary(
            layer_name=self.layer_name,
            data_source_name=self.data_source_name,
            duration_seconds=self.duration_seconds + other.duration_seconds,
            total_windows_requested=self.total_windows_requested
            + other.total_windows_requested,
            num_windows_materialized=self.num_windows_materialized
            + other.num_windows_materialized,
            materialize_attempts=self.materialize_attempts + other.materialize_attempts,
            windows_failed=self.windows_failed + other.windows_failed,
            error_messages=self.error_messages + other.error_messages,
        )

merge 

merge(other: MaterializeWindowLayersSummary) -> MaterializeWindowLayersSummary

Combine two summaries for the same layer.

Source code in rslearn/dataset/handler_summaries.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def merge(
    self, other: MaterializeWindowLayersSummary
) -> MaterializeWindowLayersSummary:
    """Combine two summaries for the same layer."""
    return MaterializeWindowLayersSummary(
        layer_name=self.layer_name,
        data_source_name=self.data_source_name,
        duration_seconds=self.duration_seconds + other.duration_seconds,
        total_windows_requested=self.total_windows_requested
        + other.total_windows_requested,
        num_windows_materialized=self.num_windows_materialized
        + other.num_windows_materialized,
        materialize_attempts=self.materialize_attempts + other.materialize_attempts,
        windows_failed=self.windows_failed + other.windows_failed,
        error_messages=self.error_messages + other.error_messages,
    )

MaterializeDatasetWindowsSummary dataclass

Results from materialize_dataset_windows operation for telemetry purposes.

Source code in rslearn/dataset/handler_summaries.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
@dataclass
class MaterializeDatasetWindowsSummary:
    """Results from materialize_dataset_windows operation for telemetry purposes."""

    # Timing
    duration_seconds: float

    # Counts
    total_windows_requested: int

    # Per-layer summaries
    layer_summaries: dict[str, MaterializeWindowLayersSummary]

    def merge(
        self, other: MaterializeDatasetWindowsSummary
    ) -> MaterializeDatasetWindowsSummary:
        """Combine two summaries (e.g. from separate batches)."""
        merged: dict[str, MaterializeWindowLayersSummary] = dict(self.layer_summaries)
        for name, summary in other.layer_summaries.items():
            if name in merged:
                merged[name] = merged[name].merge(summary)
            else:
                merged[name] = summary
        return MaterializeDatasetWindowsSummary(
            duration_seconds=self.duration_seconds + other.duration_seconds,
            total_windows_requested=self.total_windows_requested
            + other.total_windows_requested,
            layer_summaries=merged,
        )

merge 

merge(other: MaterializeDatasetWindowsSummary) -> MaterializeDatasetWindowsSummary

Combine two summaries (e.g. from separate batches).

Source code in rslearn/dataset/handler_summaries.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def merge(
    self, other: MaterializeDatasetWindowsSummary
) -> MaterializeDatasetWindowsSummary:
    """Combine two summaries (e.g. from separate batches)."""
    merged: dict[str, MaterializeWindowLayersSummary] = dict(self.layer_summaries)
    for name, summary in other.layer_summaries.items():
        if name in merged:
            merged[name] = merged[name].merge(summary)
        else:
            merged[name] = summary
    return MaterializeDatasetWindowsSummary(
        duration_seconds=self.duration_seconds + other.duration_seconds,
        total_windows_requested=self.total_windows_requested
        + other.total_windows_requested,
        layer_summaries=merged,
    )

summarize_errors 

summarize_errors(error_messages: list[str], top_n: int = 5) -> list[tuple[str, int]]

Count occurrences of each unique error message, return top N by frequency.

Source code in rslearn/dataset/handler_summaries.py 
12
13
14
15
16
def summarize_errors(
    error_messages: list[str], top_n: int = 5
) -> list[tuple[str, int]]:
    """Count occurrences of each unique error message, return top N by frequency."""
    return Counter(error_messages).most_common(top_n)