fd99c96
commit fd99c96
5 files changed
+174 -84
Showing 5 changed files+174-84
▼pipelines/orchestrator.py
+1-1
| @@ -105,7 +105,7 @@ class Orchestrator: | ||
105 | 105 | _step(report, "gold", self._gold) |
106 | 106 | if report.success: |
107 | 107 | _step(report, "quality", self._quality) |
108 | - if report.success and self.settings.huggingface_token: | |
108 | + if report.success and self.settings.huggingface_token and not self.settings.mock_mode: | |
109 | 109 | _step(report, "upload_hf", self._upload_hf) |
110 | 110 | else: |
111 | 111 | report.steps["upload_hf"] = {"status": "skipped", "result": "HF_TOKEN not set"} |
112 | 112 | |
▼pipelines/warehouse.py
+155-83
| @@ -106,6 +106,105 @@ _EMPTY_FACT_NOTAMS = { | ||
106 | 106 | "collected_at": "VARCHAR", |
107 | 107 | } |
108 | 108 | |
109 | +_DIM_AIRPORT_COLUMNS = ( | |
110 | + "airport_icao", | |
111 | + "name", | |
112 | + "type", | |
113 | + "latitude_deg", | |
114 | + "longitude_deg", | |
115 | + "elevation_ft", | |
116 | + "iso_country", | |
117 | + "municipality", | |
118 | + "iata_code", | |
119 | + "score", | |
120 | +) | |
121 | +_DIM_AIRPORT_NUMERIC = {"latitude_deg", "longitude_deg", "elevation_ft", "score"} | |
122 | +_EMPTY_DIM_AIRPORT = { | |
123 | + c: ("DOUBLE" if c in _DIM_AIRPORT_NUMERIC else "VARCHAR") for c in _DIM_AIRPORT_COLUMNS | |
124 | +} | |
125 | + | |
126 | +_DIM_AIRLINE_COLUMNS = ("airline_icao", "airline_name") | |
127 | +_EMPTY_DIM_AIRLINE = {"airline_icao": "VARCHAR", "airline_name": "VARCHAR"} | |
128 | + | |
129 | +_DIM_FUEL_COLUMNS = ("date", "region", "price_per_litre", "currency") | |
130 | +_EMPTY_DIM_FUEL = { | |
131 | + "date": "DATE", | |
132 | + "region": "VARCHAR", | |
133 | + "price_per_litre": "DOUBLE", | |
134 | + "currency": "VARCHAR", | |
135 | +} | |
136 | + | |
137 | +_DIM_AIRCRAFT_COLUMNS = ( | |
138 | + "type_icao", | |
139 | + "type_iata", | |
140 | + "manufacturer", | |
141 | + "family", | |
142 | + "engine", | |
143 | + "capacity", | |
144 | + "range_km", | |
145 | +) | |
146 | +_DIM_AIRCRAFT_NUMERIC = {"capacity", "range_km"} | |
147 | +_EMPTY_DIM_AIRCRAFT = { | |
148 | + c: ("DOUBLE" if c in _DIM_AIRCRAFT_NUMERIC else "VARCHAR") for c in _DIM_AIRCRAFT_COLUMNS | |
149 | +} | |
150 | + | |
151 | +_DIM_ROUTE_COLUMNS = ("origin", "destination", "airline", "stops", "equipment", "distance_km") | |
152 | +_DIM_ROUTE_NUMERIC = {"stops", "distance_km"} | |
153 | +_EMPTY_DIM_ROUTE = { | |
154 | + c: ("DOUBLE" if c in _DIM_ROUTE_NUMERIC else "VARCHAR") for c in _DIM_ROUTE_COLUMNS | |
155 | +} | |
156 | + | |
157 | +_FACT_POSITIONS_COLUMNS = ( | |
158 | + "icao24", | |
159 | + "callsign", | |
160 | + "registration", | |
161 | + "aircraft_type", | |
162 | + "aircraft_class", | |
163 | + "emitter_class", | |
164 | + "is_cargo", | |
165 | + "is_military", | |
166 | + "operator_name", | |
167 | + "operator_country", | |
168 | + "operator_category", | |
169 | + "type_name", | |
170 | + "manufacturer", | |
171 | + "airframe", | |
172 | + "wake_category", | |
173 | + "latitude", | |
174 | + "longitude", | |
175 | + "altitude", | |
176 | + "velocity", | |
177 | + "heading", | |
178 | + "vertical_rate", | |
179 | + "squawk", | |
180 | + "emergency", | |
181 | + "co2_kg_per_hour", | |
182 | + "fuel_burn_kg_per_hour", | |
183 | + "co2_estimated", | |
184 | + "source", | |
185 | + "collected_at", | |
186 | +) | |
187 | +_POSITION_NUMERIC = { | |
188 | + "latitude", | |
189 | + "longitude", | |
190 | + "altitude", | |
191 | + "velocity", | |
192 | + "heading", | |
193 | + "vertical_rate", | |
194 | + "co2_kg_per_hour", | |
195 | + "fuel_burn_kg_per_hour", | |
196 | +} | |
197 | +_POSITION_BOOLEAN = {"is_cargo", "is_military", "co2_estimated"} | |
198 | +_EMPTY_FACT_POSITIONS = { | |
199 | + column: ( | |
200 | + "BOOLEAN" | |
201 | + if column in _POSITION_BOOLEAN | |
202 | + else "DOUBLE" | |
203 | + if column in _POSITION_NUMERIC | |
204 | + else "VARCHAR" | |
205 | + ) | |
206 | + for column in _FACT_POSITIONS_COLUMNS | |
207 | +} | |
109 | 208 | _EMPTY_WEATHER = { |
110 | 209 | "station_icao": "VARCHAR", |
111 | 210 | "timestamp": "TIMESTAMP", |
| @@ -130,6 +229,33 @@ _EMPTY_WEATHER = { | ||
130 | 229 | "raw_metar": "VARCHAR", |
131 | 230 | } |
132 | 231 | |
232 | +_WEATHER_COLUMNS = tuple(_EMPTY_WEATHER.keys()) | |
233 | + | |
234 | + | |
235 | +def _ensure_columns( | |
236 | + df: pl.DataFrame, | |
237 | + columns: tuple[str, ...], | |
238 | + numeric: set[str] | frozenset[str] = frozenset(), | |
239 | + boolean: set[str] | frozenset[str] = frozenset(), | |
240 | +) -> pl.DataFrame: | |
241 | + """Project a frame onto a fixed schema, filling absent columns with NULLs. | |
242 | + | |
243 | + dbt staging views reference the full column set; a partial Silver frame (or | |
244 | + an empty one) would otherwise leave them missing and fail to compile. | |
245 | + """ | |
246 | + missing = [c for c in columns if c not in df.columns] | |
247 | + if missing: | |
248 | + exprs = [] | |
249 | + for column in missing: | |
250 | + if column in boolean: | |
251 | + exprs.append(pl.lit(None, dtype=pl.Boolean).alias(column)) | |
252 | + elif column in numeric: | |
253 | + exprs.append(pl.lit(None, dtype=pl.Float64).alias(column)) | |
254 | + else: | |
255 | + exprs.append(pl.lit(None, dtype=pl.Utf8).alias(column)) | |
256 | + df = df.with_columns(exprs) | |
257 | + return df.select(list(columns)) | |
258 | + | |
133 | 259 | |
134 | 260 | class WarehouseBuilder: |
135 | 261 | """Builds and validates the DuckDB Star Schema.""" |
| @@ -160,33 +286,22 @@ class WarehouseBuilder: | ||
160 | 286 | df = pl.read_parquet(airports_parquet) if airports_parquet.exists() else pl.DataFrame() |
161 | 287 | if df.is_empty(): |
162 | 288 | logger.info("[warehouse] dim_airport: no data") |
163 | - self._create_or_replace_empty("dim_airport", {"airport_icao": "VARCHAR"}) | |
289 | + self._create_or_replace_empty("dim_airport", _EMPTY_DIM_AIRPORT) | |
164 | 290 | return |
165 | 291 | clean = df.rename({"ident": "airport_icao"}) |
166 | - clean = clean.select( | |
167 | - "airport_icao", | |
168 | - "name", | |
169 | - "type", | |
170 | - "latitude_deg", | |
171 | - "longitude_deg", | |
172 | - "elevation_ft", | |
173 | - "iso_country", | |
174 | - "municipality", | |
175 | - "iata_code", | |
176 | - "score", | |
177 | - ).unique(subset=["airport_icao"]) | |
292 | + clean = _ensure_columns(clean, _DIM_AIRPORT_COLUMNS, _DIM_AIRPORT_NUMERIC).unique( | |
293 | + subset=["airport_icao"] | |
294 | + ) | |
178 | 295 | self._load_upsert("dim_airport", clean, ["airport_icao"]) |
179 | 296 | logger.info("[warehouse] dim_airport rows=%s", clean.height) |
180 | 297 | |
181 | 298 | def build_dim_airline(self) -> None: |
182 | 299 | flights = self._read_silver("flights") |
183 | 300 | if flights.is_empty(): |
184 | - self._create_or_replace_empty( | |
185 | - "dim_airline", {"airline_icao": "VARCHAR", "airline_name": "VARCHAR"} | |
186 | - ) | |
301 | + self._create_or_replace_empty("dim_airline", _EMPTY_DIM_AIRLINE) | |
187 | 302 | return |
188 | 303 | airlines = ( |
189 | - flights.select("airline_icao", "airline_name") | |
304 | + _ensure_columns(flights, _DIM_AIRLINE_COLUMNS) | |
190 | 305 | .filter(pl.col("airline_icao").is_not_null()) |
191 | 306 | .unique(subset=["airline_icao"]) |
192 | 307 | ) |
| @@ -213,11 +328,11 @@ class WarehouseBuilder: | ||
213 | 328 | def build_dim_fuel(self) -> None: |
214 | 329 | fuel = self._read_silver("fuel") |
215 | 330 | if fuel.is_empty(): |
216 | - self._create_or_replace_empty("dim_fuel", {"date": "DATE", "region": "VARCHAR"}) | |
331 | + self._create_or_replace_empty("dim_fuel", _EMPTY_DIM_FUEL) | |
217 | 332 | return |
218 | 333 | self._load_upsert( |
219 | 334 | "dim_fuel", |
220 | - fuel.select("date", "region", "price_per_litre", "currency"), | |
335 | + _ensure_columns(fuel, _DIM_FUEL_COLUMNS, {"price_per_litre"}), | |
221 | 336 | ["date", "region"], |
222 | 337 | ) |
223 | 338 | logger.info("[warehouse] dim_fuel rows=%s", fuel.height) |
| @@ -225,35 +340,22 @@ class WarehouseBuilder: | ||
225 | 340 | def build_dim_aircraft(self) -> None: |
226 | 341 | aircraft = self._read_silver("aircraft") |
227 | 342 | if aircraft.is_empty(): |
228 | - self._create_or_replace_empty("dim_aircraft", {"type_icao": "VARCHAR"}) | |
343 | + self._create_or_replace_empty("dim_aircraft", _EMPTY_DIM_AIRCRAFT) | |
229 | 344 | return |
230 | - dim = aircraft.select( | |
231 | - "type_icao", | |
232 | - "type_iata", | |
233 | - "manufacturer", | |
234 | - "family", | |
235 | - "engine", | |
236 | - "capacity", | |
237 | - "range_km", | |
238 | - ).unique(subset=["type_icao"]) | |
345 | + dim = _ensure_columns(aircraft, _DIM_AIRCRAFT_COLUMNS, _DIM_AIRCRAFT_NUMERIC).unique( | |
346 | + subset=["type_icao"] | |
347 | + ) | |
239 | 348 | self._load_upsert("dim_aircraft", dim, ["type_icao"]) |
240 | 349 | logger.info("[warehouse] dim_aircraft rows=%s", dim.height) |
241 | 350 | |
242 | 351 | def build_dim_route(self) -> None: |
243 | 352 | routes = self._read_silver("routes") |
244 | 353 | if routes.is_empty(): |
245 | - self._create_or_replace_empty( | |
246 | - "dim_route", {"origin": "VARCHAR", "destination": "VARCHAR"} | |
247 | - ) | |
354 | + self._create_or_replace_empty("dim_route", _EMPTY_DIM_ROUTE) | |
248 | 355 | return |
249 | - dim = routes.select( | |
250 | - "origin", | |
251 | - "destination", | |
252 | - "airline", | |
253 | - "stops", | |
254 | - "equipment", | |
255 | - "distance_km", | |
256 | - ).unique(subset=["origin", "destination", "airline"]) | |
356 | + dim = _ensure_columns(routes, _DIM_ROUTE_COLUMNS, _DIM_ROUTE_NUMERIC).unique( | |
357 | + subset=["origin", "destination", "airline"] | |
358 | + ) | |
257 | 359 | self._load_upsert("dim_route", dim, ["origin", "destination", "airline"]) |
258 | 360 | logger.info("[warehouse] dim_route rows=%s", dim.height) |
259 | 361 | |
| @@ -262,7 +364,7 @@ class WarehouseBuilder: | ||
262 | 364 | if emissions.is_empty(): |
263 | 365 | self._create_or_replace_empty("fact_emissions", _EMPTY_FACT_EMISSIONS) |
264 | 366 | return |
265 | - fact = emissions.select([c for c in _FACT_EMISSIONS_COLUMNS if c in emissions.columns]) | |
367 | + fact = _ensure_columns(emissions, _FACT_EMISSIONS_COLUMNS) | |
266 | 368 | self._load_replace("fact_emissions", fact) |
267 | 369 | logger.info("[warehouse] fact_emissions rows=%s", fact.height) |
268 | 370 | |
| @@ -271,7 +373,7 @@ class WarehouseBuilder: | ||
271 | 373 | if notams.is_empty(): |
272 | 374 | self._create_or_replace_empty("fact_notams", _EMPTY_FACT_NOTAMS) |
273 | 375 | return |
274 | - fact = notams.select([c for c in _FACT_NOTAMS_COLUMNS if c in notams.columns]) | |
376 | + fact = _ensure_columns(notams, _FACT_NOTAMS_COLUMNS) | |
275 | 377 | self._load_replace("fact_notams", fact) |
276 | 378 | logger.info("[warehouse] fact_notams rows=%s", fact.height) |
277 | 379 | |
| @@ -281,7 +383,7 @@ class WarehouseBuilder: | ||
281 | 383 | if flights.is_empty(): |
282 | 384 | self._create_or_replace_empty("fact_flights", _EMPTY_FACT_FLIGHTS) |
283 | 385 | return |
284 | - fact = flights.select([c for c in _FACT_FLIGHTS_COLUMNS if c in flights.columns]) | |
386 | + fact = _ensure_columns(flights, _FACT_FLIGHTS_COLUMNS, {"delay_minutes"}, {"cancelled"}) | |
285 | 387 | self._load_replace("fact_flights", fact) |
286 | 388 | logger.info("[warehouse] fact_flights rows=%s", fact.height) |
287 | 389 | |
| @@ -290,6 +392,11 @@ class WarehouseBuilder: | ||
290 | 392 | if weather.is_empty(): |
291 | 393 | self._create_or_replace_empty("weather", _EMPTY_WEATHER) |
292 | 394 | return |
395 | + weather = _ensure_columns( | |
396 | + weather, | |
397 | + _WEATHER_COLUMNS, | |
398 | + numeric={"temperature_c", "humidity_pct", "wind_speed_ms", "visibility_m"}, | |
399 | + ) | |
293 | 400 | self._load_replace("weather", weather) |
294 | 401 | logger.info("[warehouse] weather rows=%s", weather.height) |
295 | 402 | |
| @@ -297,47 +404,12 @@ class WarehouseBuilder: | ||
297 | 404 | """Latest live aircraft position per airframe (with class + CO₂).""" |
298 | 405 | positions = self._read_silver("positions") |
299 | 406 | if positions.is_empty(): |
300 | - self._create_or_replace_empty( | |
301 | - "fact_positions", {"icao24": "VARCHAR", "aircraft_class": "VARCHAR"} | |
302 | - ) | |
407 | + self._create_or_replace_empty("fact_positions", _EMPTY_FACT_POSITIONS) | |
303 | 408 | return |
304 | - keep = [ | |
305 | - col | |
306 | - for col in ( | |
307 | - "icao24", | |
308 | - "callsign", | |
309 | - "registration", | |
310 | - "aircraft_type", | |
311 | - "aircraft_class", | |
312 | - "emitter_class", | |
313 | - "is_cargo", | |
314 | - "is_military", | |
315 | - "operator_name", | |
316 | - "operator_country", | |
317 | - "operator_category", | |
318 | - "type_name", | |
319 | - "manufacturer", | |
320 | - "airframe", | |
321 | - "wake_category", | |
322 | - "latitude", | |
323 | - "longitude", | |
324 | - "altitude", | |
325 | - "velocity", | |
326 | - "heading", | |
327 | - "vertical_rate", | |
328 | - "squawk", | |
329 | - "emergency", | |
330 | - "co2_kg_per_hour", | |
331 | - "fuel_burn_kg_per_hour", | |
332 | - "co2_estimated", | |
333 | - "source", | |
334 | - "collected_at", | |
335 | - ) | |
336 | - if col in positions.columns | |
337 | - ] | |
338 | - fact = positions.select(keep) | |
339 | - if "icao24" in fact.columns: | |
340 | - fact = fact.unique(subset=["icao24"], keep="last") | |
409 | + fact = _ensure_columns( | |
410 | + positions, _FACT_POSITIONS_COLUMNS, _POSITION_NUMERIC, _POSITION_BOOLEAN | |
411 | + ) | |
412 | + fact = fact.unique(subset=["icao24"], keep="last") | |
341 | 413 | self._load_replace("fact_positions", fact) |
342 | 414 | logger.info("[warehouse] fact_positions rows=%s", fact.height) |
343 | 415 | |
344 | 416 | |
▼scripts/lake_sync.py
+14-0
| @@ -19,6 +19,19 @@ from config.settings import settings | ||
19 | 19 | from services import hf_lake |
20 | 20 | |
21 | 21 | |
22 | +def _assert_not_mock(action: str) -> None: | |
23 | + """Mock data must never reach the lake. | |
24 | + | |
25 | + A CI run with MOCK_MODE=true once defaulted onto the schedule and uploaded | |
26 | + synthetic flights/weather, which then failed the dbt build downstream. | |
27 | + """ | |
28 | + if settings.mock_mode: | |
29 | + raise SystemExit( | |
30 | + f"[lake_sync] refusing to {action} with MOCK_MODE=true — " | |
31 | + "the lake must only ever contain real data." | |
32 | + ) | |
33 | + | |
34 | + | |
22 | 35 | def pull_bronze() -> int: |
23 | 36 | return hf_lake.download_prefix( |
24 | 37 | f"{settings.hf_bronze_prefix}/parquet", settings.warehouse_dir, app_settings=settings |
| @@ -26,6 +39,7 @@ def pull_bronze() -> int: | ||
26 | 39 | |
27 | 40 | |
28 | 41 | def push_silver() -> int: |
42 | + _assert_not_mock("push silver") | |
29 | 43 | return hf_lake.upload_directory( |
30 | 44 | settings.silver_dir, settings.hf_silver_prefix, app_settings=settings |
31 | 45 | ) |
32 | 46 | |
▼scripts/upload_hf.py
+3-0
| @@ -68,6 +68,9 @@ _Updated automatically on each ETL run._ | ||
68 | 68 | |
69 | 69 | |
70 | 70 | def upload_warehouse() -> dict[str, Any]: |
71 | + if settings.mock_mode: | |
72 | + logger.warning("[hf] MOCK_MODE is on — refusing to upload synthetic data to the lake") | |
73 | + return {"status": "skipped", "reason": "mock_mode"} | |
71 | 74 | """Push bronze/silver/gold artifacts to the configured HF repo.""" |
72 | 75 | if not settings.huggingface_token: |
73 | 76 | logger.warning("[hf] HF_TOKEN not set — skipping upload") |
74 | 77 | |
▼tests/unit/test_scripts.py
+1-0
| @@ -91,6 +91,7 @@ def test_upload_hf_skips_without_token(monkeypatch): | ||
91 | 91 | class _FakeSettings: |
92 | 92 | huggingface_token = "" |
93 | 93 | huggingface_repo = "x/y" |
94 | + mock_mode = False | |
94 | 95 | silver_dir = upload_hf.settings.silver_dir |
95 | 96 | gold_dir = upload_hf.settings.gold_dir |
96 | 97 | duckdb_path = upload_hf.settings.duckdb_path |
97 | 98 | |