Data

def fp_modes_adsb_jsonl(base_dir: Path) -> Path:
    return base_dir / "adsblol_modes_adsb.jsonl"

def fp_datalinks_jsonl(base_dir: Path) -> Path:
    return base_dir / "adsblol_datalinks.jsonl"

def get_github_auth_token() -> str | None:
    for env_name in ("GH_TOKEN", "GITHUB_TOKEN"):
        token = os.environ.get(env_name)
        if token is not None and token.strip():
            return token.strip()

    try:
        result = subprocess.run(
            ["gh", "auth", "token"],
            check=True,
            capture_output=True,
            text=True,
        )
    except (FileNotFoundError, subprocess.CalledProcessError):
        return None

    token = result.stdout.strip()
    return token or None

def github_headers(auth_token: str | None) -> dict[str, str]:
    headers = {
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    if auth_token is not None:
        headers["Authorization"] = f"Bearer {auth_token}"
    return headers

async def fetch_release_page(
    client: AsyncClient,
    repository: str,
    *,
    page: int = 1,
    per_page: int = MAX_RELEASES_PER_PAGE,
) -> ReleasePage | None:
    response = await client.get(
        f"/repos/{repository}/releases",
        params={"page": page, "per_page": per_page},
    )
    if not response.is_success:
        return None
    releases: list[GitHubRelease] = response.json()
    return {
        "repository": repository,
        "page": page,
        "per_page": per_page,
        "releases": releases,
    }

def iter_release_asset_downloads(page: ReleasePage) -> Iterator[AdsblolAsset]:
    for release in page["releases"]:
        for asset in release["assets"]:
            yield {
                "repository": page["repository"],
                "release_tag": release["tag_name"],
                "asset_id": int(asset["id"]),
                "asset_name": str(asset["name"]),
                "asset_size": int(asset["size"]),
                "asset_digest": asset.get("digest"),
                "browser_download_url": str(asset["browser_download_url"]),
            }

def iter_metadata_rows(path: Path) -> Iterator[AdsblolAsset]:
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            if stripped := line.strip():
                yield json.loads(stripped)

async def run_metadata(
    *,
    grouped_specs: Mapping[Path, Iterable[str]],
    base_url: str,
    auth_token: str | None,
    per_page: int = MAX_RELEASES_PER_PAGE,
    max_pages: int | None = None,
) -> MetadataSyncResult:
    """Incrementally append newest adsb.lol GitHub release asset metadata.

    It treats the release history as an append-only newest-first log and stops
    per repository once a fetched page contains no unknown asset ids. It does
    not repair or backfill, in those cases delete the metadata JSONL and
    rerun to rebuild.
    """
    import httpx

    headers = github_headers(auth_token)
    file_results: list[MetadataFileSyncResult] = []

    async with httpx.AsyncClient(
        base_url=base_url,
        headers=headers,
        timeout=60.0,
        follow_redirects=True,
    ) as client:
        for output_path, repositories in grouped_specs.items():
            output_path = output_path.expanduser()
            known_asset_keys = _known_asset_keys(output_path)
            existing_assets = len(known_asset_keys)
            repository_results: list[RepositoryMetadataSyncResult] = []
            rows_to_append: list[AdsblolAsset] = []

            for repository in repositories:
                result, rows = await _sync_repository_incremental(
                    client,
                    repository,
                    known_asset_keys,
                    per_page=per_page,
                    max_pages=max_pages,
                )
                repository_results.append(result)
                rows_to_append.extend(rows)

            appended_assets = append_jsonl_rows_sync(
                output_path,
                rows_to_append,
            )
            file_results.append(
                MetadataFileSyncResult(
                    path=output_path,
                    existing_assets=existing_assets,
                    appended_assets=appended_assets,
                    repositories=tuple(repository_results),
                )
            )

    return MetadataSyncResult(files=tuple(file_results))

def append_jsonl_rows_sync(path: Path, rows: Iterable[AdsblolAsset]) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    count = 0
    with path.open("a", encoding="utf-8") as handle:
        for row in rows:
            handle.write(json.dumps(row, ensure_ascii=False) + "\n")
            count += 1
    return count

def parse_asset_time_period(asset_name: str) -> AssetTimePeriod:
    base_name = asset_name.removesuffix(".zst")

    if match := _DATALINK_RANGE_RE.match(base_name):
        return AssetTimePeriod(
            asset_name=asset_name,
            start_at=_parse_utc_datetime(match["start"], "%Y%m%d-%H%M%S"),
            end_at=_parse_utc_datetime(match["end"], "%Y%m%d-%H%M%S"),
        )

    if match := _DATALINK_DAY_RE.match(base_name):
        start_at = _parse_utc_datetime(match["day"], "%Y-%m-%d")
        return AssetTimePeriod(
            asset_name=asset_name,
            start_at=start_at,
            end_at=start_at + timedelta(days=1),
        )

    if match := _ADSB_DAY_RE.match(base_name):
        start_at = _parse_utc_datetime(
            match["day"].replace(".", "-"),
            "%Y-%m-%d",
        )
        return AssetTimePeriod(
            asset_name=asset_name,
            start_at=start_at,
            end_at=start_at + timedelta(days=1),
        )

    raise ValueError(f"unrecognized asset name: {asset_name}")

def parse_mode_asset_period(asset_name: str) -> ModeAssetPeriod:
    if not (match := _MODE_DAY_RE.match(asset_name)):
        raise ValueError(f"unrecognized modes asset name: {asset_name}")

    return ModeAssetPeriod(
        asset_name=asset_name,
        day=_parse_utc_datetime(
            match["day"].replace(".", "-"),
            "%Y-%m-%d",
        ).date(),
        variant=cast(ModeVariant, match["variant"]),
        replica=match["replica"],
        part=match["part"],
    )

def dir_datalinks(base_dir: Path) -> Path:
    return base_dir / "adsblol_datalinks"

def plan_datalinks(
    *,
    metadata_path: Path,
    output_root: Path,
    start_date: date | datetime | None = None,
    end_date: date | datetime | None = None,
) -> DatalinkPlan:
    metadata_path = metadata_path.expanduser()
    output_root = output_root.expanduser()
    start = (
        start_date.date() if isinstance(start_date, datetime) else start_date
    )
    end = end_date.date() if isinstance(end_date, datetime) else end_date
    existing_jsonl = {
        path.name
        for path in output_root.glob("*.jsonl")
        if path.is_file() and path.stat().st_size > 0
    }
    selected_by_repo: Counter[str] = Counter()
    existing_by_repo: Counter[str] = Counter()
    items: list[DatalinkDownloadItem] = []
    seen_asset_keys: set[AssetKey] = set()
    seen_output_names: set[str] = set()

    for row in iter_metadata_rows(metadata_path):
        repository = row["repository"]
        if not repository.startswith("adsblol/aircraft-data-links-"):
            continue
        if row["browser_download_url"] in KNOWN_UNAVAILABLE_DATALINK_URLS:
            continue

        key = AssetKey(repository, int(row["asset_id"]))
        if key in seen_asset_keys:
            continue
        seen_asset_keys.add(key)

        asset_day = parse_asset_time_period(row["asset_name"]).day
        if start is not None and asset_day < start:
            continue
        if end is not None and asset_day > end:
            continue

        output_name = _asset_output_name(row["asset_name"])
        if output_name in seen_output_names:
            raise ValueError(f"duplicate datalink output name: {output_name}")
        seen_output_names.add(output_name)

        already_downloaded = output_name in existing_jsonl
        selected_by_repo[repository] += 1
        if already_downloaded:
            existing_by_repo[repository] += 1

        items.append(
            DatalinkDownloadItem(
                repository=repository,
                release_tag=row["release_tag"],
                asset_name=row["asset_name"],
                asset_day=asset_day,
                asset_size=int(row["asset_size"]),
                browser_download_url=row["browser_download_url"],
                output_path=output_root / output_name,
                already_downloaded=already_downloaded,
            )
        )

    return DatalinkPlan(
        metadata_path=metadata_path,
        output_root=output_root,
        items=tuple(
            sorted(
                items,
                key=lambda item: (
                    item.asset_day,
                    item.repository,
                    item.release_tag,
                    item.asset_name,
                ),
            )
        ),
        selected_by_repo=dict(sorted(selected_by_repo.items())),
        existing_by_repo=dict(sorted(existing_by_repo.items())),
    )

def datalinks_download(
    *,
    metadata_path: Path,
    output_root: Path,
    start_date: date | datetime | None = None,
    end_date: date | datetime | None = None,
    dry_run: bool = False,
    jobs: int | None = None,
) -> DatalinkDownloadResult:
    plan = plan_datalinks(
        metadata_path=metadata_path,
        output_root=output_root,
        start_date=start_date,
        end_date=end_date,
    )
    missing_items = plan.missing_items
    if dry_run or not missing_items:
        return DatalinkDownloadResult(
            plan=plan,
            existing=plan.existing,
            downloaded=0,
            unavailable=0,
            output_paths=tuple(
                item.output_path
                for item in plan.items
                if item.already_downloaded
            ),
        )

    result = _download_missing_items(missing_items, jobs=jobs)
    return DatalinkDownloadResult(
        plan=plan,
        existing=result.existing,
        downloaded=result.downloaded,
        unavailable=result.unavailable,
        output_paths=result.output_paths,
    )

def load_metadata_daily_sizes(
    paths: Iterable[Path],
    stem: str,
) -> dict[date, int]:
    daily_sizes: dict[date, int] = {}
    seen_keys: set[AssetKey] = set()

    for path in _iter_input_paths(paths, stem):
        for row in iter_metadata_rows(path):
            key = AssetKey(row["repository"], int(row["asset_id"]))
            if key in seen_keys:
                continue
            seen_keys.add(key)
            day = parse_asset_time_period(row["asset_name"]).day
            daily_sizes[day] = daily_sizes.get(day, 0) + int(row["asset_size"])

    return dict(sorted(daily_sizes.items()))

def load_modes_daily_sizes(
    paths: Iterable[Path],
    stem: str,
) -> dict[date, ModeDailySizes]:
    daily_sizes: dict[date, ModeDailySizes] = {}
    seen_keys: set[AssetKey] = set()

    for path in _iter_input_paths(paths, stem):
        for row in iter_metadata_rows(path):
            key = AssetKey(row["repository"], int(row["asset_id"]))
            if key in seen_keys:
                continue
            seen_keys.add(key)
            try:
                period = parse_mode_asset_period(row["asset_name"])
            except ValueError:
                continue
            day_sizes = daily_sizes.setdefault(period.day, ModeDailySizes())
            asset_size = int(row["asset_size"])
            if period.variant == "prod":
                day_sizes = ModeDailySizes(
                    prod=day_sizes.prod + asset_size,
                    staging=day_sizes.staging,
                    mlatonly=day_sizes.mlatonly,
                    test=day_sizes.test,
                )
            elif period.variant == "staging":
                day_sizes = ModeDailySizes(
                    prod=day_sizes.prod,
                    staging=day_sizes.staging + asset_size,
                    mlatonly=day_sizes.mlatonly,
                    test=day_sizes.test,
                )
            elif period.variant == "mlatonly":
                day_sizes = ModeDailySizes(
                    prod=day_sizes.prod,
                    staging=day_sizes.staging,
                    mlatonly=day_sizes.mlatonly + asset_size,
                    test=day_sizes.test,
                )
            elif period.variant == "test":
                day_sizes = ModeDailySizes(
                    prod=day_sizes.prod,
                    staging=day_sizes.staging,
                    mlatonly=day_sizes.mlatonly,
                    test=day_sizes.test + asset_size,
                )
            daily_sizes[period.day] = day_sizes

    return dict(sorted(daily_sizes.items()))

def build_adsblol_figure(
    modes_daily_sizes: Mapping[date, ModeDailySizes],
    datalinks_daily_sizes: Mapping[date, int],
    *,
    line_color: str = "#5470c6",
    datalink_color: str = "#d9a15b",
    prod_color: str = "#5a9a8b",
    staging_color: str = "#c97c76",
    mlatonly_color: str = "#7d8a8c",
    test_color: str = "#8c7bd6",
) -> Figure:
    import matplotlib.dates as mdates
    import matplotlib.pyplot as plt

    modes_days: list[Any] = list(modes_daily_sizes.keys())
    modes_prod_values = [modes_daily_sizes[day].prod for day in modes_days]
    modes_staging_values = [
        modes_daily_sizes[day].staging for day in modes_days
    ]
    modes_mlatonly_values = [
        modes_daily_sizes[day].mlatonly for day in modes_days
    ]
    modes_test_values = [modes_daily_sizes[day].test for day in modes_days]
    modes_total_values = [modes_daily_sizes[day].total for day in modes_days]
    modes_cumulative_values = list(accumulate(modes_total_values))

    dl_days: list[Any] = list(datalinks_daily_sizes.keys())
    dl_values = [datalinks_daily_sizes[day] for day in dl_days]
    dl_cumu_values = list(accumulate(dl_values))

    fig = plt.figure(figsize=(13, 8))
    grid = fig.add_gridspec(2, 1, height_ratios=[1, 1], hspace=0)
    ax_modes_top = fig.add_subplot(grid[0, 0])
    ax_modes_bottom = ax_modes_top.twinx()
    ax_dl_top = fig.add_subplot(grid[1, 0], sharex=ax_modes_top)
    ax_dl_bottom = ax_dl_top.twinx()

    ax_modes_top.plot(
        modes_days, modes_cumulative_values, color=prod_color, linewidth=2.4
    )
    ax_modes_bottom.bar(
        modes_days, modes_prod_values, color=prod_color, alpha=0.8, width=0.8
    )
    ax_modes_bottom.bar(
        modes_days,
        modes_staging_values,
        bottom=modes_prod_values,
        color=staging_color,
        alpha=0.55,
        width=0.8,
    )
    staging_bottom = [
        prod + staging
        for prod, staging in zip(modes_prod_values, modes_staging_values)
    ]
    ax_modes_bottom.bar(
        modes_days,
        modes_mlatonly_values,
        bottom=staging_bottom,
        color=mlatonly_color,
        alpha=0.5,
        width=0.8,
    )
    test_bottom = [
        bottom + mlatonly
        for bottom, mlatonly in zip(staging_bottom, modes_mlatonly_values)
    ]
    ax_modes_bottom.bar(
        modes_days,
        modes_test_values,
        bottom=test_bottom,
        color=test_color,
        alpha=0.45,
        width=0.8,
    )

    ax_dl_top.plot(dl_days, dl_cumu_values, color=line_color, linewidth=2.4)
    ax_dl_bottom.bar(
        dl_days, dl_values, color=datalink_color, alpha=0.28, width=0.8
    )

    ax_modes_top.set_ylabel("adsb cumulative raw hosted bytes")
    ax_modes_bottom.set_ylabel("daily raw hosted bytes")
    ax_modes_top.tick_params(axis="y", colors=prod_color)
    ax_modes_bottom.tick_params(axis="y")
    ax_modes_top.spines["left"].set_color(prod_color)
    ax_modes_top.set_ylim(bottom=0)
    ax_modes_bottom.set_ylim(bottom=0)
    ax_modes_top.tick_params(axis="x", labelbottom=False)

    ax_dl_top.set_ylabel(
        "datalinks cumulative raw hosted bytes", color=line_color
    )
    ax_dl_bottom.set_ylabel("daily raw hosted bytes", color=datalink_color)
    ax_dl_top.tick_params(axis="y", colors=line_color)
    ax_dl_bottom.tick_params(axis="y", colors=datalink_color)
    ax_dl_top.spines["left"].set_color(line_color)
    ax_dl_bottom.spines["right"].set_color(datalink_color)
    ax_dl_top.set_ylim(bottom=0)
    ax_dl_bottom.set_ylim(bottom=0)
    ax_dl_top.tick_params(axis="x", which="both", bottom=True, labelbottom=True)
    ax_dl_bottom.tick_params(axis="x", bottom=False, labelbottom=False)

    legend_items = [
        ("prod", prod_color, 0.01),
        ("staging", staging_color, 0.07),
        ("mlatonly", mlatonly_color, 0.18),
        ("test", test_color, 0.30),
    ]
    for text, color, x in legend_items:
        ax_modes_top.text(
            x,
            0.98,
            text,
            color=color,
            transform=ax_modes_top.transAxes,
            va="top",
        )
    ax_dl_top.text(
        0.01,
        0.98,
        "datalinks",
        color=datalink_color,
        transform=ax_dl_top.transAxes,
        va="top",
    )

    ax_dl_top.xaxis.set_major_locator(
        mdates.AutoDateLocator()  # type: ignore[no-untyped-call]
    )
    ax_dl_top.xaxis.set_major_formatter(
        mdates.DateFormatter("%Y-%m-%d")  # type: ignore[no-untyped-call]
    )
    for label in ax_dl_top.get_xticklabels():
        label.set_rotation(30)
        label.set_ha("right")  # type: ignore[attr-defined]

    fig.subplots_adjust(
        hspace=0,
        left=0.05,
        right=0.985,
        top=0.97,
        bottom=0.095,
    )
    return fig