mirror of
https://github.com/PlaneQuery/OpenAirframes.git
synced 2026-09-14 18:05:27 +02:00
fix: make registry join failures visible instead of silent
- accept source ids containing digits and underscores; the previous pattern dropped openairframes_uk_caa_*.csv with no output, contradicting the documented promise that adding a source needs no change here - reject a file whose source column disagrees with its filename, which would otherwise merge one registry into the union under another's label - refuse duplicate source files, skip an empty optional source rather than aborting, and log every skipped file and the column names shared across sources - correct the MIN_EXPECTED_ROWS comment, which described a 35k floor for a 1000 value Generated-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
+31
-5
@@ -19,7 +19,10 @@ import pandas as pd
|
|||||||
|
|
||||||
# Sources that are registries. Community and ADS-B are published separately: they are
|
# Sources that are registries. Community and ADS-B are published separately: they are
|
||||||
# observations and contributions, not registration records, and do not share this schema.
|
# observations and contributions, not registration records, and do not share this schema.
|
||||||
FILENAME_RE = re.compile(r"openairframes_(?P<source>[a-z]+)_(?P<start>\d{4}-\d{2}-\d{2})_(?P<end>\d{4}-\d{2}-\d{2})\.csv$")
|
FILENAME_RE = re.compile(
|
||||||
|
r"\Aopenairframes_(?P<source>[a-z0-9_]+?)_"
|
||||||
|
r"(?P<start>\d{4}-\d{2}-\d{2})_(?P<end>\d{4}-\d{2}-\d{2})\.csv\Z"
|
||||||
|
)
|
||||||
EXCLUDED_SOURCES = {"community", "adsb", "registry"}
|
EXCLUDED_SOURCES = {"community", "adsb", "registry"}
|
||||||
|
|
||||||
# Identifier columns lead the union so the table is usable without reading 70 headers.
|
# Identifier columns lead the union so the table is usable without reading 70 headers.
|
||||||
@@ -36,11 +39,13 @@ def discover(input_dir: Path) -> list[tuple[str, str, str, Path]]:
|
|||||||
"""Return (source, start, end, path) for each per-source registry CSV found."""
|
"""Return (source, start, end, path) for each per-source registry CSV found."""
|
||||||
found = []
|
found = []
|
||||||
for path in sorted(input_dir.rglob("openairframes_*.csv")):
|
for path in sorted(input_dir.rglob("openairframes_*.csv")):
|
||||||
match = FILENAME_RE.search(path.name)
|
match = FILENAME_RE.match(path.name)
|
||||||
if not match:
|
if not match:
|
||||||
|
print(f" SKIP {path.name}: does not match {FILENAME_RE.pattern}")
|
||||||
continue
|
continue
|
||||||
source = match.group("source")
|
source = match.group("source")
|
||||||
if source in EXCLUDED_SOURCES:
|
if source in EXCLUDED_SOURCES:
|
||||||
|
print(f" SKIP {path.name}: {source!r} is published as its own asset")
|
||||||
continue
|
continue
|
||||||
found.append((source, match.group("start"), match.group("end"), path))
|
found.append((source, match.group("start"), match.group("end"), path))
|
||||||
return found
|
return found
|
||||||
@@ -51,18 +56,37 @@ def build(input_dir: Path, date_str: str) -> tuple[pd.DataFrame, str, str]:
|
|||||||
if not parts:
|
if not parts:
|
||||||
raise SystemExit(f"No per-source registry CSVs found under {input_dir}")
|
raise SystemExit(f"No per-source registry CSVs found under {input_dir}")
|
||||||
|
|
||||||
|
seen = [p[0] for p in parts]
|
||||||
|
duplicated = {s for s in seen if seen.count(s) > 1}
|
||||||
|
if duplicated:
|
||||||
|
raise SystemExit(f"More than one file claims source {sorted(duplicated)}")
|
||||||
|
|
||||||
frames = []
|
frames = []
|
||||||
for source, _, _, path in parts:
|
for source, _, _, path in parts:
|
||||||
# keep_default_na=False so a literal "NA" survives the round trip unchanged.
|
# keep_default_na=False so a literal "NA" survives the round trip unchanged.
|
||||||
df = pd.read_csv(path, dtype=str, keep_default_na=False)
|
df = pd.read_csv(path, dtype=str, keep_default_na=False)
|
||||||
if "source" not in df.columns:
|
if "source" not in df.columns:
|
||||||
raise SystemExit(f"{path.name}: no source column; cannot discriminate rows")
|
raise SystemExit(f"{path.name}: no source column; cannot discriminate rows")
|
||||||
actual = set(df["source"].unique())
|
if df.empty:
|
||||||
if len(actual) != 1:
|
print(f" {source}: empty, skipping")
|
||||||
raise SystemExit(f"{path.name}: expected one source value, found {sorted(actual)}")
|
continue
|
||||||
|
actual = {v.strip().lower() for v in df["source"].unique()}
|
||||||
|
if actual != {source.lower()}:
|
||||||
|
# A file whose rows disagree with its name would duplicate another source into
|
||||||
|
# the union under the wrong label.
|
||||||
|
raise SystemExit(
|
||||||
|
f"{path.name}: filename says {source!r}, rows say {sorted(actual)}"
|
||||||
|
)
|
||||||
print(f" {source}: {len(df)} rows, {len(df.columns)} columns from {path.name}")
|
print(f" {source}: {len(df)} rows, {len(df.columns)} columns from {path.name}")
|
||||||
frames.append(df)
|
frames.append(df)
|
||||||
|
|
||||||
|
if not frames:
|
||||||
|
raise SystemExit("Every discovered source was empty; refusing to publish an empty registry")
|
||||||
|
|
||||||
|
if len(frames) > 1:
|
||||||
|
shared = set.intersection(*(set(f.columns) for f in frames)) - set(LEADING_COLUMNS)
|
||||||
|
print(f" columns shared across sources ({len(shared)}): {sorted(shared)}")
|
||||||
|
|
||||||
columns = list(dict.fromkeys(c for df in frames for c in df.columns))
|
columns = list(dict.fromkeys(c for df in frames for c in df.columns))
|
||||||
ordered = [c for c in LEADING_COLUMNS if c in columns]
|
ordered = [c for c in LEADING_COLUMNS if c in columns]
|
||||||
ordered += [c for c in columns if c not in ordered]
|
ordered += [c for c in columns if c not in ordered]
|
||||||
@@ -72,6 +96,8 @@ def build(input_dir: Path, date_str: str) -> tuple[pd.DataFrame, str, str]:
|
|||||||
df_union = pd.concat([df.reindex(columns=ordered) for df in frames], ignore_index=True)
|
df_union = pd.concat([df.reindex(columns=ordered) for df in frames], ignore_index=True)
|
||||||
df_union = df_union.fillna("")
|
df_union = df_union.fillna("")
|
||||||
|
|
||||||
|
# Earliest start across sources, not per-source coverage: a source added today still
|
||||||
|
# carries the oldest source's start date in the filename.
|
||||||
start = min(p[1] for p in parts)
|
start = min(p[1] for p in parts)
|
||||||
end = max(p[2] for p in parts + [("", "", date_str, Path())])
|
end = max(p[2] for p in parts + [("", "", date_str, Path())])
|
||||||
return df_union, start, end
|
return df_union, start, end
|
||||||
|
|||||||
@@ -50,8 +50,10 @@ OWNER_ADDRESS_COLUMNS = {
|
|||||||
|
|
||||||
FOOTER_RE = re.compile(r"\s*(\d+) rows selected\.\s*")
|
FOOTER_RE = re.compile(r"\s*(\d+) rows selected\.\s*")
|
||||||
|
|
||||||
# Canada's register is ~35k aircraft. Any parse yielding less than this means the
|
# Floor, not an expectation: Canada's register is ~35k aircraft and carsownr is larger
|
||||||
# export was truncated upstream, which must not be published as a real snapshot.
|
# still, so 1000 only catches a grossly truncated export. The footer row-count check above
|
||||||
|
# is what actually validates the parse; this guards the case where the footer agrees with a
|
||||||
|
# near-empty body.
|
||||||
MIN_EXPECTED_ROWS = 1000
|
MIN_EXPECTED_ROWS = 1000
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user