swadhin / eu-air-traffic public

style: apply ruff formatting

swadhinbiswas committed 2 months ago
706e694
1 parent 3862474
commit 706e694
9 files changed
+205 -145
Showing 9 changed files+205-145
Air Traffic Warehouse/01 - Domain Knowledge/Airlines.mdAir Traffic Warehouse/01 - Domain Knowledge/Airports.mdAir Traffic Warehouse/01 - Domain Knowledge/Delay Types.mdAir Traffic Warehouse/01 - Domain Knowledge/Weather.mdAir Traffic Warehouse/02 - Data Engineering Concepts/ETL vs ELT.mdAir Traffic Warehouse/02 - Data Engineering Concepts/Medallion Architecture.mdAir Traffic Warehouse/07 - Polars/GroupBy.mdAir Traffic Warehouse/07 - Polars/Lazy API.mdstreamlit_app.py
▼Air Traffic Warehouse/01 - Domain Knowledge/Airlines.md
+10-5
@@ -148,16 +148,21 @@ airline_id,airline_name,alias,iata,icao,callsign,country,active
148
148
import polars as pl
149
149
150
150
schema = {
151
- "id": pl.Int32, "name": pl.Utf8, "alias": pl.Utf8,
152
- "iata": pl.Utf8, "icao": pl.Utf8, "callsign": pl.Utf8,
153
- "country": pl.Utf8, "active": pl.Utf8
151
+ "id": pl.Int32,
152
+ "name": pl.Utf8,
153
+ "alias": pl.Utf8,
154
+ "iata": pl.Utf8,
155
+ "icao": pl.Utf8,
156
+ "callsign": pl.Utf8,
157
+ "country": pl.Utf8,
158
+ "active": pl.Utf8,
154
159
}
155
160
156
161
airlines = (
157
162
pl.read_csv("airlines.dat", has_header=False, new_columns=list(schema.keys()))
158
163
.filter(pl.col("active") == "Y")
159
- .filter(pl.col("iata") != "") # Only airlines with IATA code
160
- .filter(pl.col("icao") != "") # Only airlines with ICAO code
164
+ .filter(pl.col("iata") != "") # Only airlines with IATA code
165
+ .filter(pl.col("icao") != "") # Only airlines with ICAO code
161
166
)
162
167
```
163
168
164
169
▼Air Traffic Warehouse/01 - Domain Knowledge/Airports.md
+16-13
@@ -111,6 +111,7 @@ For route efficiency analysis, we calculate distance between airports:
111
111
```python
112
112
import math
113
113
114
+
114
115
def haversine(lat1, lon1, lat2, lon2):
115
116
"""Calculate great-circle distance between two airports in km."""
116
117
R = 6371 # Earth radius in km
@@ -119,12 +120,14 @@ def haversine(lat1, lon1, lat2, lon2):
119
120
delta_phi = math.radians(lat2 - lat1)
120
121
delta_lambda = math.radians(lon2 - lon1)
121
122
122
- a = math.sin(delta_phi/2)**2 \
123
- + math.cos(phi1) * math.cos(phi2) \
124
- * math.sin(delta_lambda/2)**2
125
- c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
123
+ a = (
124
+ math.sin(delta_phi / 2) ** 2
125
+ + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2) ** 2
126
+ )
127
+ c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
126
128
return R * c
127
129
130
+
128
131
# Example: JFK → LHR
129
132
distance = haversine(40.6413, -73.7781, 51.4700, -0.4543)
130
133
# ≈ 5,541 km
@@ -144,12 +147,14 @@ import polars as pl
144
147
airport_delays = (
145
148
pl.scan_parquet("silver/flights/*.parquet")
146
149
.group_by("departure_airport")
147
- .agg([
148
- pl.col("delay_minutes").mean().alias("avg_delay"),
149
- pl.col("delay_minutes").quantile(0.95).alias("p95_delay"),
150
- pl.len().alias("flight_count"),
151
- (pl.col("status") == "C").sum().alias("cancellations"),
152
- ])
150
+ .agg(
151
+ [
152
+ pl.col("delay_minutes").mean().alias("avg_delay"),
153
+ pl.col("delay_minutes").quantile(0.95).alias("p95_delay"),
154
+ pl.len().alias("flight_count"),
155
+ (pl.col("status") == "C").sum().alias("cancellations"),
156
+ ]
157
+ )
153
158
.sort("avg_delay", descending=True)
154
159
.collect()
155
160
)
@@ -165,9 +170,7 @@ The `airports` ingestion module pulls from OurAirports/OpenFlights:
165
170
```python
166
171
# Airports ingestion: Filter to relevant airports
167
172
airports_df = pl.read_csv("airports.dat")
168
-commercial = airports_df.filter(
169
- pl.col("type").is_in(["large_airport", "medium_airport"])
170
-)
173
+commercial = airports_df.filter(pl.col("type").is_in(["large_airport", "medium_airport"]))
171
174
```
172
175
173
176
## Airport In SCD Tracking
174
177
▼Air Traffic Warehouse/01 - Domain Knowledge/Delay Types.md
+4-2
@@ -229,8 +229,10 @@ GROUP BY
229
229
validation_rules = [
230
230
(pl.col("delay_minutes") >= 0, "Negative delay"),
231
231
(pl.col("delay_minutes") <= 1440, "Delay > 24 hours (unrealistic)"),
232
- (pl.col("actual_departure") > pl.col("scheduled_departure"),
233
- "Early departure with positive delay — inconsistent"),
232
+ (
233
+ pl.col("actual_departure") > pl.col("scheduled_departure"),
234
+ "Early departure with positive delay — inconsistent",
235
+ ),
234
236
]
235
237
```
236
238
237
239
▼Air Traffic Warehouse/01 - Domain Knowledge/Weather.md
+18-23
@@ -67,10 +67,10 @@ Weather is the **#1 cause of flight delays globally**, responsible for ~30–40%
67
67
def classify_adverse_weather(row):
68
68
"""Return True if weather is likely to cause delays."""
69
69
return (
70
- row["visibility_km"] < 3.0 or
71
- row["wind_speed_kmh"] > 50 or
72
- row["precipitation_mm"] > 5 or
73
- row["condition"] in ["Thunderstorm", "Snow", "Fog", "Blizzard"]
70
+ row["visibility_km"] < 3.0
71
+ or row["wind_speed_kmh"] > 50
72
+ or row["precipitation_mm"] > 5
73
+ or row["condition"] in ["Thunderstorm", "Snow", "Fog", "Blizzard"]
74
74
)
75
75
```
76
76
@@ -104,11 +104,10 @@ import requests
104
104
105
105
url = "https://api.open-meteo.com/v1/forecast"
106
106
params = {
107
- "latitude": 40.6413, # JFK
107
+ "latitude": 40.6413, # JFK
108
108
"longitude": -73.7781,
109
- "hourly": ["temperature_2m", "relative_humidity_2m",
110
- "wind_speed_10m", "visibility"],
111
- "timezone": "UTC"
109
+ "hourly": ["temperature_2m", "relative_humidity_2m", "wind_speed_10m", "visibility"],
110
+ "timezone": "UTC",
112
111
}
113
112
response = requests.get(url, params=params)
114
113
weather_data = response.json()
@@ -123,16 +122,12 @@ Weather data is typically **hourly**. Flights have specific departure/arrival ti
123
122
124
123
```python
125
124
# Polars temporal join: nearest weather for each flight
126
-flights_with_weather = (
127
- flights
128
- .sort("scheduled_departure")
129
- .join_asof(
130
- weather.sort("observation_time"),
131
- left_on="scheduled_departure",
132
- right_on="observation_time",
133
- tolerance="1h", # within 1 hour
134
- strategy="nearest"
135
- )
125
+flights_with_weather = flights.sort("scheduled_departure").join_asof(
126
+ weather.sort("observation_time"),
127
+ left_on="scheduled_departure",
128
+ right_on="observation_time",
129
+ tolerance="1h", # within 1 hour
130
+ strategy="nearest",
136
131
)
137
132
```
138
133
@@ -174,11 +169,11 @@ ORDER BY avg_delay DESC;
174
169
# Weather validation rules
175
170
def validate_weather_row(row) -> bool:
176
171
return (
177
- -80 <= row["temperature_c"] <= 60 and
178
- 0 <= row["humidity_pct"] <= 100 and
179
- 0 <= row["wind_speed_kmh"] <= 300 and
180
- 0 <= row["visibility_km"] <= 100 and
181
- 800 <= row["pressure_hpa"] <= 1100
172
+ -80 <= row["temperature_c"] <= 60
173
+ and 0 <= row["humidity_pct"] <= 100
174
+ and 0 <= row["wind_speed_kmh"] <= 300
175
+ and 0 <= row["visibility_km"] <= 100
176
+ and 800 <= row["pressure_hpa"] <= 1100
182
177
)
183
178
```
184
179
185
180
▼Air Traffic Warehouse/02 - Data Engineering Concepts/ETL vs ELT.md
+1-2
@@ -86,8 +86,7 @@ import polars as pl
86
86
raw = pl.scan_parquet("bronze/flights/*.parquet")
87
87
88
88
silver = (
89
- raw
90
- .unique(subset=["flight_id", "date"])
89
+ raw.unique(subset=["flight_id", "date"])
91
90
.with_columns(
92
91
pl.col("scheduled_departure").str.to_datetime().dt.convert_time_zone("UTC"),
93
92
pl.col("actual_departure").str.to_datetime().dt.convert_time_zone("UTC"),
94
93
▼Air Traffic Warehouse/02 - Data Engineering Concepts/Medallion Architecture.md
+40-32
@@ -58,8 +58,8 @@ bronze/
58
58
def load_to_bronze(raw_data: pl.DataFrame, source: str, ingestion_date: str):
59
59
"""Store raw data in Bronze layer. No transformations."""
60
60
path = f"bronze/{source}/{ingestion_date}/ingestion_{ingestion_date}.parquet"
61
- raw_data.write_parquet(path) # As-is, zero processing
62
- upload_to_hugging_face(path) # Persist to HF
61
+ raw_data.write_parquet(path) # As-is, zero processing
62
+ upload_to_hugging_face(path) # Persist to HF
63
63
log.info(f"Bronze: {source} | {raw_data.height} rows stored at {path}")
64
64
```
65
65
@@ -106,25 +106,25 @@ def bronze_to_silver(bronze_path: str, source: str):
106
106
# 1. Deduplicate by business key
107
107
.unique(subset=["flight_id", "date"], keep="last")
108
108
# 2. Normalize timestamps to UTC
109
- .with_columns([
110
- pl.col("scheduled_departure")
111
- .str.to_datetime()
112
- .dt.convert_time_zone("UTC"),
113
- pl.col("actual_departure")
114
- .str.to_datetime()
115
- .dt.convert_time_zone("UTC"),
116
- ])
109
+ .with_columns(
110
+ [
111
+ pl.col("scheduled_departure").str.to_datetime().dt.convert_time_zone("UTC"),
112
+ pl.col("actual_departure").str.to_datetime().dt.convert_time_zone("UTC"),
113
+ ]
114
+ )
117
115
# 3. Validate business rules
118
116
.filter(
119
- (pl.col("delay_minutes") >= 0) &
120
- (pl.col("airport_code").str.len_bytes() == 3) &
121
- (pl.col("actual_arrival") > pl.col("actual_departure"))
117
+ (pl.col("delay_minutes") >= 0)
118
+ & (pl.col("airport_code").str.len_bytes() == 3)
119
+ & (pl.col("actual_arrival") > pl.col("actual_departure"))
122
120
)
123
121
# 4. Cast types
124
- .with_columns([
125
- pl.col("delay_minutes").cast(pl.Int32),
126
- pl.col("airport_code").str.to_uppercase(),
127
- ])
122
+ .with_columns(
123
+ [
124
+ pl.col("delay_minutes").cast(pl.Int32),
125
+ pl.col("airport_code").str.to_uppercase(),
126
+ ]
127
+ )
128
128
.collect()
129
129
)
130
130
@@ -132,8 +132,10 @@ def bronze_to_silver(bronze_path: str, source: str):
132
132
output_path = f"silver/{source}/{month}/"
133
133
silver.write_parquet(output_path)
134
134
135
- log.info(f"Silver: {source} | {silver.height} clean rows | "
136
- f"{raw.collect().height - silver.height} rows discarded")
135
+ log.info(
136
+ f"Silver: {source} | {silver.height} clean rows | "
137
+ f"{raw.collect().height - silver.height} rows discarded"
138
+ )
137
139
```
138
140
139
141
## Gold Layer — The "Business Zone"
@@ -183,21 +185,27 @@ def silver_to_gold(silver_path: str, airport_dim: pl.DataFrame):
183
185
184
186
# Generate surrogate keys
185
187
fact_flights = (
186
- flights
187
- .join(airport_dim_lazy.select(["airport_key", "airport_code"]),
188
- left_on="departure_airport", right_on="airport_code",
189
- how="left")
190
- .with_columns([
191
- pl.col("airport_key").alias("departure_airport_key"),
192
- ])
188
+ flights.join(
189
+ airport_dim_lazy.select(["airport_key", "airport_code"]),
190
+ left_on="departure_airport",
191
+ right_on="airport_code",
192
+ how="left",
193
+ )
194
+ .with_columns(
195
+ [
196
+ pl.col("airport_key").alias("departure_airport_key"),
197
+ ]
198
+ )
193
199
.drop(["airport_code_right", "departure_airport"])
194
200
# Calculate metrics
195
- .with_columns([
196
- (pl.col("actual_arrival") - pl.col("actual_departure"))
197
- .dt.total_minutes().alias("flight_duration_min"),
198
- pl.when(pl.col("status") == "C").then(True).otherwise(False)
199
- .alias("is_cancelled"),
200
- ])
201
+ .with_columns(
202
+ [
203
+ (pl.col("actual_arrival") - pl.col("actual_departure"))
204
+ .dt.total_minutes()
205
+ .alias("flight_duration_min"),
206
+ pl.when(pl.col("status") == "C").then(True).otherwise(False).alias("is_cancelled"),
207
+ ]
208
+ )
201
209
.collect()
202
210
)
203
211
204
212
▼Air Traffic Warehouse/07 - Polars/GroupBy.md
+6-4
@@ -6,10 +6,12 @@ Grouping data is a core transformation operation.
6
6
7
7
## Syntax
8
8
```python
9
-df.group_by("airline_code").agg([
10
- pl.col("delay_minutes").mean().alias("avg_delay"),
11
- pl.col("flight_id").count().alias("total_flights")
12
-])
9
+df.group_by("airline_code").agg(
10
+ [
11
+ pl.col("delay_minutes").mean().alias("avg_delay"),
12
+ pl.col("flight_id").count().alias("total_flights"),
13
+ ]
14
+)
13
15
```
14
16
Polars executes these groupings in parallel, making it vastly superior to Pandas for large datasets.
15
17
16
18
▼Air Traffic Warehouse/07 - Polars/Lazy API.md
+2-6
@@ -10,12 +10,8 @@ Polars offers both Eager and Lazy execution. The Lazy API builds a query plan an
10
10
11
11
## Example
12
12
```python
13
-q = (
14
- pl.scan_parquet("data/*.parquet")
15
- .filter(pl.col("delay") > 0)
16
- .select(["flight_id", "delay"])
17
-)
18
-result = q.collect() # Execution happens here
13
+q = pl.scan_parquet("data/*.parquet").filter(pl.col("delay") > 0).select(["flight_id", "delay"])
14
+result = q.collect() # Execution happens here
19
15
```
20
16
21
17
---
22
18
▼streamlit_app.py
+108-58
@@ -46,16 +46,21 @@ def get_connection():
46
46
"""Read-only DuckDB connection (cached across reruns)."""
47
47
if settings.duckdb_path.exists():
48
48
return duckdb.connect(str(settings.duckdb_path), read_only=True)
49
-
49
+
50
50
# If no local DB, try pulling from Hugging Face Hub (for Streamlit Cloud)
51
51
import os
52
+
52
53
try:
53
54
from huggingface_hub import hf_hub_download
55
+
54
56
repo_id = os.environ.get("HF_REPO", "swadhinbiswas/air-traffic")
55
- db_path = hf_hub_download(repo_id=repo_id, repo_type="dataset", filename="air_traffic.duckdb")
57
+ db_path = hf_hub_download(
58
+ repo_id=repo_id, repo_type="dataset", filename="air_traffic.duckdb"
59
+ )
56
60
return duckdb.connect(db_path, read_only=True)
57
61
except Exception as e: # noqa: BLE001
58
62
import streamlit as st
63
+
59
64
st.error(f"Failed to connect to Hugging Face Hub: {e}")
60
65
return None
61
66
@@ -150,13 +155,20 @@ if page == "Overview":
150
155
151
156
with col_left:
152
157
st.subheader("Flight Status Breakdown")
153
- status_data = query_df("SELECT status, COUNT(*) as count FROM fact_flights GROUP BY status")
158
+ status_data = query_df(
159
+ "SELECT status, COUNT(*) as count FROM fact_flights GROUP BY status"
160
+ )
154
161
if status_data is not None and not status_data.is_empty():
155
162
fig = px.pie(
156
163
status_data.to_pandas(),
157
164
values="count",
158
165
names="status",
159
- color_discrete_map={"scheduled": "#38bdf8", "delayed": "#fbbf24", "cancelled": "#f87171", "landed": "#34d399"},
166
+ color_discrete_map={
167
+ "scheduled": "#38bdf8",
168
+ "delayed": "#fbbf24",
169
+ "cancelled": "#f87171",
170
+ "landed": "#34d399",
171
+ },
160
172
hole=0.4,
161
173
)
162
174
fig.update_layout(height=350, margin={"t": 20, "b": 20, "l": 20, "r": 20})
@@ -195,7 +207,9 @@ if page == "Overview":
195
207
"Critical (>3 hours)": "#dc2626",
196
208
},
197
209
)
198
- fig.update_layout(height=350, showlegend=False, xaxis_title="", yaxis_title="Flights")
210
+ fig.update_layout(
211
+ height=350, showlegend=False, xaxis_title="", yaxis_title="Flights"
212
+ )
199
213
st.plotly_chart(fig, use_container_width=True)
200
214
else:
201
215
st.info("No data available. Run the pipeline to generate flight data.")
@@ -241,16 +255,20 @@ elif page == "Airports":
241
255
color_continuous_scale="RdYlGn",
242
256
labels={"on_time_rate": "OTP %"},
243
257
)
244
- fig.update_layout(height=450, xaxis_title="Total Flights", yaxis_title="Avg Delay (min)")
258
+ fig.update_layout(
259
+ height=450, xaxis_title="Total Flights", yaxis_title="Avg Delay (min)"
260
+ )
245
261
st.plotly_chart(fig, use_container_width=True)
246
262
247
263
st.subheader("Airport Details")
248
264
st.dataframe(
249
- data.to_pandas().style.format({
250
- "avg_delay_minutes": "{:.1f}",
251
- "max_delay_minutes": "{:.0f}",
252
- "on_time_rate": "{:.1%}",
253
- }),
265
+ data.to_pandas().style.format(
266
+ {
267
+ "avg_delay_minutes": "{:.1f}",
268
+ "max_delay_minutes": "{:.0f}",
269
+ "on_time_rate": "{:.1%}",
270
+ }
271
+ ),
254
272
use_container_width=True,
255
273
)
256
274
else:
@@ -281,7 +299,9 @@ elif page == "Airlines":
281
299
color_continuous_scale="RdYlGn",
282
300
labels={"on_time_rate": "OTP %"},
283
301
)
284
- fig.update_layout(height=400, xaxis_title="", yaxis_title="On-Time Rate", yaxis_tickformat=".0%")
302
+ fig.update_layout(
303
+ height=400, xaxis_title="", yaxis_title="On-Time Rate", yaxis_tickformat=".0%"
304
+ )
285
305
st.plotly_chart(fig, use_container_width=True)
286
306
287
307
with col2:
@@ -298,10 +318,12 @@ elif page == "Airlines":
298
318
299
319
st.subheader("Airline Rankings")
300
320
st.dataframe(
301
- data.to_pandas().style.format({
302
- "avg_delay_minutes": "{:.1f}",
303
- "on_time_rate": "{:.1%}",
304
- }),
321
+ data.to_pandas().style.format(
322
+ {
323
+ "avg_delay_minutes": "{:.1f}",
324
+ "on_time_rate": "{:.1%}",
325
+ }
326
+ ),
305
327
use_container_width=True,
306
328
)
307
329
else:
@@ -346,16 +368,20 @@ elif page == "Weather":
346
368
color_continuous_scale="Viridis",
347
369
labels={"avg_wind_speed_ms": "Wind (m/s)"},
348
370
)
349
- fig.update_layout(height=400, xaxis_title="Avg Temperature (C)", yaxis_title="Avg Delay (min)")
371
+ fig.update_layout(
372
+ height=400, xaxis_title="Avg Temperature (C)", yaxis_title="Avg Delay (min)"
373
+ )
350
374
st.plotly_chart(fig, use_container_width=True)
351
375
352
376
st.subheader("Weather Conditions Detail")
353
377
st.dataframe(
354
- data.to_pandas().style.format({
355
- "avg_delay_minutes": "{:.1f}",
356
- "avg_temperature_c": "{:.1f}",
357
- "avg_wind_speed_ms": "{:.1f}",
358
- }),
378
+ data.to_pandas().style.format(
379
+ {
380
+ "avg_delay_minutes": "{:.1f}",
381
+ "avg_temperature_c": "{:.1f}",
382
+ "avg_wind_speed_ms": "{:.1f}",
383
+ }
384
+ ),
359
385
use_container_width=True,
360
386
)
361
387
else:
@@ -390,7 +416,12 @@ elif page == "Delays":
390
416
x="status",
391
417
y="flight_count",
392
418
color="status",
393
- color_discrete_map={"scheduled": "#38bdf8", "delayed": "#fbbf24", "cancelled": "#f87171", "landed": "#34d399"},
419
+ color_discrete_map={
420
+ "scheduled": "#38bdf8",
421
+ "delayed": "#fbbf24",
422
+ "cancelled": "#f87171",
423
+ "landed": "#34d399",
424
+ },
394
425
)
395
426
fig.update_layout(height=350, showlegend=False, xaxis_title="", yaxis_title="Flights")
396
427
st.plotly_chart(fig, use_container_width=True)
@@ -398,30 +429,36 @@ elif page == "Delays":
398
429
with col2:
399
430
st.subheader("Delay Statistics by Status")
400
431
st.dataframe(
401
- status_data.to_pandas().style.format({
402
- "avg_delay_minutes": "{:.1f}",
403
- "min_delay_minutes": "{:.0f}",
404
- "max_delay_minutes": "{:.0f}",
405
- }),
432
+ status_data.to_pandas().style.format(
433
+ {
434
+ "avg_delay_minutes": "{:.1f}",
435
+ "min_delay_minutes": "{:.0f}",
436
+ "max_delay_minutes": "{:.0f}",
437
+ }
438
+ ),
406
439
use_container_width=True,
407
440
)
408
441
409
442
if seasonal_data is not None and not seasonal_data.is_empty():
410
443
st.subheader("Daily Flight Volume and Delay Trend")
411
444
fig = go.Figure()
412
- fig.add_trace(go.Bar(
413
- x=seasonal_data["flight_date"].to_list(),
414
- y=seasonal_data["daily_flights"].to_list(),
415
- name="Flights",
416
- yaxis="y",
417
- ))
418
- fig.add_trace(go.Scatter(
419
- x=seasonal_data["flight_date"].to_list(),
420
- y=seasonal_data["avg_delay"].to_list(),
421
- name="Avg Delay (min)",
422
- yaxis="y2",
423
- line={"color": "#f87171", "width": 2},
424
- ))
445
+ fig.add_trace(
446
+ go.Bar(
447
+ x=seasonal_data["flight_date"].to_list(),
448
+ y=seasonal_data["daily_flights"].to_list(),
449
+ name="Flights",
450
+ yaxis="y",
451
+ )
452
+ )
453
+ fig.add_trace(
454
+ go.Scatter(
455
+ x=seasonal_data["flight_date"].to_list(),
456
+ y=seasonal_data["avg_delay"].to_list(),
457
+ name="Avg Delay (min)",
458
+ yaxis="y2",
459
+ line={"color": "#f87171", "width": 2},
460
+ )
461
+ )
425
462
fig.update_layout(
426
463
height=400,
427
464
yaxis={"title": "Flights"},
@@ -455,14 +492,16 @@ elif page == "Quality":
455
492
if sources:
456
493
rows = []
457
494
for source, metrics in sources.items():
458
- rows.append({
459
- "Source": source,
460
- "Bronze": metrics.get("bronze_rows", 0),
461
- "Silver": metrics.get("silver_rows", 0),
462
- "Quarantined": metrics.get("quarantined_rows", 0),
463
- "Pass Rate": metrics.get("pass_rate"),
464
- "Freshness": metrics.get("freshness", "N/A"),
465
- })
495
+ rows.append(
496
+ {
497
+ "Source": source,
498
+ "Bronze": metrics.get("bronze_rows", 0),
499
+ "Silver": metrics.get("silver_rows", 0),
500
+ "Quarantined": metrics.get("quarantined_rows", 0),
501
+ "Pass Rate": metrics.get("pass_rate"),
502
+ "Freshness": metrics.get("freshness", "N/A"),
503
+ }
504
+ )
466
505
df = pd.DataFrame(rows)
467
506
468
507
fig = px.bar(
@@ -531,13 +570,19 @@ LIMIT 10"""
531
570
col1, col2, col3 = st.columns(3)
532
571
with col1:
533
572
if st.button("Top Airports"):
534
- st.session_state["quick_sql"] = "SELECT * FROM gold_airport_metrics ORDER BY total_flights DESC LIMIT 10"
573
+ st.session_state["quick_sql"] = (
574
+ "SELECT * FROM gold_airport_metrics ORDER BY total_flights DESC LIMIT 10"
575
+ )
535
576
with col2:
536
577
if st.button("Airline Rankings"):
537
- st.session_state["quick_sql"] = "SELECT * FROM gold_airline_rankings ORDER BY avg_delay_minutes"
578
+ st.session_state["quick_sql"] = (
579
+ "SELECT * FROM gold_airline_rankings ORDER BY avg_delay_minutes"
580
+ )
538
581
with col3:
539
582
if st.button("Weather Impact"):
540
- st.session_state["quick_sql"] = "SELECT * FROM gold_weather_impact ORDER BY flight_count DESC"
583
+ st.session_state["quick_sql"] = (
584
+ "SELECT * FROM gold_weather_impact ORDER BY flight_count DESC"
585
+ )
541
586
542
587
if "quick_sql" in st.session_state:
543
588
st.code(st.session_state["quick_sql"], language="sql")
@@ -550,11 +595,14 @@ elif page == "Pipeline":
550
595
551
596
with col1:
552
597
st.subheader("Run Pipeline")
553
- st.caption("Execute the full ETL pipeline (collect, bronze, silver, gold, warehouse, quality)")
598
+ st.caption(
599
+ "Execute the full ETL pipeline (collect, bronze, silver, gold, warehouse, quality)"
600
+ )
554
601
555
602
if st.button("Run Full Pipeline", type="primary"):
556
603
with st.spinner("Running pipeline..."):
557
604
from pipelines.orchestrator import Orchestrator
605
+
558
606
report = Orchestrator().run()
559
607
st.session_state["last_report"] = report
560
608
@@ -568,11 +616,13 @@ elif page == "Pipeline":
568
616
st.subheader("Step Timings")
569
617
steps = []
570
618
for name, step in report.steps.items():
571
- steps.append({
572
- "Step": name,
573
- "Status": step.get("status", "unknown"),
574
- "Duration (s)": step.get("elapsed_seconds", 0),
575
- })
619
+ steps.append(
620
+ {
621
+ "Step": name,
622
+ "Status": step.get("status", "unknown"),
623
+ "Duration (s)": step.get("elapsed_seconds", 0),
624
+ }
625
+ )
576
626
st.dataframe(pd.DataFrame(steps), use_container_width=True)
577
627
578
628
with col2:
579
629