analysis.py

"""
Project module setup.

Authors:
- Shadab Alam <md_shadab_alam@outlook.com>
- Pavlo Bazilinskyy <pavlo.bazilinskyy@gmail.com>
"""

from __future__ import annotations

import ast
import contextlib
import math
import os
import pickle
import shutil
import sys
import warnings
import multiprocessing
from concurrent.futures import ProcessPoolExecutor
from typing import Set, Optional, Dict

import polars as pl
from tqdm import tqdm

import common
from custom_logger import CustomLogger
from logmod import logs
from utils.analytics.aggregation import Aggregation
from utils.analytics.durations import Duration
from utils.analytics.events import Events
from utils.analytics.geo import Geo
from utils.analytics.io import IO
from utils.analytics.mapping_enrichment import Mapping_Enrich
from utils.analytics.metrics_cache import MetricsCache
from utils.analytics.structure import analyse_structure
from utils.analytics import remote_detections, speed_recompute
from utils.segmentation import crossing_pass as segmentation_pass
from utils.segmentation.store import INDEX_SCHEMA as SEGMENTATION_INDEX_SCHEMA
from utils.analytics.parquet_store import (
    DETECTION_FOLDER,
    configured_parquet_roots,
    sync_detection_parquet_store,
)
from utils.analytics.csv_parallel import (
    COCO_CLASSES as PARALLEL_COCO_CLASSES,
    initialise_csv_worker,
    process_csv_task,
)
from utils.core.dataset_stats import Dataset_Stats
from utils.core.metadata import MetaData
from utils.core.tools import Tools
from utils.crossing.detection import Detection
from utils.crossing.metrics import Metrics
import utils.crossing.metrics as crossing_metrics_module
from utils.plotting.bivariate import Bivariate
from utils.plotting.correlations import Correlations
from utils.plotting.crossings import Crossings, crossing_metric_label
from utils.plotting.distributions import Distributions
from utils.plotting.maps import Maps
from utils.plotting.stacked import Stacked
from utils.plotting.structure import StructurePlots

# ---------------------------------------------------------------------
# Global configuration
# ---------------------------------------------------------------------

# Suppress a specific FutureWarning emitted by plotly.
warnings.filterwarnings("ignore", category=FutureWarning, module="plotly")

logs(show_level=common.get_configs("logger_level"), show_color=True)
logger = CustomLogger(__name__)  # use custom logger

# ---------------------------------------------------------------------
# Class instances (singletons for this module)
# ---------------------------------------------------------------------

tools = Tools()
metadata = MetaData()
maps = Maps()
bivariate = Bivariate()
stacked = Stacked()
distribution = Distributions()

dataset_stats = Dataset_Stats()
metrics_cache = MetricsCache()
analytics_IO = IO()
duration = Duration()
mapping_enrich = Mapping_Enrich()
detection = Detection()
metrics = Metrics()
events = Events()
crossing = Crossings()
geo = Geo()
correlation = Correlations()
aggregation = Aggregation()
structure_plots = StructurePlots()

# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------

# File to store the locality coordinates.
file_results: str = "results.pickle"

video_paths = common.get_configs("videos")

# Populated once during Waymo preparation and reused by integrated reporting.
waymo_processed_roots: list[object] = []

# Common junk files/folders to ignore.
MISC_FILES: Set[str] = {"DS_Store", "seg", "bbox"}


class Analysis():

    def __init__(self) -> None:
        pass

    # Emoji flag mapping for ISO3 codes
    iso3_to_flag = {
        'ABW': '🇦🇼',  # Aruba
        'AFG': '🇦🇫',  # Afghanistan
        'AGO': '🇦🇴',  # Angola
        'AIA': '🇦🇮',  # Anguilla
        'ALA': '🇦🇽',  # Åland Islands
        'ALB': '🇦🇱',  # Albania
        'AND': '🇦🇩',  # Andorra
        'ARE': '🇦🇪',  # United Arab Emirates
        'ARG': '🇦🇷',  # Argentina
        'ARM': '🇦🇲',  # Armenia
        'ASM': '🇦🇸',  # American Samoa
        'ATA': '🇦🇶',  # Antarctica
        'ATF': '🏳️',  # French Southern Territories (no Unicode flag)
        'ATG': '🇦🇬',  # Antigua and Barbuda
        'AUS': '🇦🇺',  # Australia
        'AUT': '🇦🇹',  # Austria
        'AZE': '🇦🇿',  # Azerbaijan
        'BDI': '🇧🇮',  # Burundi
        'BEL': '🇧🇪',  # Belgium
        'BEN': '🇧🇯',  # Benin
        'BES': '🇧🇶',  # Bonaire, Sint Eustatius and Saba
        'BFA': '🇧🇫',  # Burkina Faso
        'BGD': '🇧🇩',  # Bangladesh
        'BGR': '🇧🇬',  # Bulgaria
        'BHR': '🇧🇭',  # Bahrain
        'BHS': '🇧🇸',  # Bahamas
        'BIH': '🇧🇦',  # Bosnia and Herzegovina
        'BLM': '🇧🇱',  # Saint Barthélemy
        'BLR': '🇧🇾',  # Belarus
        'BLZ': '🇧🇿',  # Belize
        'BMU': '🇧🇲',  # Bermuda
        'BOL': '🇧🇴',  # Bolivia
        'BRA': '🇧🇷',  # Brazil
        'BRB': '🇧🇧',  # Barbados
        'BRN': '🇧🇳',  # Brunei
        'BTN': '🇧🇹',  # Bhutan
        'BVT': '🏳️',  # Bouvet Island (no Unicode flag)
        'BWA': '🇧🇼',  # Botswana
        'CAF': '🇨🇫',  # Central African Republic
        'CAN': '🇨🇦',  # Canada
        'CCK': '🇨🇨',  # Cocos (Keeling) Islands
        'CHE': '🇨🇭',  # Switzerland
        'CHL': '🇨🇱',  # Chile
        'CHN': '🇨🇳',  # China
        'CIV': '🇨🇮',  # Côte d'Ivoire
        'CMR': '🇨🇲',  # Cameroon
        'COD': '🇨🇩',  # DR Congo
        'COG': '🇨🇬',  # Congo
        'COK': '🇨🇰',  # Cook Islands
        'COL': '🇨🇴',  # Colombia
        'COM': '🇰🇲',  # Comoros
        'CPV': '🇨🇻',  # Cape Verde
        'CRI': '🇨🇷',  # Costa Rica
        'CUB': '🇨🇺',  # Cuba
        'CUW': '🇨🇼',  # Curaçao
        'CXR': '🇨🇽',  # Christmas Island
        'CYM': '🇰🇾',  # Cayman Islands
        'CYP': '🇨🇾',  # Cyprus
        'CZE': '🇨🇿',  # Czechia
        'DEU': '🇩🇪',  # Germany
        'DJI': '🇩🇯',  # Djibouti
        'DMA': '🇩🇲',  # Dominica
        'DNK': '🇩🇰',  # Denmark
        'DOM': '🇩🇴',  # Dominican Republic
        'DZA': '🇩🇿',  # Algeria
        'ECU': '🇪🇨',  # Ecuador
        'EGY': '🇪🇬',  # Egypt
        'ERI': '🇪🇷',  # Eritrea
        'ESH': '🇪🇭',  # Western Sahara
        'ESP': '🇪🇸',  # Spain
        'EST': '🇪🇪',  # Estonia
        'ETH': '🇪🇹',  # Ethiopia
        'FIN': '🇫🇮',  # Finland
        'FJI': '🇫🇯',  # Fiji
        'FLK': '🇫🇰',  # Falkland Islands
        'FRA': '🇫🇷',  # France
        'FRO': '🇫🇴',  # Faroe Islands
        'FSM': '🇫🇲',  # Micronesia
        'GAB': '🇬🇦',  # Gabon
        'GBR': '🇬🇧',  # United Kingdom
        'GEO': '🇬🇪',  # Georgia
        'GGY': '🇬🇬',  # Guernsey
        'GHA': '🇬🇭',  # Ghana
        'GIB': '🇬🇮',  # Gibraltar
        'GIN': '🇬🇳',  # Guinea
        'GLP': '🇬🇵',  # Guadeloupe
        'GMB': '🇬🇲',  # Gambia
        'GNB': '🇬🇼',  # Guinea-Bissau
        'GNQ': '🇬🇶',  # Equatorial Guinea
        'GRC': '🇬🇷',  # Greece
        'GRD': '🇬🇩',  # Grenada
        'GRL': '🇬🇱',  # Greenland
        'GTM': '🇬🇹',  # Guatemala
        'GUF': '🇬🇫',  # French Guiana
        'GUM': '🇬🇺',  # Guam
        'GUY': '🇬🇾',  # Guyana
        'HKG': '🇭🇰',  # Hong Kong
        'HMD': '🇭🇲',  # Heard Island and McDonald Islands
        'HND': '🇭🇳',  # Honduras
        'HRV': '🇭🇷',  # Croatia
        'HTI': '🇭🇹',  # Haiti
        'HUN': '🇭🇺',  # Hungary
        'IDN': '🇮🇩',  # Indonesia
        'IMN': '🇮🇲',  # Isle of Man
        'IND': '🇮🇳',  # India
        'IOT': '🇮🇴',  # British Indian Ocean Territory
        'IRL': '🇮🇪',  # Ireland
        'IRN': '🇮🇷',  # Iran
        'IRQ': '🇮🇶',  # Iraq
        'ISL': '🇮🇸',  # Iceland
        'ISR': '🇮🇱',  # Israel
        'ITA': '🇮🇹',  # Italy
        'JAM': '🇯🇲',  # Jamaica
        'JEY': '🇯🇪',  # Jersey
        'JOR': '🇯🇴',  # Jordan
        'JPN': '🇯🇵',  # Japan
        'KAZ': '🇰🇿',  # Kazakhstan
        'KEN': '🇰🇪',  # Kenya
        'KGZ': '🇰🇬',  # Kyrgyzstan
        'KHM': '🇰🇭',  # Cambodia
        'KIR': '🇰🇮',  # Kiribati
        'KNA': '🇰🇳',  # Saint Kitts and Nevis
        'KOR': '🇰🇷',  # South Korea
        'KWT': '🇰🇼',  # Kuwait
        'LAO': '🇱🇦',  # Laos
        'LBN': '🇱🇧',  # Lebanon
        'LBR': '🇱🇷',  # Liberia
        'LBY': '🇱🇾',  # Libya
        'LCA': '🇱🇨',  # Saint Lucia
        'LIE': '🇱🇮',  # Liechtenstein
        'LKA': '🇱🇰',  # Sri Lanka
        'LSO': '🇱🇸',  # Lesotho
        'LTU': '🇱🇹',  # Lithuania
        'LUX': '🇱🇺',  # Luxembourg
        'LVA': '🇱🇻',  # Latvia
        'MAC': '🇲🇴',  # Macao
        'MAF': '🇲🇫',  # Saint Martin
        'MAR': '🇲🇦',  # Morocco
        'MCO': '🇲🇨',  # Monaco
        'MDA': '🇲🇩',  # Moldova
        'MDG': '🇲🇬',  # Madagascar
        'MDV': '🇲🇻',  # Maldives
        'MEX': '🇲🇽',  # Mexico
        'MHL': '🇲🇭',  # Marshall Islands
        'MKD': '🇲🇰',  # North Macedonia
        'MLI': '🇲🇱',  # Mali
        'MLT': '🇲🇹',  # Malta
        'MMR': '🇲🇲',  # Myanmar
        'MNE': '🇲🇪',  # Montenegro
        'MNG': '🇲🇳',  # Mongolia
        'MNP': '🇲🇵',  # Northern Mariana Islands
        'MOZ': '🇲🇿',  # Mozambique
        'MRT': '🇲🇷',  # Mauritania
        'MSR': '🇲🇸',  # Montserrat
        'MTQ': '🇲🇶',  # Martinique
        'MUS': '🇲🇺',  # Mauritius
        'MWI': '🇲🇼',  # Malawi
        'MYS': '🇲🇾',  # Malaysia
        'MYT': '🇾🇹',  # Mayotte
        'NAM': '🇳🇦',  # Namibia
        'NCL': '🇳🇨',  # New Caledonia
        'NER': '🇳🇪',  # Niger
        'NFK': '🇳🇫',  # Norfolk Island
        'NGA': '🇳🇬',  # Nigeria
        'NIC': '🇳🇮',  # Nicaragua
        'NIU': '🇳🇺',  # Niue
        'NLD': '🇳🇱',  # Netherlands
        'NOR': '🇳🇴',  # Norway
        'NPL': '🇳🇵',  # Nepal
        'NRU': '🇳🇷',  # Nauru
        'NZL': '🇳🇿',  # New Zealand
        'OMN': '🇴🇲',  # Oman
        'PAK': '🇵🇰',  # Pakistan
        'PAN': '🇵🇦',  # Panama
        'PCN': '🇵🇳',  # Pitcairn Islands
        'PER': '🇵🇪',  # Peru
        'PHL': '🇵🇭',  # Philippines
        'PLW': '🇵🇼',  # Palau
        'PNG': '🇵🇬',  # Papua New Guinea
        'POL': '🇵🇱',  # Poland
        'PRI': '🇵🇷',  # Puerto Rico
        'PRK': '🇰🇵',  # North Korea
        'PRT': '🇵🇹',  # Portugal
        'PRY': '🇵🇾',  # Paraguay
        'PSE': '🇵🇸',  # Palestine
        'PYF': '🇵🇫',  # French Polynesia
        'QAT': '🇶🇦',  # Qatar
        'REU': '🇷🇪',  # Réunion
        'ROU': '🇷🇴',  # Romania
        'RUS': '🇷🇺',  # Russia
        'RWA': '🇷🇼',  # Rwanda
        'SAU': '🇸🇦',  # Saudi Arabia
        'SDN': '🇸🇩',  # Sudan
        'SEN': '🇸🇳',  # Senegal
        'SGP': '🇸🇬',  # Singapore
        'SGS': '🏳️',  # South Georgia & South Sandwich Islands (no Unicode flag)
        'SHN': '🇸🇭',  # Saint Helena
        'SJM': '🏳️',  # Svalbard and Jan Mayen (no Unicode flag)
        'SLB': '🇸🇧',  # Solomon Islands
        'SLE': '🇸🇱',  # Sierra Leone
        'SLV': '🇸🇻',  # El Salvador
        'SMR': '🇸🇲',  # San Marino
        'SOM': '🇸🇴',  # Somalia
        'SPM': '🇵🇲',  # Saint Pierre and Miquelon
        'SRB': '🇷🇸',  # Serbia
        'SSD': '🇸🇸',  # South Sudan
        'STP': '🇸🇹',  # São Tomé and Príncipe
        'SUR': '🇸🇷',  # Suriname
        'SVK': '🇸🇰',  # Slovakia
        'SVN': '🇸🇮',  # Slovenia
        'SWE': '🇸🇪',  # Sweden
        'SWZ': '🇸🇿',  # Eswatini
        'SXM': '🇸🇽',  # Sint Maarten
        'SYC': '🇸🇨',  # Seychelles
        'SYR': '🇸🇾',  # Syria
        'TCA': '🇹🇨',  # Turks and Caicos Islands
        'TCD': '🇹🇩',  # Chad
        'TGO': '🇹🇬',  # Togo
        'THA': '🇹🇭',  # Thailand
        'TJK': '🇹🇯',  # Tajikistan
        'TKL': '🇹🇰',  # Tokelau
        'TKM': '🇹🇲',  # Turkmenistan
        'TLS': '🇹🇱',  # Timor-Leste
        'TON': '🇹🇴',  # Tonga
        'TTO': '🇹🇹',  # Trinidad and Tobago
        'TUN': '🇹🇳',  # Tunisia
        'TUR': '🇹🇷',  # Turkey
        'TUV': '🇹🇻',  # Tuvalu
        'TWN': '🇹🇼',  # Taiwan
        'TZA': '🇹🇿',  # Tanzania
        'UGA': '🇺🇬',  # Uganda
        'UKR': '🇺🇦',  # Ukraine
        'UMI': '🇺🇲',  # U.S. Minor Outlying Islands
        'URY': '🇺🇾',  # Uruguay
        'USA': '🇺🇸',  # United States
        'UZB': '🇺🇿',  # Uzbekistan
        'VAT': '🇻🇦',  # Vatican City
        'VCT': '🇻🇨',  # Saint Vincent and the Grenadines
        'VEN': '🇻🇪',  # Venezuela
        'VGB': '🇻🇬',  # British Virgin Islands
        'VIR': '🇻🇮',  # U.S. Virgin Islands
        'VNM': '🇻🇳',  # Vietnam
        'VUT': '🇻🇺',  # Vanuatu
        'WLF': '🇼🇫',  # Wallis and Futuna
        'WSM': '🇼🇸',  # Samoa
        'XKX': '🇽🇰',  # Kosovo
        'YEM': '🇾🇪',  # Yemen
        'ZAF': '🇿🇦',  # South Africa
        'ZMB': '🇿🇲',  # Zambia
        'ZWE': '🇿🇼',  # Zimbabwe
    }

    # vehicle type mapping
    vehicle_map = {
        0: "Car",
        1: "Bus",
        2: "Truck",
        3: "Two-wheeler",
        4: "Bicycle",
        5: "Automated car",
        6: "Electric scooter",
        7: "Monowheel/unicycle",
        8: "Automated bus",
        9: "Automated truck",
        10: "Automated two-wheeler",
        11: "Non-electric scooter",
        12: "Pedestrian"
    }

    # time of day mapping
    time_map = {0: "Day", 1: "Night"}


analysis_class = Analysis()

# ---------------------------------------------------------------------
# Extra rollups / summaries (continent, day-night, vehicle, max/min)
# ---------------------------------------------------------------------


def log_rollups(df_mapping: "pl.DataFrame") -> None:
    """
    Rollups + logging (paper-aligned) with COUNT-CONSISTENCY fixes.

    Key fixes vs your previous version
    ----------------------------------
    1) "Segment records" now means *segment-level label entries* (time_of_day entries),
       consistent with the paper definition:
          - time_of_day entry [0,1] contributes 2 segment records (Day + Night)
       This makes:
          - Total segment_records == Total time-of-day label entries (the "total_entries" table)
       rather than counting only videos.

    2) Unique uploads by continent uses a *canonical continent per upload* to avoid double-counting,
       so the "Total" unique uploads equals the global unique upload count:
          - pick the most frequent continent among mapped rows for that upload
          - if tied, pick the continent with the greatest total mapped duration (seconds)

    Assumptions / dependencies
    --------------------------
    - Requires: `import polars as pl`, `import ast`, and a `logger`.
    - Uses `analysis_class.vehicle_map` and `analysis_class.time_map` if present (same as your code).
      If you don't have `analysis_class`, replace those with your own dicts (vehicle_map/time_map).
    """

    # -----------------------------
    # Helpers
    # -----------------------------
    def _df_full_str(
        df: "pl.DataFrame",
        *,
        rows: int = 10_000,
        cols: int = 1_000,
        width_chars: int = 5_000,
        str_len: int = 5_000,
    ) -> str:
        """Render a Polars DataFrame as a FULL string for logging (no `...` truncation)."""
        with pl.Config(
            tbl_rows=rows,
            tbl_cols=cols,
            tbl_width_chars=width_chars,
            fmt_str_lengths=str_len,
        ):
            return df.__repr__()

    def _ensure_zero_cols(df: "pl.DataFrame", cols: list[str], dtype: "pl.DataType") -> "pl.DataFrame":
        """Add missing columns with zeros (useful after pivots when a category is absent)."""
        for c in cols:
            if c not in df.columns:
                df = df.with_columns(pl.lit(0).cast(dtype).alias(c))
        return df

    def _vstack_like(df_base: "pl.DataFrame", df_to_add: "pl.DataFrame") -> "pl.DataFrame":
        """
        Concatenate vertically, aligning df_to_add to df_base schema:
          - add missing columns as null
          - drop extra columns
          - reorder columns to match df_base
          - cast to df_base dtypes (strict=False)
        Prevents Polars schema/order mismatches.
        """
        base_cols = df_base.columns
        base_schema = df_base.schema

        for c in base_cols:
            if c not in df_to_add.columns:
                df_to_add = df_to_add.with_columns(pl.lit(None).alias(c))

        df_to_add = df_to_add.select(base_cols)

        cast_exprs = []
        for c in base_cols:
            target_dtype = base_schema.get(c)
            if target_dtype is not None:
                cast_exprs.append(pl.col(c).cast(target_dtype, strict=False).alias(c))
        if cast_exprs:
            df_to_add = df_to_add.with_columns(cast_exprs)

        return pl.concat([df_base, df_to_add], how="vertical_relaxed")

    CONTINENT_ORDER = ["Europe", "Asia", "North America", "Africa", "Oceania", "South America", "Total"]
    _continent_order_map = {name: i for i, name in enumerate(CONTINENT_ORDER)}

    def _sort_continent(df: "pl.DataFrame") -> "pl.DataFrame":
        if "continent" not in df.columns:
            return df
        return (
            df.with_columns(
                pl.col("continent")
                .map_elements(lambda x: _continent_order_map.get(str(x), 999), return_dtype=pl.Int64)
                .alias("__k")
            )
            .sort("__k")
            .drop("__k")
        )

    def _pct_2dp_str(expr: "pl.Expr") -> "pl.Expr":
        return expr.map_elements(lambda x: f"{float(x):.2f}", return_dtype=pl.Utf8)

    def _count_pct_fmt(count_expr: "pl.Expr", pct_expr: "pl.Expr") -> "pl.Expr":
        """Format as '<count> (<pct>%)' with pct at 2 decimals."""
        return pl.concat_str(
            [
                count_expr.cast(pl.Int64).cast(pl.Utf8),
                pl.lit(" ("),
                _pct_2dp_str(pct_expr.round(2)),
                pl.lit("%)"),
            ]
        )

    def _safe_eval_list(x):
        """
        Parse list-like cells robustly:
          - None -> []
          - list -> list
          - string -> ast.literal_eval when possible, else best-effort fallback
        """
        if x is None:
            return []
        if isinstance(x, list):
            return x
        if not isinstance(x, str):
            return []

        s = x.strip()
        if s == "" or s.lower() in {"nan", "none", "null"}:
            return []

        # strip wrapping quotes if the entire cell is quoted
        if (len(s) >= 2) and ((s[0] == s[-1]) and s[0] in {"'", '"'}):
            s = s[1:-1].strip()

        if s == "" or s == "[]":
            return []

        # Try python literal eval first (works for numeric nested lists)
        try:
            v = ast.literal_eval(s)
            return v if isinstance(v, list) else []
        except Exception:
            # Fallback: if looks like [a,b,c] with bare tokens (common for video ids)
            if s.startswith("[") and s.endswith("]"):
                inner = s[1:-1].strip()
                if inner == "":
                    return []
                parts = [p.strip().strip('"').strip("'") for p in inner.split(",")]
                return [p for p in parts if p != ""]
            return []

    def _safe_num_to_int(v):
        if v is None:
            return None
        try:
            return int(v)
        except Exception:
            try:
                return int(float(str(v).strip()))
            except Exception:
                return None

    def _to_list_of_ints(x) -> list[int]:
        """Scalar/list -> list[int] (dropping unparseables)."""
        if x is None:
            return []
        if isinstance(x, list):
            out = []
            for t in x:
                iv = _safe_num_to_int(t)
                if iv is not None:
                    out.append(iv)
            return out
        iv = _safe_num_to_int(x)
        return [iv] if iv is not None else []

    def _norm_list_of_lists_int(x) -> list[list[int]]:
        """
        Ensure time_of_day / similar becomes list[list[int]]:
          - [[0],[0,1]] stays same (coerced to ints)
          - [0,1] becomes [[0],[1]]
          - [] stays []
        """
        lst = _safe_eval_list(x)
        out: list[list[int]] = []
        for item in lst:
            if isinstance(item, list):
                out.append(_to_list_of_ints(item))
            else:
                out.append(_to_list_of_ints(item))
        # Drop empties at inner level (but keep structure mostly clean)
        return [a for a in out if len(a) > 0]

    def _norm_list_of_lists_float(x) -> list[list[float]]:
        """Ensure start_time/end_time becomes list[list[float]] (best effort)."""
        lst = _safe_eval_list(x)
        out: list[list[float]] = []
        for item in lst:
            if isinstance(item, list):
                arr = []
                for t in item:
                    try:
                        fv = float(t)
                        if math.isfinite(fv):
                            arr.append(fv)
                    except Exception:
                        continue
                if arr:
                    out.append(arr)
            else:
                try:
                    fv = float(item)
                    if math.isfinite(fv):
                        out.append([fv])
                except Exception:
                    continue
        return out

    def _count_label_entries(tods_lol: list[list[int]]) -> int:
        """Segment record count = number of time_of_day label entries (sum of inner lengths)."""
        if not isinstance(tods_lol, list):
            return 0
        total = 0
        for item in tods_lol:
            if isinstance(item, list):
                total += len(item)
            else:
                total += 1
        return int(total)

    def _count_segments_from_start_end(st_lol: list[list[float]], en_lol: list[list[float]]) -> int:
        """Count segments as total zipped pairs across per-video lists."""
        if not (isinstance(st_lol, list) and isinstance(en_lol, list)):
            return 0
        c = 0
        for st_i, en_i in zip(st_lol, en_lol):
            st_list = st_i if isinstance(st_i, list) else [st_i]
            en_list = en_i if isinstance(en_i, list) else [en_i]
            c += min(len(st_list), len(en_list))
        return int(c)

    def _compute_footage_time_s_from_lol(st_lol: list[list[float]], en_lol: list[list[float]]) -> int:
        """Sum of (end-start) across per-video aligned start/end lists."""
        if not (isinstance(st_lol, list) and isinstance(en_lol, list)):
            return 0
        total = 0.0
        for st_i, en_i in zip(st_lol, en_lol):
            st_list = st_i if isinstance(st_i, list) else [st_i]
            en_list = en_i if isinstance(en_i, list) else [en_i]
            for s, e in zip(st_list, en_list):
                try:
                    total += float(e) - float(s)
                except Exception:
                    continue
        return int(total)

    def _map_with_fallback(dct: dict, v):
        if v is None:
            return None
        if v in dct:
            return dct[v]
        sv = v.strip() if isinstance(v, str) else str(v).strip()
        if sv in dct:
            return dct[sv]
        try:
            iv = int(sv)
            if iv in dct:
                return dct[iv]
        except Exception:
            try:
                iv = int(float(sv))
                if iv in dct:
                    return dct[iv]
            except Exception:
                pass
        try:
            fv = float(sv)
            if fv in dct:
                return dct[fv]
        except Exception:
            pass
        return None

    # -----------------------------
    # Required columns check
    # -----------------------------
    required_cols = {"continent", "country", "locality", "iso3", "videos", "start_time", "end_time"}
    missing_cols = sorted(list(required_cols - set(df_mapping.columns)))
    if missing_cols:
        logger.warning(f"[rollups] Missing columns in df_mapping: {missing_cols}. Some rollups may be incomplete.")

    # -----------------------------
    # Parse videos robustly (string-split fallback)
    # -----------------------------
    videos_raw = pl.col("videos").cast(pl.Utf8).fill_null("").str.strip_chars().str.strip_chars("\"'")
    is_list = videos_raw.str.starts_with("[") & videos_raw.str.ends_with("]")

    videos_inner = (
        pl.when(is_list)
        .then(videos_raw.str.strip_chars("[]"))
        .otherwise(videos_raw)
        .fill_null("")
        .str.replace_all(r"[\"']", "")
        .str.strip_chars()
    )

    videos_list_expr = (
        pl.when(videos_inner == "")
        .then(pl.lit([]).cast(pl.List(pl.Utf8)))
        .otherwise(
            videos_inner
            .str.split(",")
            .list.eval(pl.element().str.strip_chars())
            .list.filter(pl.element() != "")
        )
        .alias("videos_list")
    )

    # -----------------------------
    # Parse time_of_day / start_time / end_time into list-of-lists
    # -----------------------------
    tod_lol_dtype = pl.List(pl.List(pl.Int64))
    se_lol_dtype = pl.List(pl.List(pl.Float64))

    has_tod = "time_of_day" in df_mapping.columns
    has_st = "start_time" in df_mapping.columns
    has_en = "end_time" in df_mapping.columns

    df_base = df_mapping.with_columns([videos_list_expr])

    if has_tod:
        df_base = df_base.with_columns(
            pl.col("time_of_day")
            .map_elements(_norm_list_of_lists_int, return_dtype=tod_lol_dtype)
            .alias("time_of_day_lol")
        )
    else:
        df_base = df_base.with_columns(pl.lit([]).cast(tod_lol_dtype).alias("time_of_day_lol"))

    if has_st:
        df_base = df_base.with_columns(
            pl.col("start_time")
            .map_elements(_norm_list_of_lists_float, return_dtype=se_lol_dtype)
            .alias("start_time_lol")
        )
    else:
        df_base = df_base.with_columns(pl.lit([]).cast(se_lol_dtype).alias("start_time_lol"))

    if has_en:
        df_base = df_base.with_columns(
            pl.col("end_time")
            .map_elements(_norm_list_of_lists_float, return_dtype=se_lol_dtype)
            .alias("end_time_lol")
        )
    else:
        df_base = df_base.with_columns(pl.lit([]).cast(se_lol_dtype).alias("end_time_lol"))

    # -----------------------------
    # Compute: upload_records, segment_records (COUNT-CONSISTENT), duration
    # -----------------------------
    df_base = df_base.with_columns(
        pl.col("videos_list").list.len().fill_null(0).cast(pl.Int64).alias("upload_records")
    )

    # segment_records = label-entry count from time_of_day_lol; fallback to start/end count; fallback to uploads
    df_base = df_base.with_columns(
        pl.col("time_of_day_lol")
        .map_elements(_count_label_entries, return_dtype=pl.Int64)
        .alias("segment_records_tod")
    ).with_columns(
        pl.struct(["start_time_lol", "end_time_lol"])
        .map_elements(lambda r: _count_segments_from_start_end(r["start_time_lol"],
                                                               r["end_time_lol"]), return_dtype=pl.Int64)
        .alias("segment_records_se")
    ).with_columns(
        pl.when(pl.col("segment_records_tod") > 0)
        .then(pl.col("segment_records_tod"))
        .when(pl.col("segment_records_se") > 0)
        .then(pl.col("segment_records_se"))
        .otherwise(pl.col("upload_records"))
        .alias("segment_records")
    )

    df_base = df_base.with_columns(
        pl.struct(["start_time_lol", "end_time_lol"])
        .map_elements(lambda r: _compute_footage_time_s_from_lol(r["start_time_lol"],
                                                                 r["end_time_lol"]), return_dtype=pl.Int64)
        .alias("footage_time_s")
    ).with_columns((pl.col("footage_time_s") / 3600).round(2).alias("footage_time_h"))

    # -----------------------------
    # High-level consistency diagnostics
    # -----------------------------
    total_rows = df_base.height

    # top-level alignment: videos vs time_of_day (per-video list length)
    # NOTE: time_of_day_lol is list-of-lists; its length should equal upload_records ideally.
    df_diag = df_base.select(
        [
            pl.col("upload_records"),
            pl.col("time_of_day_lol").list.len().cast(pl.Int64).alias("tod_video_slots"),
            pl.col("segment_records_tod"),
            pl.col("segment_records_se"),
        ]
    )

    n_mismatch_slots = int(
        df_diag.filter(pl.col("upload_records") != pl.col("tod_video_slots")).height
    )
    n_mismatch_seg = int(
        df_diag.filter(
            (pl.col("segment_records_tod") > 0)
            & (pl.col("segment_records_se") > 0)
            & (pl.col("segment_records_tod") != pl.col("segment_records_se"))
        ).height
    )

    total_upload_records = int(df_base.select(pl.sum("upload_records")).item() or 0)
    total_segment_records = int(df_base.select(pl.sum("segment_records")).item() or 0)
    total_dur_s = int(df_base.select(pl.sum("footage_time_s")).item() or 0)

    logger.info("\n=== [rollups] Dataset summary ===")
    logger.info(
        f"rows={total_rows} | upload_records(sum videos per row)={total_upload_records}",
    )
    logger.info(
        f"segment_records(label-entries)={total_segment_records} | duration_h={round(total_dur_s / 3600, 2)}",
    )
    if n_mismatch_slots > 0:
        logger.warning(
            f"[rollups] {n_mismatch_slots}/{total_rows} rows have len(videos_list) != len(time_of_day_lol). "
            "Those rows may have parsing/alignment issues (zips will truncate).",
        )
    if n_mismatch_seg > 0:
        logger.warning(
            f"[rollups] {n_mismatch_seg}/{total_rows} rows have segment_records_tod != segment_records_se (both present)."  # noqa: E501
            "Using time_of_day-derived counts for segment_records to match paper definition.",
        )

    # =========================================================================
    # A) Continent segment distribution (segment label-entries, shares, duration) + Total
    # =========================================================================
    cont = (
        df_base
        .filter(pl.col("continent").is_not_null() & (pl.col("continent") != ""))
        .group_by("continent")
        .agg(
            [
                pl.sum("segment_records").alias("segment_records"),
                pl.sum("footage_time_s").alias("duration_s"),
            ]
        )
    )

    tot_records = int(cont.select(pl.sum("segment_records")).item() or 0)
    tot_dur_s2 = int(cont.select(pl.sum("duration_s")).item() or 0)
    denom_r = max(tot_records, 1)
    denom_d = max(tot_dur_s2, 1)

    cont = cont.with_columns(
        [
            (pl.col("segment_records") / pl.lit(denom_r) * 100).round(2).alias("segment_share_pct"),
            (pl.col("duration_s") / pl.lit(denom_d) * 100).round(2).alias("duration_share_pct"),
            (pl.col("duration_s") / 3600).round(2).alias("duration_h"),
        ]
    )

    cont_total = pl.DataFrame(
        {
            "continent": ["Total"],
            "segment_records": [tot_records],
            "duration_s": [tot_dur_s2],
            "segment_share_pct": [100.0],
            "duration_share_pct": [100.0],
            "duration_h": [round(tot_dur_s2 / 3600, 2)],
        }
    )
    cont_table = _sort_continent(_vstack_like(cont, cont_total))

    logger.info("\n=== [rollups] A) Continent segment distribution (label-entries; paper) ===")
    logger.info(f"\n{_df_full_str(cont_table)}")

    # =========================================================================
    # Build a canonical (video_id -> continent) mapping to avoid double counting
    #
    # Note:
    # - Segment and duration totals in (A) are computed by summing mapping rows within each continent,
    #   so uploads mapped to multiple continents contribute to every continent to which they are mapped.
    # - For counts of unique uploads by continent, each upload is assigned to a single continent:
    #     1) most frequent continent among its mapped rows
    #     2) if tied, continent with greatest total mapped duration (seconds) for that upload
    # =========================================================================

    vid_cont_row_dtype = pl.List(
        pl.Struct(
            [
                pl.Field("video_id", pl.Utf8),
                pl.Field("continent", pl.Utf8),
                pl.Field("duration_s", pl.Int64),
            ]
        )
    )

    def _expand_vid_cont_dur(row: dict) -> list[dict]:
        contv = "" if row.get("continent") is None else str(row.get("continent"))
        vids = row.get("videos_list") or []
        st_lol = row.get("start_time_lol") or []
        en_lol = row.get("end_time_lol") or []

        out: list[dict] = []
        for i, vid in enumerate(vids):
            if vid is None:
                continue
            vid_s = str(vid).strip()
            if vid_s == "":
                continue

            dur = 0.0
            if i < len(st_lol) and i < len(en_lol):
                st_i = st_lol[i] if isinstance(st_lol[i], list) else [st_lol[i]]
                en_i = en_lol[i] if isinstance(en_lol[i], list) else [en_lol[i]]
                for s, e in zip(st_i, en_i):
                    try:
                        dur += float(e) - float(s)
                    except Exception:
                        continue

            out.append({"video_id": vid_s, "continent": contv, "duration_s": int(dur)})

        return out

    df_vid_rows = (
        df_base
        .select(["continent", "videos_list", "start_time_lol", "end_time_lol"])
        .filter(pl.col("continent").is_not_null() & (pl.col("continent") != ""))
        .with_columns(
            pl.struct(["continent", "videos_list", "start_time_lol", "end_time_lol"])
            .map_elements(_expand_vid_cont_dur, return_dtype=vid_cont_row_dtype)
            .alias("vid_rows")
        )
        .explode("vid_rows")
        .with_columns(
            [
                pl.col("vid_rows").struct.field("video_id").alias("video_id"),
                pl.col("vid_rows").struct.field("continent").alias("continent"),
                pl.col("vid_rows").struct.field("duration_s").alias("duration_s"),
            ]
        )
        .drop(["vid_rows", "videos_list", "start_time_lol", "end_time_lol"])
        .filter(pl.col("video_id").is_not_null() & (pl.col("video_id") != ""))
        .filter(pl.col("continent").is_not_null() & (pl.col("continent") != ""))
    )

    # For each upload and continent: row frequency + total mapped duration
    df_vid_per_cont = (
        df_vid_rows
        .group_by(["video_id", "continent"])
        .agg(
            [
                pl.len().alias("n_rows"),
                pl.sum("duration_s").alias("duration_s"),
            ]
        )
        .fill_null(0)
    )

    # Canonical continent selection:
    # - max n_rows (most frequent)
    # - tie break by max duration_s
    # - final deterministic tie break by continent name
    df_vid_canon_full = (
        df_vid_per_cont
        .sort(["video_id", "n_rows", "duration_s", "continent"], descending=[False, True, True, False])
        .group_by("video_id", maintain_order=True)
        .agg(
            [
                pl.col("continent").first().alias("continent"),
                pl.col("continent").n_unique().alias("n_continents"),
            ]
        )
    )

    n_multi_cont = int(df_vid_canon_full.filter(pl.col("n_continents") > 1).height)
    df_vid_canon = df_vid_canon_full.select(["video_id", "continent"])

    total_unique_global = int(df_vid_canon.select(pl.col("video_id").n_unique()).item() or 0)

    if n_multi_cont > 0:
        logger.warning(
            f"[rollups] {n_multi_cont} uploads appear in more than one continent in the mapping. "
            "Segment and duration rollups sum over mapping rows, so those uploads contribute to each mapped continent."  # noqa: E501
            "Unique upload counts assign each upload to one continent using most frequent mapping (ties by mapped duration).",  # noqa: E501
        )

    # =========================================================================
    # B) Unique uploads by continent (+% over global unique uploads) + Total (GLOBAL)
    # =========================================================================
    cont_unique = (
        df_vid_canon
        .group_by("continent")
        .agg(pl.col("video_id").n_unique().alias("unique_uploads"))
    )
    denom_u = max(total_unique_global, 1)

    cont_unique = cont_unique.with_columns(
        (pl.col("unique_uploads") / pl.lit(denom_u) * 100).round(2).alias("unique_uploads_pct")
    )

    cont_unique_total = pl.DataFrame(
        {"continent": ["Total"], "unique_uploads": [total_unique_global], "unique_uploads_pct": [100.0]}
    )
    cont_unique_table = _sort_continent(_vstack_like(cont_unique, cont_unique_total))

    logger.info("\n=== [rollups] B) Continent unique uploads (canonical; % of GLOBAL unique uploads) ===")
    logger.info(f"\n{_df_full_str(cont_unique_table)}")

    # =========================================================================
    # C) Vehicle distributions (unique videos and segment-level label entries)
    # =========================================================================
    if {"videos", "vehicle_type", "time_of_day"} <= set(df_mapping.columns):

        pair_dtype = pl.List(
            pl.Struct(
                [
                    pl.Field("vehicle_type", pl.Utf8),
                    pl.Field("time_of_day", pl.Utf8),
                ]
            )
        )

        video_pair_dtype = pl.List(
            pl.Struct(
                [
                    pl.Field("video_id", pl.Utf8),
                    pl.Field("vehicle_type", pl.Utf8),
                    pl.Field("time_of_day", pl.Utf8),
                ]
            )
        )

        def _expand_vehicle_tod(row: dict) -> list[dict]:
            try:
                vts = _safe_eval_list(row.get("vehicle_type"))
                tods = _safe_eval_list(row.get("time_of_day"))
                if not (isinstance(vts, list) and isinstance(tods, list)):
                    return []
            except Exception:
                return []
            out = []
            for vt, tod in zip(vts, tods):
                vt_list = vt if isinstance(vt, list) else [vt]
                tod_list = tod if isinstance(tod, list) else [tod]
                for v in vt_list:
                    for t in tod_list:
                        out.append({"vehicle_type": str(v), "time_of_day": str(t)})
            return out

        def _expand_video_vehicle_tod(row: dict) -> list[dict]:
            try:
                vids = _safe_eval_list(row.get("videos"))
                vts = _safe_eval_list(row.get("vehicle_type"))
                tods = _safe_eval_list(row.get("time_of_day"))
                if not (isinstance(vids, list) and isinstance(vts, list) and isinstance(tods, list)):
                    return []
            except Exception:
                return []

            out = []
            for vid, vt, tod in zip(vids, vts, tods):
                vid_str = "" if vid is None else str(vid).strip()
                if vid_str == "":
                    continue
                vt_list = vt if isinstance(vt, list) else [vt]
                tod_list = tod if isinstance(tod, list) else [tod]
                for v in vt_list:
                    for t in tod_list:
                        out.append(
                            {
                                "video_id": vid_str,
                                "vehicle_type": str(v),
                                "time_of_day": str(t),
                            }
                        )
            return out

        # maps (keep your existing objects)
        vehicle_map = getattr(analysis_class, "vehicle_map", {})
        time_map = getattr(analysis_class, "time_map", {})

        # -----------------------------
        # C1) Unique-video-based vehicle distributions
        # -----------------------------
        df_video_pairs = (
            df_mapping
            .select(["videos", "vehicle_type", "time_of_day"])
            .with_columns(
                pl.struct(["videos", "vehicle_type", "time_of_day"])
                .map_elements(lambda r: _expand_video_vehicle_tod(r), return_dtype=video_pair_dtype)
                .alias("pairs")
            )
            .explode("pairs")
            .with_columns(
                [
                    pl.col("pairs").struct.field("video_id").alias("video_id"),
                    pl.col("pairs").struct.field("vehicle_type").alias("vehicle_type_raw"),
                    pl.col("pairs").struct.field("time_of_day").alias("time_of_day_raw"),
                ]
            )
            .drop("pairs")
            .with_columns(
                [
                    pl.col("vehicle_type_raw")
                    .map_elements(lambda x: _map_with_fallback(vehicle_map, x), return_dtype=pl.Utf8)
                    .alias("vehicle_type_name"),
                    pl.col("time_of_day_raw")
                    .map_elements(lambda x: _map_with_fallback(time_map, x), return_dtype=pl.Utf8)
                    .alias("time_of_day_name"),
                ]
            )
            .filter(
                pl.col("video_id").is_not_null()
                & (pl.col("video_id") != "")
                & pl.col("vehicle_type_name").is_not_null()
                & pl.col("time_of_day_name").is_not_null()
            )
            .unique(["video_id", "vehicle_type_name", "time_of_day_name"])
        )

        denom_unique_videos = max(int(total_unique_global), 1)

        veh_videos = (
            df_video_pairs
            .group_by("vehicle_type_name")
            .agg(pl.col("video_id").n_unique().alias("unique_videos"))
            .with_columns(
                (pl.col("unique_videos") / pl.lit(denom_unique_videos) * 100)
                .round(2)
                .alias("pct_of_global_unique_videos")
            )
            .sort(["vehicle_type_name"])
        )

        logger.info("\n=== [rollups] C1) Vehicle distribution (unique videos) ===")
        logger.info(f"\n{_df_full_str(veh_videos)}")

        veh_video_tod = (
            df_video_pairs
            .group_by(["vehicle_type_name", "time_of_day_name"])
            .agg(pl.col("video_id").n_unique().alias("unique_videos"))
            .with_columns(
                (pl.col("unique_videos") / pl.lit(denom_unique_videos) * 100)
                .round(2)
                .alias("pct_of_global_unique_videos")
            )
            .sort(["vehicle_type_name", "time_of_day_name"])
        )

        logger.info("\n=== [rollups] C2) Vehicle x time-of-day (unique videos; deduped within video) ===")
        logger.info(f"\n{_df_full_str(veh_video_tod)}")

        # -----------------------------
        # C3) Segment-level label-entry distribution (existing paper-aligned table)
        # -----------------------------
        df_pairs = (
            df_mapping
            .select(["vehicle_type", "time_of_day"])
            .with_columns(
                pl.struct(["vehicle_type", "time_of_day"])
                .map_elements(lambda r: _expand_vehicle_tod(r), return_dtype=pair_dtype)
                .alias("pairs")
            )
            .explode("pairs")
            .with_columns(
                [
                    pl.col("pairs").struct.field("vehicle_type").alias("vehicle_type_raw"),
                    pl.col("pairs").struct.field("time_of_day").alias("time_of_day_raw"),
                ]
            )
            .drop("pairs")
        )

        df_pairs = (
            df_pairs
            .with_columns(
                [
                    pl.col("vehicle_type_raw")
                    .map_elements(lambda x: _map_with_fallback(vehicle_map, x), return_dtype=pl.Utf8)
                    .alias("vehicle_type_name"),
                    pl.col("time_of_day_raw")
                    .map_elements(lambda x: _map_with_fallback(time_map, x), return_dtype=pl.Utf8)
                    .alias("time_of_day_name"),
                ]
            )
            .filter(pl.col("vehicle_type_name").is_not_null() & pl.col("time_of_day_name").is_not_null())
        )

        veh_tod = (
            df_pairs
            .group_by(["vehicle_type_name", "time_of_day_name"])
            .len()
            .rename({"len": "count"})
        )

        tot_pairs = int(veh_tod.select(pl.sum("count")).item() or 0)
        denom_p = max(tot_pairs, 1)

        veh_tod = (
            veh_tod
            .with_columns((pl.col("count") / pl.lit(denom_p) * 100).round(2).alias("pct"))
            .sort(["vehicle_type_name", "time_of_day_name"])
        )

        logger.info("\n=== [rollups] C3) Vehicle x time-of-day (pair counts + %) ===")
        logger.info(f"\n{_df_full_str(veh_tod)}")
    # =========================================================================
    # D) Continent x time-of-day LABEL-ENTRY distribution (Day/Night entries + % within continent) + Total
    # =========================================================================
    if {"continent", "time_of_day"} <= set(df_mapping.columns):

        cont_pair_dtype = pl.List(
            pl.Struct(
                [
                    pl.Field("continent", pl.Utf8),
                    pl.Field("time_of_day", pl.Utf8),
                ]
            )
        )

        def _expand_continent_tod(row: dict) -> list[dict]:
            contv = "" if row.get("continent") is None else str(row.get("continent"))
            tods = _safe_eval_list(row.get("time_of_day"))
            out = []
            for tod in tods:
                tod_list = tod if isinstance(tod, list) else [tod]
                for t in tod_list:
                    out.append({"continent": contv, "time_of_day": str(t)})
            return out

        df_ct = (
            df_mapping
            .select(["continent", "time_of_day"])
            .with_columns(
                pl.struct(["continent", "time_of_day"])
                .map_elements(lambda r: _expand_continent_tod(r), return_dtype=cont_pair_dtype)
                .alias("pairs")
            )
            .explode("pairs")
            .with_columns(
                [
                    pl.col("pairs").struct.field("continent").alias("continent"),
                    pl.col("pairs").struct.field("time_of_day").alias("time_of_day_raw"),
                ]
            )
            .drop("pairs")
        )

        time_map = getattr(analysis_class, "time_map", {})

        df_ct = (
            df_ct
            .with_columns(
                pl.col("time_of_day_raw")
                .map_elements(lambda x: _map_with_fallback(time_map, x), return_dtype=pl.Utf8)
                .alias("time_of_day_name")
            )
            .filter(
                pl.col("continent").is_not_null()
                & (pl.col("continent") != "")
                & pl.col("time_of_day_name").is_not_null()
            )
        )

        cont_tod_long = (
            df_ct
            .group_by(["continent", "time_of_day_name"])
            .len()
            .rename({"len": "entries"})
        )

        cont_tod_wide = (
            cont_tod_long
            .pivot(index="continent", on="time_of_day_name", values="entries", aggregate_function="first")
            .fill_null(0)
        )
        cont_tod_wide = _ensure_zero_cols(cont_tod_wide, ["Day", "Night"], dtype=pl.Int64)  # type: ignore

        cont_tod_wide = cont_tod_wide.with_columns((pl.col("Day") + pl.col("Night")).alias("total_entries"))

        cont_tod_wide = cont_tod_wide.with_columns(
            [
                pl.when(pl.col("total_entries") > 0)
                .then((pl.col("Day") / pl.col("total_entries") * 100).round(2))
                .otherwise(pl.lit(0.0))
                .alias("day_pct"),
                pl.when(pl.col("total_entries") > 0)
                .then((pl.col("Night") / pl.col("total_entries") * 100).round(2))
                .otherwise(pl.lit(0.0))
                .alias("night_pct"),
            ]
        )

        tot_day = int(cont_tod_wide.select(pl.sum("Day")).item() or 0)
        tot_night = int(cont_tod_wide.select(pl.sum("Night")).item() or 0)
        tot_all = max(tot_day + tot_night, 1)

        cont_tod_total = pl.DataFrame(
            {
                "continent": ["Total"],
                "Day": [tot_day],
                "Night": [tot_night],
                "total_entries": [tot_day + tot_night],
                "day_pct": [round(tot_day / tot_all * 100, 2)],
                "night_pct": [round(tot_night / tot_all * 100, 2)],
            }
        )

        cont_tod_wide_total = _sort_continent(_vstack_like(cont_tod_wide, cont_tod_total))

        cont_tod_paper = (
            cont_tod_wide_total
            .with_columns(
                [
                    _count_pct_fmt(pl.col("Day"), pl.col("day_pct")).alias("day_entries"),
                    _count_pct_fmt(pl.col("Night"), pl.col("night_pct")).alias("night_entries"),
                ]
            )
            .select(["continent", "day_entries", "night_entries", "total_entries"])
        )

        logger.info("\n=== [rollups] D) Continent x time-of-day label entries (paper) ===")
        logger.info(f"\n{_df_full_str(cont_tod_paper)}")

        logger.info("\n=== [rollups] D) Continent x time-of-day label entries (raw wide; with pct) ===")
        logger.info(f"\n{_df_full_str(cont_tod_wide_total)}")

        # sanity check vs segment_records total (should match globally, if same mapping)
        total_entries_global = int(cont_tod_wide_total.filter(pl.col("continent") == "Total")
                                   .select("total_entries").item() or 0)
        if total_entries_global != total_segment_records:
            logger.warning(
                f"[rollups] Segment count sanity: total_entries_global={total_entries_global} != total_segment_records(df_base)={total_segment_records}."  # noqa: E501
                "This can happen if time_of_day parsing differs between df_mapping and df_base.",
            )
        else:
            logger.info(
                f"[rollups] Segment count sanity: total_entries_global matches total_segment_records ({total_segment_records}).",  # noqa: E501
            )

    # =========================================================================
    # E) Upload day/night composition (global): day-only / night-only / both (+%) + Total
    # =========================================================================
    if {"videos", "time_of_day"} <= set(df_mapping.columns):

        vid_flag_dtype = pl.List(
            pl.Struct(
                [
                    pl.Field("video_id", pl.Utf8),
                    pl.Field("has_day", pl.Boolean),
                    pl.Field("has_night", pl.Boolean),
                ]
            )
        )

        def _video_daynight_pairs(row: dict) -> list[dict]:
            vids = row.get("videos_list", [])
            tods = row.get("time_of_day_lol", [])
            out = []
            for i, vid in enumerate(vids):
                vid_str = str(vid).strip()
                if not vid_str:
                    continue
                tod = tods[i] if (isinstance(tods, list) and i < len(tods)) else []
                flags = set()
                for t in (tod if isinstance(tod, list) else [tod]):
                    iv = _safe_num_to_int(t)
                    if iv in (0, 1):
                        flags.add(iv)
                out.append({"video_id": vid_str, "has_day": (0 in flags), "has_night": (1 in flags)})
            return out

        df_video_flags = (
            df_base
            .select(["videos_list", "time_of_day_lol"])
            .with_columns(
                pl.struct(["videos_list", "time_of_day_lol"])
                .map_elements(lambda r: _video_daynight_pairs(r), return_dtype=vid_flag_dtype)
                .alias("video_flags")
            )
            .explode("video_flags")
            .with_columns(
                [
                    pl.col("video_flags").struct.field("video_id").alias("video_id"),
                    pl.col("video_flags").struct.field("has_day").alias("has_day"),
                    pl.col("video_flags").struct.field("has_night").alias("has_night"),
                ]
            )
            .drop("video_flags")
            .filter(pl.col("video_id").is_not_null() & (pl.col("video_id") != ""))
            .group_by("video_id")
            .agg(
                [
                    pl.any("has_day").alias("has_day"),  # type: ignore
                    pl.any("has_night").alias("has_night"),  # type: ignore
                ]
            )
            .with_columns(
                pl.when(pl.col("has_day") & pl.col("has_night")).then(pl.lit("both_day_night"))
                .when(pl.col("has_day")).then(pl.lit("only_day"))
                .when(pl.col("has_night")).then(pl.lit("only_night"))
                .otherwise(pl.lit("unknown"))
                .alias("daynight_category")
            )
        )

        daynight_global = (
            df_video_flags
            .group_by("daynight_category")
            .len()
            .rename({"len": "uploads"})
        )

        order_map = {"only_day": 0, "only_night": 1, "both_day_night": 2, "unknown": 3}
        daynight_global = (
            daynight_global
            .with_columns(
                pl.col("daynight_category")
                .map_elements(lambda x: order_map.get(str(x), 99), return_dtype=pl.Int64)
                .alias("__k")
            )
            .sort("__k")
            .drop("__k")
        )

        tot_uploads = int(daynight_global.select(pl.sum("uploads")).item() or 0)
        denom = max(tot_uploads, 1)
        daynight_global = daynight_global.with_columns((pl.col("uploads") / pl.lit(denom) * 100).round(2).alias("pct"))

        daynight_total = pl.DataFrame({"daynight_category": ["Total"], "uploads": [tot_uploads], "pct": [100.0]})
        daynight_global_table = _vstack_like(daynight_global, daynight_total)

        logger.info("\n=== [rollups] E) Upload day/night composition (global; paper) ===")
        logger.info(f"\n{_df_full_str(daynight_global_table)}")

    # =========================================================================
    # F) Upload day/night composition by continent (CANONICAL continent; totals consistent)
    # =========================================================================
    if {"continent", "videos", "time_of_day"} <= set(df_mapping.columns):

        # join canonical continent onto global video flags
        df_flags_canon = (
            df_video_flags
            .join(df_vid_canon, on="video_id", how="left")
            .filter(pl.col("continent").is_not_null() & (pl.col("continent") != ""))
        )

        cont_daynight_long = (
            df_flags_canon
            .group_by(["continent", "daynight_category"])
            .len()
            .rename({"len": "unique_uploads"})
        )

        cont_daynight_wide = (
            cont_daynight_long
            .pivot(index="continent", on="daynight_category", values="unique_uploads", aggregate_function="first")
            .fill_null(0)
        )
        cont_daynight_wide = _ensure_zero_cols(
            cont_daynight_wide,
            ["only_day", "only_night", "both_day_night", "unknown"], dtype=pl.Int64,  # type: ignore
        )

        cont_daynight_wide = cont_daynight_wide.with_columns(
            (pl.col("only_day") + pl.col("only_night") + pl.col("both_day_night") + pl.col("unknown")).
            alias("unique_uploads")
        )

        cont_daynight_wide = cont_daynight_wide.with_columns(
            [
                pl.when(pl.col("unique_uploads") > 0)
                .then((pl.col("only_day") / pl.col("unique_uploads") * 100).round(2))
                .otherwise(pl.lit(0.0))
                .alias("only_day_pct"),
                pl.when(pl.col("unique_uploads") > 0)
                .then((pl.col("only_night") / pl.col("unique_uploads") * 100).round(2))
                .otherwise(pl.lit(0.0))
                .alias("only_night_pct"),
                pl.when(pl.col("unique_uploads") > 0)
                .then((pl.col("both_day_night") / pl.col("unique_uploads") * 100).round(2))
                .otherwise(pl.lit(0.0))
                .alias("both_day_night_pct"),
                pl.when(pl.col("unique_uploads") > 0)
                .then((pl.col("unknown") / pl.col("unique_uploads") * 100).round(2))
                .otherwise(pl.lit(0.0))
                .alias("unknown_pct"),
            ]
        )

        # add Total row (global, consistent)
        tot_only_day = int(df_flags_canon.filter(pl.col("daynight_category") == "only_day").height)
        tot_only_night = int(df_flags_canon.filter(pl.col("daynight_category") == "only_night").height)
        tot_both = int(df_flags_canon.filter(pl.col("daynight_category") == "both_day_night").height)
        tot_unknown = int(df_flags_canon.filter(pl.col("daynight_category") == "unknown").height)
        tot_all = max(tot_only_day + tot_only_night + tot_both + tot_unknown, 1)

        cont_dn_total = pl.DataFrame(
            {
                "continent": ["Total"],
                "only_day": [tot_only_day],
                "only_night": [tot_only_night],
                "both_day_night": [tot_both],
                "unknown": [tot_unknown],
                "unique_uploads": [tot_only_day + tot_only_night + tot_both + tot_unknown],
                "only_day_pct": [round(tot_only_day / tot_all * 100, 2)],
                "only_night_pct": [round(tot_only_night / tot_all * 100, 2)],
                "both_day_night_pct": [round(tot_both / tot_all * 100, 2)],
                "unknown_pct": [round(tot_unknown / tot_all * 100, 2)],
            }
        )
        cont_daynight_wide_total = _sort_continent(_vstack_like(cont_daynight_wide, cont_dn_total))

        cont_daynight_paper = (
            cont_daynight_wide_total
            .with_columns(
                [
                    _count_pct_fmt(pl.col("only_day"), pl.col("only_day_pct")).alias("day_only"),
                    _count_pct_fmt(pl.col("only_night"), pl.col("only_night_pct")).alias("night_only"),
                    _count_pct_fmt(pl.col("both_day_night"), pl.col("both_day_night_pct")).alias("both_day_night"),
                    _count_pct_fmt(pl.col("unknown"), pl.col("unknown_pct")).alias("unknown"),
                ]
            )
            .select(["continent", "day_only", "night_only", "both_day_night", "unknown", "unique_uploads"])
        )

        logger.info("\n=== [rollups] F) Upload day/night composition by continent (canonical; paper) ===")
        logger.info(f"\n{_df_full_str(cont_daynight_paper)}")

        logger.info("\n=== [rollups] F) Upload day/night composition by continent (raw wide; with pct) ===")
        logger.info(f"\n{_df_full_str(cont_daynight_wide_total)}")

        # ensure totals consistent with global unique
        tot_u = int(cont_dn_total.select("unique_uploads").item() or 0)
        if tot_u != total_unique_global:
            logger.warning(
                f"[rollups] Canonical by-continent unique_uploads total={tot_u} != global unique videos={total_unique_global} (unexpected).",  # noqa: E501
            )
        else:
            logger.info(f"[rollups] Canonical by-continent totals match global unique videos {total_unique_global}.")

    # =========================================================================
    # G) Vehicle types per continent summary (paper table) + per-continent x time-of-day pairs
    # =========================================================================
    if {"continent", "vehicle_type", "time_of_day"} <= set(df_mapping.columns):

        cont_vehicle_pair_dtype = pl.List(
            pl.Struct(
                [
                    pl.Field("continent", pl.Utf8),
                    pl.Field("vehicle_type", pl.Utf8),
                    pl.Field("time_of_day", pl.Utf8),
                ]
            )
        )

        def _expand_continent_vehicle_tod(row: dict) -> list[dict]:
            contv = "" if row.get("continent") is None else str(row.get("continent"))
            try:
                vts = _safe_eval_list(row.get("vehicle_type"))
                tods = _safe_eval_list(row.get("time_of_day"))
                if not (isinstance(vts, list) and isinstance(tods, list)):
                    return []
            except Exception:
                return []
            out = []
            for vt, tod in zip(vts, tods):
                vt_list = vt if isinstance(vt, list) else [vt]
                tod_list = tod if isinstance(tod, list) else [tod]
                for v in vt_list:
                    for t in tod_list:
                        out.append({"continent": contv, "vehicle_type": str(v), "time_of_day": str(t)})
            return out

        df_cont_veh_pairs = (
            df_mapping
            .select(["continent", "vehicle_type", "time_of_day"])
            .with_columns(
                pl.struct(["continent", "vehicle_type", "time_of_day"])
                .map_elements(lambda r: _expand_continent_vehicle_tod(r), return_dtype=cont_vehicle_pair_dtype)
                .alias("pairs")
            )
            .explode("pairs")
            .with_columns(
                [
                    pl.col("pairs").struct.field("continent").alias("continent"),
                    pl.col("pairs").struct.field("vehicle_type").alias("vehicle_type_raw"),
                    pl.col("pairs").struct.field("time_of_day").alias("time_of_day_raw"),
                ]
            )
            .drop("pairs")
            .filter(pl.col("continent").is_not_null() & (pl.col("continent") != ""))
        )

        vehicle_map = getattr(analysis_class, "vehicle_map", {})
        time_map = getattr(analysis_class, "time_map", {})

        df_cont_veh_pairs = (
            df_cont_veh_pairs
            .with_columns(
                [
                    pl.col("vehicle_type_raw")
                    .map_elements(lambda x: _map_with_fallback(vehicle_map, x), return_dtype=pl.Utf8)
                    .alias("vehicle_type_name"),
                    pl.col("time_of_day_raw")
                    .map_elements(lambda x: _map_with_fallback(time_map, x), return_dtype=pl.Utf8)
                    .alias("time_of_day_name"),
                ]
            )
            .filter(pl.col("vehicle_type_name").is_not_null() & pl.col("time_of_day_name").is_not_null())
        )

        cont_vehicle_presence = (
            df_cont_veh_pairs
            .with_columns(
                [
                    (pl.col("time_of_day_name") == "Day").alias("has_day"),
                    (pl.col("time_of_day_name") == "Night").alias("has_night"),
                ]
            )
            .group_by(["continent", "vehicle_type_name"])
            .agg([pl.any("has_day").alias("has_day"), pl.any("has_night").alias("has_night")])  # type: ignore
            .with_columns(
                pl.when(pl.col("has_day") & pl.col("has_night")).then(pl.lit("both_day_night"))
                .when(pl.col("has_day")).then(pl.lit("only_day"))
                .when(pl.col("has_night")).then(pl.lit("only_night"))
                .otherwise(pl.lit("unknown"))
                .alias("daynight_category")
            )
        )

        vehicle_types_continent = (
            cont_vehicle_presence
            .group_by("continent")
            .agg(
                [
                    pl.col("vehicle_type_name").n_unique().alias("unique_vehicle_types"),
                    (pl.col("daynight_category") == "only_day").sum().cast(pl.Int64).alias("day_only_types"),
                    (pl.col("daynight_category") == "only_night").sum().cast(pl.Int64).alias("night_only_types"),
                    (pl.col("daynight_category") == "both_day_night").sum().cast(
                        pl.Int64).alias("both_day_night_types"),
                ]
            )
        )

        vehicle_types_paper = (
            vehicle_types_continent
            .select(["continent", "unique_vehicle_types", "day_only_types",
                     "night_only_types", "both_day_night_types"])
            .rename({"both_day_night_types": "types_in_both_day_night"})
        )
        vehicle_types_paper = _sort_continent(vehicle_types_paper)

        logger.info("\n=== [rollups] G) Vehicle types per continent (paper) ===")
        logger.info(f"\n{_df_full_str(vehicle_types_paper)}")

        cont_veh_tod = (
            df_cont_veh_pairs
            .group_by(["continent", "vehicle_type_name", "time_of_day_name"])
            .len()
            .rename({"len": "count"})
            .with_columns((pl.col("count") / pl.sum("count").
                           over("continent") * 100).round(2).alias("pct_within_continent"))
        )
        cont_veh_tod = _sort_continent(cont_veh_tod)

        logger.info("\n=== [rollups] G) Vehicle types per continent x time-of-day (pair counts + % within continent) ===")  # noqa: E501
        logger.info(f"\n{_df_full_str(cont_veh_tod)}")

    # =========================================================================
    # H) Coverage extremes: max/min locality and country by uploads, segments, and duration
    # =========================================================================
    locality_rank = (
        df_base
        .filter(pl.col("locality").is_not_null() & (pl.col("locality") != ""))
        .group_by(["continent", "country", "locality", "iso3"])
        .agg(
            [
                pl.sum("upload_records").alias("upload_count"),
                pl.sum("segment_records").alias("segment_count"),
                pl.sum("footage_time_s").alias("footage_time_s"),
            ]
        )
        .with_columns((pl.col("footage_time_s") / 3600).round(2).alias("footage_time_h"))
    )

    # =========================================================================
    # H1) Max locality by duration within each continent
    # =========================================================================
    max_locality_per_continent_all_ties = (
        locality_rank
        .filter(pl.col("footage_time_s") > 0)
        .join(
            locality_rank
            .group_by("continent")
            .agg(pl.max("footage_time_s").alias("max_footage_time_s")),
            on="continent",
            how="inner",
        )
        .filter(pl.col("footage_time_s") == pl.col("max_footage_time_s"))
        .select(
            [
                "continent",
                "country",
                "locality",
                "iso3",
                "upload_count",
                "segment_count",
                "footage_time_s",
                "footage_time_h",
            ]
        )
        .sort(
            ["continent", "footage_time_s", "segment_count", "upload_count", "locality"],
            descending=[False, True, True, True, False],
        )
    )

    logger.info("\n=== [rollups] H1) Max locality by duration within each continent (including ties) ===")
    logger.info(f"\n{_df_full_str(max_locality_per_continent_all_ties)}")

    # Exactly one locality per continent (tie break by segment_count, then upload_count, then locality name)
    max_locality_per_continent_one = (
        max_locality_per_continent_all_ties
        .unique(subset=["continent"], keep="first")
        .sort("continent")
    )

    logger.info("\n=== [rollups] H1) Max locality by duration within each continent (one per continent) ===")
    logger.info(f"\n{_df_full_str(max_locality_per_continent_one)}")

    country_rank = (
        df_base
        .filter(pl.col("country").is_not_null() & (pl.col("country") != ""))
        .group_by(["continent", "country", "iso3"])
        .agg(
            [
                pl.sum("upload_records").alias("upload_count"),
                pl.sum("segment_records").alias("segment_count"),
                pl.sum("footage_time_s").alias("footage_time_s"),
            ]
        )
        .with_columns((pl.col("footage_time_s") / 3600).round(2).alias("footage_time_h"))
    )

    def _top_bottom(df: "pl.DataFrame", col: str):
        top = df.sort(col, descending=True).head(1)
        bot = df.filter(pl.col(col) > 0).sort(col).head(1)
        return top, bot

    top_locality_u, bot_locality_u = _top_bottom(locality_rank, "upload_count")
    top_locality_s, bot_locality_s = _top_bottom(locality_rank, "segment_count")
    top_locality_t, bot_locality_t = _top_bottom(locality_rank, "footage_time_s")

    top_ctry_u, bot_ctry_u = _top_bottom(country_rank, "upload_count")
    top_ctry_s, bot_ctry_s = _top_bottom(country_rank, "segment_count")
    top_ctry_t, bot_ctry_t = _top_bottom(country_rank, "footage_time_s")

    logger.info("\n=== [rollups] H) Max/Min locality by upload_count ===")
    logger.info(f"\nMAX:\n{_df_full_str(top_locality_u)}\nMIN (non-zero):\n{_df_full_str(bot_locality_u)}")

    logger.info("\n=== [rollups] H) Max/Min locality by segment_count ===")
    logger.info(f"\nMAX:\n{_df_full_str(top_locality_s)}\nMIN (non-zero):\n{_df_full_str(bot_locality_s)}")

    logger.info("\n=== [rollups] H) Max/Min locality by duration ===")
    logger.info(f"\nMAX:\n{_df_full_str(top_locality_t)}\nMIN (non-zero):\n{_df_full_str(bot_locality_t)}")

    logger.info("\n=== [rollups] H) Max/Min COUNTRY by upload_count ===")
    logger.info(f"\nMAX:\n{_df_full_str(top_ctry_u)}\nMIN (non-zero):\n{_df_full_str(bot_ctry_u)}")

    logger.info("\n=== [rollups] H) Max/Min COUNTRY by segment_count ===")
    logger.info(f"\nMAX:\n{_df_full_str(top_ctry_s)}\nMIN (non-zero):\n{_df_full_str(bot_ctry_s)}")

    logger.info("\n=== [rollups] H) Max/Min COUNTRY by duration ===")
    logger.info(f"\nMAX:\n{_df_full_str(top_ctry_t)}\nMIN (non-zero):\n{_df_full_str(bot_ctry_t)}")


def _environment_truthy(name: str, default: bool = False) -> bool:
    """Read a boolean environment variable without adding command parsing."""
    value = os.environ.get(name)
    if value is None:
        return default
    text = str(value).strip().lower()
    if text in {"1", "true", "yes", "y", "on"}:
        return True
    if text in {"0", "false", "no", "n", "off", ""}:
        return False
    return default


def _normalise_city_limit(value: object) -> int:
    """Return the configured city limit, treating an empty value as unlimited."""
    if value is None:
        return 0
    if isinstance(value, str) and not value.strip():
        return 0
    if isinstance(value, bool):
        raise ValueError("n_cities must be an integer, an empty string, or null")
    try:
        numeric = float(value)
    except (TypeError, ValueError) as error:
        raise ValueError(
            "n_cities must be an integer, an empty string, or null"
        ) from error
    if not math.isfinite(numeric) or not numeric.is_integer():
        raise ValueError("n_cities must be a finite integer")
    return int(numeric)


# ---------------------------------------------------------------------
# Analysis results cache metadata
# ---------------------------------------------------------------------

CACHE_CONFIG_KEYS: tuple[str, ...] = (
    "countries_analyse",
    "n_cities",
    "max_footage_hours_per_city",
    "processing_fps",
    "vehicles_analyse",
    "crossing_rule",
    "target_crossings_per_city",
    # Which segments count as available depends on it, so it changes the draw.
    "fetch_detections_on_demand",
    # Segmentation settings change the derived crossing metrics, so a cached
    # run must not silently reuse values computed under different ones.
    "use_segmentation",
    "segmentation_model",
    "segmentation_coarse_hz",
    "segmentation_refine_hz",
    "segmentation_min_confidence",
    "segmentation_is_primary",
)
CACHE_RESULTS_COUNT = 43
# Identifies how max_footage_hours_per_city picks segments; bump when it changes.
FOOTAGE_SELECTION_RULE = "seeded_random_available_v1"
# Zero hesitation times are left out of the segmentation averages.
SEGMENTATION_METRICS_RULE = "hesitation_excludes_zero_v1"
# The box-based initiation time is measured from the frames its checks span.
HESITATION_RULE = "frame_timed_checks_v1"
CACHE_METADATA_VERSION = 1


# Settings that only matter once the segmentation pass actually runs.
SEGMENTATION_CACHE_KEYS: tuple[str, ...] = (
    "segmentation_model",
    "segmentation_coarse_hz",
    "segmentation_refine_hz",
    "segmentation_min_confidence",
    "segmentation_is_primary",
)


def _current_cache_config() -> Dict[str, object]:
    # Return the configuration values that determine results.pickle reuse.
    #
    # When segmentation is switched off none of its settings affect a single
    # reported value, so they are left out of the fingerprint entirely. Were
    # they included, merely pulling a version that defines them would discard
    # an existing results.pickle and force a full reanalysis for no change in
    # output. use_segmentation itself always counts, because turning it on or
    # off does change what is produced.
    segmentation_enabled = bool(common.get_configs("use_segmentation"))
    skipped = set(SEGMENTATION_CACHE_KEYS)
    if not segmentation_enabled:
        # A run with segmentation off produces exactly what a run from before
        # the feature existed produced, so its fingerprint is the same one and
        # an existing results.pickle stays valid.
        skipped.add("use_segmentation")
    config = {
        key: common.get_configs(key)
        for key in CACHE_CONFIG_KEYS
        if segmentation_enabled or key not in skipped
    }
    config["hesitation_rule"] = HESITATION_RULE
    from utils.crossing import track_joining

    if track_joining.TRACK_JOINING_ENABLED:
        config["track_joining"] = track_joining.TRACK_JOINING_VERSION
    if common.get_configs("crossing_rule") == "road_crossing":
        from utils.crossing.road_crossing import ROAD_CROSSING_RULE_VERSION

        config["road_crossing_rule_version"] = ROAD_CROSSING_RULE_VERSION
    cities = common.get_configs("cities_analyse")
    if cities:
        # Left out when empty, so a results.pickle from before the setting
        # existed still matches a run that analyses every city.
        config["cities_analyse"] = list(cities)
    if segmentation_enabled:
        # The surface-label schema changes whenever stored labels stop being
        # valid (for example the move to the probed video frame rate), so a
        # cached run must not keep segmentation metrics derived from old ones.
        config["segmentation_index_schema"] = SEGMENTATION_INDEX_SCHEMA
        # Bumped when the per-track segmentation metrics change meaning, so a
        # cached run does not keep values computed under the old definition.
        config["segmentation_metrics_rule"] = SEGMENTATION_METRICS_RULE
    if MetaData._normalise_max_footage_seconds(
        common.get_configs("max_footage_hours_per_city")
    ) is not None or common.get_configs("target_crossings_per_city") is not None:
        # With a cap the seed decides which segments fill each city's budget,
        # and the selection rule itself changed from mapping order to a seeded
        # random draw over segments that have a detection file. Both belong
        # in the fingerprint, so neither a new seed nor a pickle from the old
        # rule is mistaken for a match. Uncapped runs index every segment and
        # keep their existing fingerprint.
        config["footage_sampling_seed"] = MetaData._normalise_sampling_seed(
            common.get_configs("footage_sampling_seed")
        )
        config["footage_selection"] = FOOTAGE_SELECTION_RULE
    return config


def _build_cache_metadata(
    cache_config: Dict[str, object],
    segmentation: Optional[Dict[str, object]] = None,
) -> Dict[str, object]:
    # Build metadata appended to results.pickle after the 43 result values.
    # The speed unit records what the cached values actually are, so a later
    # cached run labels its figures from the data rather than from whichever
    # Waymo model happens to be installed at that moment.
    #
    # The segmentation aggregates travel here rather than in the 43-value
    # tuple so that pickles written before this feature existed still load.
    metadata: Dict[str, object] = {
        "__results_cache_metadata__": CACHE_METADATA_VERSION,
        "config": dict(cache_config),
        "speed_unit": os.environ.get("CROWD_CROSSING_SPEED_UNIT", "relative"),
    }
    if segmentation:
        metadata["segmentation"] = segmentation
    return metadata


def _segmentation_from_payload(payload: object) -> Dict[str, object]:
    # Return the cached segmentation aggregates, or an empty mapping.
    if not isinstance(payload, (tuple, list)) or len(payload) <= CACHE_RESULTS_COUNT:
        return {}
    metadata_value = payload[CACHE_RESULTS_COUNT]
    if not isinstance(metadata_value, dict):
        return {}
    segmentation = metadata_value.get("segmentation")
    return dict(segmentation) if isinstance(segmentation, dict) else {}


def _speed_unit_from_payload(payload: object) -> Optional[str]:
    # Return the crossing speed unit recorded alongside the cached results.
    if not isinstance(payload, (tuple, list)):
        return None
    if len(payload) <= CACHE_RESULTS_COUNT:
        return None
    metadata_value = payload[CACHE_RESULTS_COUNT]
    if not isinstance(metadata_value, dict):
        return None
    unit = metadata_value.get("speed_unit")
    return str(unit) if unit else None


def _apply_cached_speed_unit(unit: Optional[str]) -> None:
    # Make figure labels describe the cached values, not the installed model.
    model_unit = os.environ.get("CROWD_CROSSING_SPEED_UNIT", "relative")
    if unit is None:
        # Pickles written before the unit was recorded predate the metric speed
        # rollout, so their speed values are the dimensionless relative index.
        unit = "relative"
        logger.warning(
            "The cached results record no crossing speed unit; assuming the "
            "relative index. Recompute the speeds to report metres per second."
        )
    if unit != model_unit:
        logger.warning(
            "Cached crossing speed values are in '{}' but the installed Waymo "
            "model reports '{}'. Labelling figures as '{}' to match the data. "
            "Set reanalyse_speed to true and rerun analysis.py to recompute "
            "crossing speeds in '{}'.",
            unit, model_unit, unit, model_unit,
        )
    os.environ["CROWD_CROSSING_SPEED_UNIT"] = unit


def _cache_config_from_payload(payload: object) -> Optional[Dict[str, object]]:
    # Extract cache configuration from a new-format results payload.
    if not isinstance(payload, (tuple, list)):
        return None
    if len(payload) <= CACHE_RESULTS_COUNT:
        return None

    metadata_value = payload[CACHE_RESULTS_COUNT]
    if not isinstance(metadata_value, dict):
        return None
    if metadata_value.get("__results_cache_metadata__") != CACHE_METADATA_VERSION:
        return None

    config_value = metadata_value.get("config")
    if not isinstance(config_value, dict):
        return None

    return dict(config_value)


def _load_results_cache(
    path: str,
) -> tuple[
    Optional[tuple],
    Optional[Dict[str, object]],
    Optional[str],
    Dict[str, object],
]:
    # Load the 43 result values plus optional cache metadata, speed unit and
    # segmentation aggregates.
    try:
        with open(path, "rb") as file:
            payload = pickle.load(file)
    except Exception as error:
        logger.warning(
            f"Could not read cached analysis results from {path}: {error}. "
            "The analysis will be recomputed."
        )
        return None, None, None, {}

    if not isinstance(payload, (tuple, list)) or len(payload) < CACHE_RESULTS_COUNT:
        logger.warning(
            f"Cached analysis results in {path} have an unexpected format. "
            "The analysis will be recomputed."
        )
        return None, None, None, {}

    results = tuple(payload[:CACHE_RESULTS_COUNT])
    cached_config = _cache_config_from_payload(payload)
    # The recorded unit is returned rather than applied here: it describes the
    # cached values only, so it must not relabel a full reanalysis that is
    # about to recompute the speeds from scratch.
    cached_speed_unit = _speed_unit_from_payload(payload)
    cached_segmentation = _segmentation_from_payload(payload)
    return results, cached_config, cached_speed_unit, cached_segmentation


def _cache_config_differences(
    cached_config: Dict[str, object],
    current_config: Dict[str, object],
) -> Dict[str, tuple[object, object]]:
    # Return changed settings as cached/current value pairs.
    keys = sorted(set(cached_config) | set(current_config))
    return {
        key: (cached_config.get(key), current_config.get(key))
        for key in keys
        if cached_config.get(key) != current_config.get(key)
    }


def _segmentation_is_primary() -> bool:
    """Return whether the figures should report the segmentation metrics."""
    return bool(common.get_configs("segmentation_is_primary")) and bool(
        common.get_configs("use_segmentation")
    )


def _primary_speed(
    segmentation: Dict[str, object],
    baseline_locality: dict,
    baseline_country: dict,
    baseline_all_locality: dict,
    baseline_all_country: dict,
) -> tuple:
    """Substitute the road-restricted speed when it is the reported metric.

    The reanalysis blocks further down re-derive the crossing averages from
    the baseline per-track values, so the substitution cannot happen once and
    be left alone: it has to be reapplied wherever the reported averages are
    set, or a run with reanalyse_speed enabled would silently fall back to the
    baseline after the switch had been honoured earlier.
    """
    if not _segmentation_is_primary() or not segmentation:
        return (
            baseline_locality,
            baseline_country,
            baseline_all_locality,
            baseline_all_country,
        )
    locality = segmentation.get("avg_speed_locality") or {}
    if not locality:
        # Substituting nothing would blank the crossing speed in every figure.
        # This happens whenever the frozen Waymo model is absent, because the
        # road-restricted speed is only defined when that model qualifies, so
        # it must not be mistaken for a legitimate empty result.
        logger.error(
            "segmentation_is_primary is enabled but the segmentation pass "
            "produced no crossing speeds, so the reported speed would be "
            "empty. Keeping the baseline speed. Check that the frozen Waymo "
            "model is present, since the road-restricted speed is only "
            "defined when it qualifies."
        )
        return (
            baseline_locality,
            baseline_country,
            baseline_all_locality,
            baseline_all_country,
        )
    return (
        locality,
        segmentation.get("avg_speed_country") or {},
        segmentation.get("all_speed_locality") or {},
        segmentation.get("all_speed_country") or {},
    )


def _primary_time(
    segmentation: Dict[str, object],
    baseline_locality: dict,
    baseline_country: dict,
    baseline_all_locality: dict,
    baseline_all_country: dict,
) -> tuple:
    """Substitute the kerb-anchored hesitation time when it is the reported metric."""
    if not _segmentation_is_primary() or not segmentation:
        return (
            baseline_locality,
            baseline_country,
            baseline_all_locality,
            baseline_all_country,
        )
    locality = segmentation.get("avg_time_locality") or {}
    if not locality:
        logger.error(
            "segmentation_is_primary is enabled but the segmentation pass "
            "produced no hesitation times, so the reported initiation time "
            "would be empty. Keeping the baseline times."
        )
        return (
            baseline_locality,
            baseline_country,
            baseline_all_locality,
            baseline_all_country,
        )
    return (
        locality,
        segmentation.get("avg_time_country") or {},
        segmentation.get("all_time_locality") or {},
        segmentation.get("all_time_country") or {},
    )


BBOX_SPEED_PREFIX = "speed_crossing_bbox"
BBOX_TIME_PREFIX = "time_crossing_bbox"


def _write_bbox_columns(
    df_mapping_source: "pl.DataFrame",
    avg_speed_locality: Optional[dict],
    avg_speed_country: Optional[dict],
    avg_time_locality: Optional[dict],
    avg_time_country: Optional[dict],
    pedestrian_cross_locality_source: dict,
    pedestrian_cross_country_source: dict,
) -> "pl.DataFrame":
    """Write the bounding-box speed and time into their own columns.

    ``speed_crossing_*`` and ``time_crossing_*`` hold the reported values,
    which every figure reads; with segmentation_is_primary they are the
    segmentation values. These ``*_bbox_*`` columns always hold the
    bounding-box derivation, so it is never lost for comparison. A metric
    passed as None leaves its columns untouched, and the columns being
    rewritten are cleared first so no stale value survives.
    """
    prefixes = []
    if avg_speed_locality is not None:
        prefixes.append(BBOX_SPEED_PREFIX)
    if avg_time_locality is not None:
        prefixes.append(BBOX_TIME_PREFIX)
    stale = [column for column in df_mapping_source.columns if column.startswith(tuple(prefixes))] if prefixes else []
    if stale:
        df_mapping_source = df_mapping_source.with_columns(
            [pl.lit(None).cast(pl.Float64).alias(column) for column in stale]
        )
    return mapping_enrich.add_speed_and_time_to_mapping(
        df_mapping=df_mapping_source,
        avg_speed_locality=avg_speed_locality,
        avg_speed_country=avg_speed_country,
        avg_time_locality=avg_time_locality,
        avg_time_country=avg_time_country,
        pedestrian_cross_locality=pedestrian_cross_locality_source,
        pedestrian_cross_country=pedestrian_cross_country_source,
        speed_col_prefix=BBOX_SPEED_PREFIX,
        time_col_prefix=BBOX_TIME_PREFIX,
    )


def _apply_segmentation_columns(
    df_mapping_source: "pl.DataFrame",
    segmentation: Dict[str, object],
    pedestrian_cross_locality_source: dict,
    pedestrian_cross_country_source: dict,
) -> "pl.DataFrame":
    """Write the segmentation-derived metrics into their own columns.

    These ``*_seg_*`` columns carry the two quantities measured against the
    segmented carriageway, next to the ``*_bbox_*`` columns with the
    bounding-box derivation (see _write_bbox_columns), so the two can be
    compared city by city whichever one ``speed_crossing_*`` reports.
    """
    if not segmentation:
        return df_mapping_source

    updated = mapping_enrich.add_speed_and_time_to_mapping(
        df_mapping=df_mapping_source,
        avg_speed_locality=segmentation.get("avg_speed_locality") or {},
        avg_speed_country=segmentation.get("avg_speed_country") or {},
        avg_time_locality=segmentation.get("avg_time_locality") or {},
        avg_time_country=segmentation.get("avg_time_country") or {},
        pedestrian_cross_locality=pedestrian_cross_locality_source,
        pedestrian_cross_country=pedestrian_cross_country_source,
        speed_col_prefix="speed_crossing_seg",
        time_col_prefix="time_crossing_seg",
    )

    populated = [
        column
        for column in updated.columns
        if column.startswith(("speed_crossing_seg", "time_crossing_seg"))
        and updated.select(pl.col(column).is_not_null().sum()).item() > 0
    ]
    logger.info(
        f"Segmentation columns populated: {', '.join(sorted(populated)) or 'none'}."
    )
    return updated


def _segment_duration_seconds(start_value: object, end_value: object) -> float:
    """Sum aligned segment durations from scalar or nested mapping values."""
    def parse(value: object) -> object:
        if not isinstance(value, str):
            return value
        text = value.strip()
        if not text or text.lower() in {"nan", "none", "null"}:
            return []
        try:
            return ast.literal_eval(text)
        except (SyntaxError, ValueError):
            return []

    def total(starts: object, ends: object) -> float:
        if isinstance(starts, (list, tuple)) and isinstance(ends, (list, tuple)):
            return sum(
                (total(start, end) for start, end in zip(starts, ends)),
                0.0,
            )
        try:
            start = float(starts)
            end = float(ends)
        except (TypeError, ValueError):
            return 0.0
        if not math.isfinite(start) or not math.isfinite(end) or end < start:
            return 0.0
        return end - start

    return float(total(parse(start_value), parse(end_value)))


def _select_cities_by_footage(
    df_mapping_source: "pl.DataFrame",
    configured_limit: object,
) -> "pl.DataFrame":
    """Select the highest or lowest footage cities using total segment duration."""
    limit = _normalise_city_limit(configured_limit)
    if limit == 0 or df_mapping_source.height == 0:
        logger.info("City footage limit is empty or zero; analysing all available cities.")
        return df_mapping_source

    required = {"locality", "start_time", "end_time"}
    missing = sorted(required - set(df_mapping_source.columns))
    if missing:
        raise ValueError(
            "Cannot apply n_cities because the mapping is missing: "
            + ", ".join(missing)
        )

    identity_columns = [
        column
        for column in ("locality", "state", "iso3", "country")
        if column in df_mapping_source.columns
    ]
    city_key = pl.concat_str(
        [
            pl.col(column).cast(pl.Utf8, strict=False).fill_null("")
            for column in identity_columns
        ],
        separator="\u241f",
    )
    footage_seconds = [
        float(_segment_duration_seconds(start_value, end_value))
        for start_value, end_value in df_mapping_source.select(
            ["start_time", "end_time"]
        ).iter_rows()
    ]
    ranked = (
        df_mapping_source
        .with_row_index("__city_row_order")
        .with_columns([
            city_key.alias("__city_key"),
            pl.Series(
                "__city_footage_seconds",
                footage_seconds,
                dtype=pl.Float64,
            ),
        ])
    )
    all_city_totals = (
        ranked
        .group_by("__city_key")
        .agg([
            pl.col("__city_footage_seconds").sum().alias("footage_seconds"),
            pl.col("locality").first().alias("locality"),
            *(
                [pl.col("state").first().alias("state")]
                if "state" in ranked.columns
                else []
            ),
            *(
                [pl.col("country").first().alias("country")]
                if "country" in ranked.columns
                else []
            ),
        ])
        .sort(
            ["footage_seconds", "__city_key"],
            descending=[limit > 0, False],
        )
    )
    city_totals = all_city_totals.head(abs(limit))
    selected_keys = city_totals.get_column("__city_key")
    selected = (
        ranked
        .filter(pl.col("__city_key").is_in(selected_keys))
        .sort("__city_row_order")
        .drop(["__city_row_order", "__city_key", "__city_footage_seconds"])
    )
    direction = "highest" if limit > 0 else "lowest"
    logger.info(
        f"Selected {city_totals.height} cities with the {direction} total segment duration "
        f"from {all_city_totals.height} available cities."
    )
    for row in city_totals.iter_rows(named=True):
        location = ", ".join(
            str(row.get(field))
            for field in ("locality", "state", "country")
            if row.get(field) not in (None, "")
        )
        logger.info(
            f"Selected city: {location}; footage="
            f"{float(row['footage_seconds']):.2f} seconds."
        )
    return selected


def _build_integrated_speed_sources(
    df_mapping_source: "pl.DataFrame",
    crossing_tracks: Dict[str, object],
) -> list[dict[str, object]]:
    """Map processed CROWD bbox Parquet files to city and crossing IDs."""
    parquet_by_source: Dict[str, str] = {}
    for parquet_root in configured_parquet_roots():
        bbox_root = os.path.join(os.fspath(parquet_root), "bbox")
        if not os.path.isdir(bbox_root):
            continue
        for file_name in os.listdir(bbox_root):
            if not file_name.lower().endswith(".parquet"):
                continue
            source_id = os.path.splitext(file_name)[0]
            parquet_by_source[source_id] = os.path.join(bbox_root, file_name)

    aspect_ratio_text = os.environ.get("CROWD_BBOX_ASPECT_RATIO", str(16.0 / 9.0))
    try:
        aspect_ratio = float(aspect_ratio_text)
    except (TypeError, ValueError):
        aspect_ratio = 16.0 / 9.0
    if not math.isfinite(aspect_ratio) or aspect_ratio <= 0.0:
        aspect_ratio = 16.0 / 9.0

    segment_lookup = metadata.segment_lookup(df_mapping_source)
    location_by_id: Dict[int, dict] = {
        int(row["id"]): row
        for row in df_mapping_source.iter_rows(named=True)
        if row.get("id") is not None
    }

    output: list[dict[str, object]] = []
    for source_id, payload in crossing_tracks.items():
        bbox_parquet = parquet_by_source.get(str(source_id))
        if bbox_parquet is None:
            logger.warning(
                f"Crossing motion report cannot find the bbox Parquet file for {source_id}; skipping."
            )
            continue
        try:
            video_id, start_text, fps_text = str(source_id).rsplit("_", 2)
            start_index = int(float(start_text))
            fps = float(fps_text)
        except (TypeError, ValueError):
            logger.warning(
                f"Crossing motion report expected video_start_fps but received {source_id}; skipping."
            )
            continue
        segment_meta = segment_lookup.get((video_id, start_index))
        locality_id = segment_meta[0] if segment_meta is not None else None
        if locality_id is None:
            logger.warning(
                f"Crossing motion report cannot map {source_id} to a CROWD city; skipping."
            )
            continue
        row = location_by_id.get(int(locality_id))
        if row is None:
            continue
        ids: object = []
        if isinstance(payload, dict):
            ids = payload.get("ids", [])
        if ids is None:
            ids = []
        elif not isinstance(ids, (list, tuple, set)):
            ids = [ids]
        output.append(
            {
                "source_id": str(source_id),
                # The reporting API retains the historical key name
                # ``bbox_csv``, but the path is a Parquet file. A loader
                # adapter installed below handles it transparently.
                "bbox_csv": bbox_parquet,
                "fps": fps,
                "aspect_ratio": aspect_ratio,
                "locality": row.get("locality") or "Unknown",
                "state": row.get("state") or "",
                "country": row.get("country") or "Unknown",
                "crossing_track_ids": list(ids),
            }
        )
    return output


def _install_parquet_bbox_loader() -> None:
    """Teach the legacy crossing report loader to consume Parquet paths."""
    if getattr(crossing_metrics_module, "_CROWD_PARQUET_LOADER_INSTALLED", False):
        return

    original_loader = crossing_metrics_module.load_bbox_csv

    def load_bbox_source(path):
        source_path = os.fspath(path)
        if os.path.splitext(source_path)[1].lower() == ".parquet":
            dataframe = pl.read_parquet(source_path)
            return crossing_metrics_module.bbox_rows_from_polars(dataframe)
        return original_loader(path)

    crossing_metrics_module.load_bbox_csv = load_bbox_source
    crossing_metrics_module._CROWD_PARQUET_LOADER_INSTALLED = True


def _run_integrated_speed_reporting(
    df_mapping_source: "pl.DataFrame",
    crossing_tracks: Dict[str, object],
) -> None:
    """Create CROWD city reports and optional processed Waymo figures."""
    if not _environment_truthy("CROWD_SPEED_REPORT", True):
        logger.info("Integrated crossing speed report disabled by CROWD_SPEED_REPORT.")
        return
    try:
        _install_parquet_bbox_loader()
        from utils.crossing.metrics import run_integrated_speed_report

        output_root = os.environ.get(
            "CROWD_SPEED_REPORT_OUTPUT",
            os.path.join(common.output_dir, "crossing_speed"),
        )
        summary = run_integrated_speed_report(
            crowd_sources=_build_integrated_speed_sources(
                df_mapping_source,
                crossing_tracks,
            ),
            repository_root=common.root_dir,
            output_root=output_root,
            waymo_roots=waymo_processed_roots or None,
            force=_environment_truthy("CROWD_SPEED_REPORT_FORCE", False),
            log=lambda message: logger.info(message),
        )
        crowd_summary = summary.get("crowd", {})
        logger.info(
            "CROWD crossing speed report: "
            f"cities={crowd_summary.get('cities_analysed', 0)}, "
            f"crossings={crowd_summary.get('detected_crossing_tracks', 0)}, "
            f"valid={crowd_summary.get('valid_relative_motion_tracks', 0)}."
        )
        if summary.get("waymo", {}).get("status") == "skipped":
            logger.info("Waymo report skipped because processed Waymo files were not found.")
    except Exception as error:
        logger.error(f"Integrated crossing speed reporting failed: {error}")


CROSSING_RULES = ("detector", "road_crossing")


def _parse_city_entry(entry: object) -> tuple:
    """Split a cities_analyse entry into (locality, state, iso3), lower-cased.

    An entry is "City", "City, ISO3" or "City, State, ISO3". State and ISO3
    are None when not given, and then match any value.
    """
    if not isinstance(entry, str) or not entry.strip():
        logger.error(f"cities_analyse entries must be non-empty strings, not {entry!r}.")
        sys.exit(1)
    parts = [part.strip().lower() for part in entry.split(",")]
    if len(parts) > 3 or not all(parts):
        logger.error(
            f"cities_analyse entry {entry!r} is not \"City\", \"City, ISO3\" or \"City, State, ISO3\"."
        )
        sys.exit(1)
    if len(parts) == 1:
        return parts[0], None, None
    if len(parts) == 2:
        return parts[0], None, parts[1]
    return parts[0], parts[1], parts[2]


def _filter_cities(df_mapping: pl.DataFrame, require_match: bool = True) -> pl.DataFrame:
    """Keep only the cities listed in cities_analyse; an empty list keeps all.

    City names repeat across countries and states (Aberdeen is in GBR and in
    two US states), so an entry can name the ISO3 code, and the state as
    well, to pick one of them. A bare name keeps every city of that name and
    logs which ones matched. Names are matched case-insensitively against
    locality and locality_aka. An entry that matches no city stops the run,
    so a typo is not mistaken for a city with no data. The second pass, after
    the population and footage thresholds, passes require_match=False,
    because a listed city those thresholds removed is a legitimate result
    there, not a typo, and its matches were already reported.
    """
    entries = common.get_configs("cities_analyse")
    if not entries:
        return df_mapping
    if not isinstance(entries, list):
        logger.error(f"cities_analyse must be a list, not {entries!r}.")
        sys.exit(1)

    locality = pl.col("locality").cast(pl.Utf8).str.strip_chars().str.to_lowercase()
    aliases = (
        pl.col("locality_aka").cast(pl.Utf8).fill_null("").str.strip_chars("[] ")
        .str.to_lowercase().str.split(",").list.eval(pl.element().str.strip_chars().str.strip_chars("'\""))
        if "locality_aka" in df_mapping.columns
        else pl.lit([], dtype=pl.List(pl.Utf8))
    )
    state = pl.col("state").cast(pl.Utf8).fill_null("").str.strip_chars().str.to_lowercase()
    iso3 = pl.col("iso3").cast(pl.Utf8).fill_null("").str.strip_chars().str.to_lowercase()

    keep = pl.lit(False)
    unmatched = []
    for entry in entries:
        name, entry_state, entry_iso3 = _parse_city_entry(entry)
        condition = (locality == name) | aliases.list.contains(name)
        if entry_state is not None:
            condition = condition & (state == entry_state)
        if entry_iso3 is not None:
            condition = condition & (iso3 == entry_iso3)
        matches = df_mapping.filter(condition).select(["locality", "state", "iso3"]).unique()
        if matches.height == 0:
            unmatched.append(entry)
        elif matches.height > 1 and entry_iso3 is None and require_match:
            found = "; ".join(
                ", ".join(str(v) for v in (row["locality"], row["state"], row["iso3"]) if v)
                for row in matches.sort(["iso3", "state"], nulls_last=True).iter_rows(named=True)
            )
            logger.warning(
                f"cities_analyse entry {entry!r} matches {matches.height} cities, all of which are kept: "
                f"{found}. Write \"City, ISO3\" or \"City, State, ISO3\" to keep only one."
            )
        keep = keep | condition

    if unmatched and require_match:
        logger.error(f"cities_analyse entries match no city in the mapping: {', '.join(repr(e) for e in unmatched)}.")
        sys.exit(1)
    if unmatched:
        logger.info(
            f"cities_analyse entries removed by the population and footage thresholds: "
            f"{', '.join(repr(e) for e in unmatched)}."
        )
    return df_mapping.filter(keep)


def _target_crossings_per_city() -> Optional[int]:
    """Return target_crossings_per_city as a positive integer, or None when off."""
    value = common.get_configs("target_crossings_per_city")
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)) or value <= 0 or float(value) != int(value):
        logger.error(f"target_crossings_per_city must be a positive integer or null, not {value!r}.")
        sys.exit(1)
    return int(value)


def _process_until_target(
    tasks: list,
    draw_order: Dict[tuple, tuple],
    target: int,
    process,
    crossings_found,
    prepare=None,
) -> list:
    """Process each city's segments in its random draw order until it has ``target`` crossings.

    Works in rounds of one segment per unfinished city, so a city stops within
    one segment of the target (whole segments are processed, so it can end a
    little above it). A city that runs out of footage keeps what it has. Returns
    the tasks processed, in processing order. ``prepare`` (optional) gets each
    round's tasks first and returns the ones that can be processed, e.g. after
    fetching their detection files.
    """
    queues: Dict[tuple, list] = {}
    for task in tasks:
        city, position = draw_order.get((str(task["video_id"]), int(task["start_index"])), (None, 0))
        queues.setdefault(city, []).append((position, task))
    for queue in queues.values():
        queue.sort(key=lambda item: item[0])
    found = {city: 0 for city in queues}
    processed: list = []
    round_number = 0
    while True:
        batch = [queue.pop(0)[1] for city, queue in queues.items() if queue and found[city] < target]
        if not batch:
            break
        if prepare is not None:
            batch = prepare(batch)
            if not batch:
                continue
        round_number += 1
        logger.info(
            f"target_crossings_per_city={target}, round {round_number}: {len(batch)} segment(s) "
            f"for {len(batch)} cities still below the target."
        )
        process(batch)
        for task in batch:
            city = draw_order.get((str(task["video_id"]), int(task["start_index"])), (None, 0))[0]
            found[city] += crossings_found(task)
        processed.extend(batch)
    reached = sum(1 for city, count in found.items() if count >= target)
    logger.info(
        f"target_crossings_per_city={target}: {reached} of {len(found)} cities reached the target "
        f"in {round_number} round(s) over {len(processed)} of {len(tasks)} available segments; "
        "the others used all their footage."
    )
    return processed


def _crossing_rule() -> str:
    """Return the configured crossing rule, stopping on an unusable setting.

    ``detector`` counts the CROWD detector's picks. ``road_crossing`` counts
    rule D from utils/crossing/road_crossing.py, which needs road-surface
    segmentation, so it refuses to run without it instead of silently
    counting crossings another way.
    """
    rule = str(common.get_configs("crossing_rule"))
    if rule not in CROSSING_RULES:
        logger.error(f"crossing_rule must be one of {CROSSING_RULES}, not {rule!r}.")
        sys.exit(1)
    if rule == "road_crossing" and not common.get_configs("use_segmentation"):
        logger.error(
            "crossing_rule is 'road_crossing', which decides crossings from the road "
            "surface under each pedestrian, but use_segmentation is false in config."
        )
        sys.exit(1)
    return rule


def _filter_nested_ids(nested: Optional[dict], stem: str, keep: set) -> Optional[dict]:
    """Keep only ``keep`` ids of ``stem`` in a ``{locality: {stem: {id: value}}}`` value."""
    if not nested:
        return None
    output = {}
    for outer_key, videos in nested.items():
        values = {
            track_id: value
            for track_id, value in (videos.get(stem) or {}).items()
            if crossing_metrics_module.normalise_id(track_id) in keep
        }
        if values:
            output[outer_key] = {stem: values}
    return output or None


def _apply_road_crossing_rule(
    selected: Dict[str, list],
    road_candidates: Dict[str, dict],
    stem_locality: Dict[str, int],
    pedestrian_crossing_count: dict,
    pedestrian_crossing_count_all: dict,
    data: dict,
    aggregate_by_locality: dict,
) -> tuple:
    """Make rule D's crossings the counted ones, in place of the detector's.

    Every structure the rest of the analysis reads (crossing ids and their
    frame bounds, crossing durations, per-city crossing counts, speeds and
    waiting times) is rebuilt from the rule-D selection, so nothing downstream
    needs to know which rule produced it. Returns the new all_speed and
    all_time.
    """
    for bucket in aggregate_by_locality.values():
        bucket["total_crossing_detect"] = 0
    all_speed: dict = {}
    all_time: dict = {}
    for stem in pedestrian_crossing_count:
        payload = road_candidates.get(stem) or {}
        chosen = list(selected.get(stem, []))
        keep = {crossing_metrics_module.normalise_id(track_id) for track_id in chosen}
        bounds = payload.get("id_bounds") or {}
        pedestrian_crossing_count[stem] = {
            "ids": chosen,
            "id_bounds": {track_id: bounds[track_id] for track_id in chosen if track_id in bounds},
        }
        pedestrian_crossing_count_all[stem] = {"ids": list(payload.get("ids") or [])}
        data[stem] = {
            track_id: value
            for track_id, value in (payload.get("temp_data") or {}).items()
            if crossing_metrics_module.normalise_id(track_id) in keep
        }
        if stem in stem_locality and stem_locality[stem] in aggregate_by_locality:
            aggregate_by_locality[stem_locality[stem]]["total_crossing_detect"] += len(chosen)
        for source, target in ((payload.get("speed_value"), all_speed), (payload.get("time_value"), all_time)):
            filtered = _filter_nested_ids(source, stem, keep)
            for outer_key, inner in (filtered or {}).items():
                target.setdefault(outer_key, {}).update(inner)
    return all_speed, all_time


def _prepare_waymo_tuned_parameters() -> Dict[str, object]:
    """Prepare or load the frozen Waymo model before CROWD CSV analysis.

    Crossing speeds are always reported in metres per second, which needs the
    Waymo-calibrated speed model to be present and qualified. The model is
    built from the raw Waymo data when process_waymo_if_missing is enabled.
    If it still cannot be loaded, the analysis stops here rather than falling
    back to the dimensionless relative-motion index.
    """
    global waymo_processed_roots
    from utils.crossing.metrics import (
        ensure_waymo_processed,
        metric_speed_is_qualified,
        tuned_crossing_parameters,
    )

    raw_dataset_path = common.get_configs("waymo_dataset_path")
    try:
        waymo_processed_roots = list(
            ensure_waymo_processed(
                raw_dataset_path=raw_dataset_path,
                repository_root=common.root_dir,
                output_root=common.output_dir,
                process_if_missing=bool(
                    common.get_configs("process_waymo_if_missing")
                ),
                log=lambda message: logger.info(message),
            )
        )
    except Exception as error:
        logger.error(f"Waymo preparation failed: {error}")
        waymo_processed_roots = []

    if not metric_speed_is_qualified():
        if not raw_dataset_path or not os.path.isdir(os.path.expanduser(str(raw_dataset_path))):
            reason = (
                f"waymo_dataset_path in config ({raw_dataset_path!r}) is not an existing "
                "directory, so the Waymo data could not be processed"
            )
        elif (
            common.get_configs("process_waymo_if_missing")
            and shutil.which("docker") is None
            and shutil.which("uv") is None
        ):
            reason = (
                "the Waymo data has not been processed yet and neither Docker nor uv, "
                "one of which the raw TFRecord export needs, is installed"
            )
        elif not common.get_configs("process_waymo_if_missing"):
            reason = (
                "no processed Waymo model exists and process_waymo_if_missing is "
                "false in config"
            )
        else:
            reason = (
                "Waymo processing or calibration did not produce a qualified speed "
                "model; see the Waymo messages above"
            )
        logger.error(
            "Crossing speeds must be in m/s, but the Waymo speed model is not "
            f"available: {reason}. Stopping instead of reporting the relative index."
        )
        sys.exit(1)

    os.environ["CROWD_CROSSING_SPEED_UNIT"] = "m/s"
    logger.info(
        "Using the fixed CROWD crossing algorithm with the qualified Waymo speed "
        "model; crossing speeds are in m/s."
    )
    return tuned_crossing_parameters()


# Execute analysis
if __name__ == "__main__":
    logger.info("Analysis started.")

    parquet_summary = sync_detection_parquet_store()
    logger.info(
        f"Parquet source ready: {parquet_summary['stored_files']} stored file(s); "
        f"{parquet_summary['source_files']} CSV file(s) currently available, "
        f"{parquet_summary['converted']} converted or refreshed, "
        f"{parquet_summary['reused']} corresponding Parquet file(s) reused, "
        f"{parquet_summary.get('skipped_empty', 0)} empty CSV file(s) skipped."
    )

    waymo_crossing_parameters = _prepare_waymo_tuned_parameters()
    crossing_rule = _crossing_rule()
    logger.info(f"Crossing rule: {crossing_rule}.")
    city_limit = _normalise_city_limit(common.get_configs("n_cities"))

    current_cache_config = _current_cache_config()
    cached_results: Optional[tuple] = None
    cached_config: Optional[Dict[str, object]] = None
    cached_segmentation: Dict[str, object] = {}
    use_cached_results = False

    # Populated by the segmentation pass, or restored from the cache, and
    # written back into the pickle so a cached rerun still reports the
    # segmentation-derived columns.
    segmentation_results: Dict[str, object] = {}

    if os.path.exists(file_results) and not common.get_configs("always_analyse"):
        (
            cached_results,
            cached_config,
            cached_speed_unit,
            cached_segmentation,
        ) = _load_results_cache(file_results)

        if cached_results is not None and cached_speed_unit != "m/s":
            logger.info(
                "Cached analysis results in {} hold crossing speeds in {!r}, not m/s; "
                "reanalysing.",
                file_results,
                cached_speed_unit or "relative",
            )
        elif cached_results is not None and cached_config == current_cache_config:
            use_cached_results = True
            # Only now that the cached values will actually be reused does
            # their recorded unit describe the data the figures will show.
            _apply_cached_speed_unit(cached_speed_unit)
            logger.info(
                "Using cached analysis results from {} because the cache "
                "configuration matches the current configuration: {}.",
                file_results,
                current_cache_config,
            )
        elif cached_results is not None and cached_config is None:
            logger.info(
                "Existing {} has no cache configuration metadata; "
                "reanalysing once and upgrading the pickle format.",
                file_results,
            )
        elif cached_results is not None and cached_config is not None:
            differences = _cache_config_differences(
                cached_config,
                current_cache_config,
            )
            logger.info(
                "Cached analysis configuration differs from the current "
                "configuration; reanalysing. Changed values: {}.",
                differences,
            )
    elif os.path.exists(file_results) and common.get_configs("always_analyse"):
        logger.info(
            "always_analyse is true; bypassing cached analysis results in {}.",
            file_results,
        )

    if use_cached_results:
        # Load the data from the pickle file
        with open(file_results, 'rb') as file:
            (data,                                          # 0
             person_counter,                                # 1
             bicycle_counter,                               # 2
             car_counter,                                   # 3
             motorcycle_counter,                            # 4
             bus_counter,                                   # 5
             truck_counter,                                 # 6
             cellphone_counter,                             # 7
             traffic_light_counter,                         # 8
             stop_sign_counter,                             # 9
             pedestrian_cross_locality,                         # 10
             pedestrian_crossing_count,                     # 11
             person_locality,                                   # 12
             bicycle_locality,                                  # 13
             car_locality,                                      # 14
             motorcycle_locality,                               # 15
             bus_locality,                                      # 16
             truck_locality,                                    # 17
             cross_evnt_locality,                               # 18
             vehicle_locality,                                  # 19
             cellphone_locality,                                # 20
             traffic_sign_locality,                             # 21
             all_speed,                                     # 22
             all_time,                                      # 23
             avg_time_locality,                                 # 24
             avg_speed_locality,                                # 25
             df_mapping,                                    # 26
             avg_speed_country,                             # 27
             avg_time_country,                              # 28
             crossings_with_traffic_equipment_locality,         # 29
             crossings_without_traffic_equipment_locality,      # 30
             crossings_with_traffic_equipment_country,      # 31
             crossings_without_traffic_equipment_country,   # 32
             min_max_speed,                                 # 33
             min_max_time,                                  # 34
             pedestrian_cross_country,                      # 35
             all_speed_locality,                                # 36
             all_time_locality,                                 # 37
             all_speed_country,                             # 38
             all_time_country,                              # 39
             df_mapping_raw,                                # 40
             pedestrian_cross_locality_all,                     # 41
             pedestrian_cross_country_all                   # 42
             ) = cached_results

        logger.info("Loaded analysis results from pickle file.")
        segmentation_results = dict(cached_segmentation)
        if segmentation_results:
            logger.info(
                "Reusing segmentation-derived crossing metrics stored with the "
                "cached results."
            )
        log_rollups(df_mapping)
    else:
        # Store the mapping file
        df_mapping = pl.read_csv(common.get_configs("mapping"))

        # Produce map with all data
        df = df_mapping.clone()  # copy df to manipulate for output
        df = df.with_columns(pl.col("state").fill_null("NA").alias("state"))

        # Sort by continent and locality, both in ascending order
        df = df.sort(by=["continent", "locality"])

        # Count of videos (handles: [id], "[id1,id2]", and [] -> 0)
        videos_clean = (
            pl.col("videos")
              .cast(pl.Utf8)
              .str.strip_chars("\"'")   # remove surrounding quotes if present
              .str.strip_chars("[]")     # remove surrounding brackets
              .str.strip_chars()         # trim whitespace
        )

        df = df.with_columns(
            pl.when(pl.col("videos").is_null() | (videos_clean == ""))
              .then(0)
              .otherwise(
                  videos_clean
                  .str.split(",")
                  .list.eval(pl.element().str.strip_chars())  # trim each item
                  .list.filter(pl.element() != "")            # drop empties (so [] -> 0)
                  .list.len()
              ).alias("video_count")
        )

        # Total amount of seconds in segments
        def flatten(lst):
            """Flattens nested lists like [[1, 2], [3, 4]] -> [1, 2, 3, 4]"""
            out = []
            for sub in lst:
                if isinstance(sub, list):
                    out.extend(sub)
                else:
                    out.append(sub)
            return out

        def compute_total_time(row: dict) -> int:
            try:
                start_raw = row.get("start_time")
                end_raw = row.get("end_time")

                start_times = flatten(ast.literal_eval(start_raw)) if start_raw is not None else []
                end_times = flatten(ast.literal_eval(end_raw)) if end_raw is not None else []

                return int(sum(e - s for s, e in zip(start_times, end_times)))
            except Exception as e:
                logger.error(f"Error in row {row.get('id')}: {e}")
                return 0

        df = df.with_columns(
            pl.struct(["id", "start_time", "end_time"])
              .map_elements(compute_total_time, return_dtype=pl.Int64)
              .alias("total_time")
        )

        # create flag_locality column
        flag_expr = pl.col("iso3").map_elements(
            lambda x: analysis_class.iso3_to_flag.get(x, "🏳️"),
            return_dtype=pl.Utf8,
        )

        # Create a new country label with emoji flag + country name
        df = df.with_columns([
            pl.concat_str([flag_expr, pl.col("locality").cast(pl.Utf8)], separator=" ").alias("flag_locality"),
            pl.concat_str([flag_expr, pl.col("country").cast(pl.Utf8)], separator=" ").alias("flag_country"),
        ])

        # Data to avoid showing on hover in scatter plots
        columns_remove = ['videos', 'time_of_day', 'start_time', 'end_time', 'upload_date', 'vehicle_type', 'channel',
                          'display_label', 'flag_locality', 'flag_country']

        hover_data = sorted(list(set(df.columns) - set(columns_remove)))

        # Sort by continent and locality, both in ascending order
        df = df.sort(["continent", "country"])

        # map with all cities
        maps.mapbox_map(df=df.to_pandas(),
                        hover_data=hover_data,
                        hover_name="flag_locality",
                        marker_size=4,
                        file_name='mapbox_map_all')

        # # map with all cities coloured by footage amount (continuous hue scale) + optional screenshot overlays
        # maps.mapbox_map_footage(df=df.to_pandas(),
        #                         footage_col="total_time",
        #                         hover_data=hover_data,
        #                         hover_name="flag_locality",
        #                         marker_size=3,
        #                         log_colour=True,
        #                         show_images=True,
        #                         file_name='mapbox_map_all_footage')

        maps.world_map_ss(
            df=df.to_pandas(),
            df_mapping=df_mapping.to_pandas(),
            show_images=True,
            hover_data=["total_time"],
            save_file=False,
            show_colorbar=True,
            colorbar_title="Footage (hours)",
        )

        # Sort by continent and locality, both in ascending order
        df = df.sort(["country", "locality"])

        # scatter plot for cities with number of videos over total time
        bivariate.scatter(df=df,
                          x="total_time",
                          y="video_count",
                          color="flag_country",
                          text="flag_locality",
                          xaxis_title='Total time of footage (s)',
                          yaxis_title='Number of videos',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="flag_locality",
                          legend_title="",
                          # legend_x=0.01,
                          # legend_y=1.0,
                          label_distance_factor=5.0,
                          marginal_x=None,  # type: ignore
                          marginal_y=None,  # type: ignore
                          file_name='scatter_all_total_time-video_count')  # type: ignore
        # scatter plot for countries with number of videos over total time

        # compute total time per locality first
        videos_raw = pl.col("videos").cast(pl.Utf8).str.strip_chars()
        videos_unquoted = videos_raw.str.strip_chars("\"'")

        videos_is_list = videos_unquoted.str.starts_with("[") & videos_unquoted.str.ends_with("]")

        videos_inner = (
            videos_unquoted
            .str.strip_chars("[]")
            .str.replace_all(r"[\"']", "")   # remove any inner quotes
            .str.replace_all(r"\s+", "")     # remove whitespace
            .str.strip_chars()
        )

        locality_video_count_expr = (
            pl.when(videos_is_list & (videos_inner != ""))
              .then(
                  videos_inner
                  .str.split(",")
                  .list.filter(pl.element() != "")
                  .list.len()
              ).otherwise(0).cast(pl.Int64).alias("locality_video_count")
        )

        # Normalize end_time string and sum all numbers found (nested-safe)
        end_raw = pl.col("end_time").cast(pl.Utf8).str.strip_chars()
        end_unquoted = end_raw.str.strip_chars("\"'")
        end_is_list = end_unquoted.str.starts_with("[")

        locality_total_time_expr = (
            pl.when(end_is_list)
              .then(
                  end_unquoted
                  .str.extract_all(r"\d+")
                  .list.eval(pl.element().cast(pl.Int64))
                  .list.sum()
                  .fill_null(0)
              ).otherwise(0).cast(pl.Int64).alias("locality_total_time")
        )

        df = df.with_columns([locality_video_count_expr, locality_total_time_expr])

        # ---------- Aggregate to country level ----------
        df_country = (
            df.group_by(["country", "iso3", "continent"])
              .agg([
                  pl.col("locality_total_time").sum().alias("total_time"),
                  pl.col("locality_video_count").sum().alias("video_count"),
              ])
        )

        # add flag + iso3 label
        df_country = df_country.with_columns(
            pl.concat_str(
                [
                    pl.col("iso3").map_elements(
                        lambda x: analysis_class.iso3_to_flag.get(x, "🏳️"),
                        return_dtype=pl.Utf8,
                    ),
                    pl.col("iso3").cast(pl.Utf8),
                ],
                separator=" ",
            ).alias("flag_country")
        )

        # sort for readability
        df_country = df_country.sort(["continent", "country"])

        # define hover data
        hover_data = ["country", "continent", "total_time", "video_count"]

        # plot (convert at plotting boundary)
        bivariate.scatter(df=df_country,
                          x="total_time",
                          y="video_count",
                          color="continent",
                          text="flag_country",
                          xaxis_title="Total time of footage (s)",
                          yaxis_title="Number of videos",
                          pretty_text=False,
                          marker_size=12,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="flag_country",
                          legend_title="",
                          label_distance_factor=0.1,
                          marginal_x=None,  # type: ignore
                          marginal_y=None,  # type: ignore
                          file_name="scatter_all_country_total_time-video_count")

        # histogram of dates of videos
        distribution.video_histogram_by_month(df=df.to_pandas(),
                                              video_count_col='video_count',
                                              upload_date_col='upload_date',
                                              xaxis_title='Year',
                                              yaxis_title='Number of videos',
                                              save_file=True)

        # maps with all cities and population heatmap
        maps.mapbox_map(df=df.to_pandas(),
                        hover_data=hover_data,
                        density_col='population_locality',
                        density_radius=10,
                        file_name='mapbox_map_all_pop')

        # maps with all cities and video count heatmap
        maps.mapbox_map(df=df.to_pandas(),
                        hover_data=hover_data,
                        density_col='video_count',
                        density_radius=10,
                        file_name='mapbox_map_all_videos')

        # maps with all cities and total time heatmap
        maps.mapbox_map(df=df.to_pandas(),
                        hover_data=hover_data,
                        density_col='total_time',
                        density_radius=10,
                        file_name='mapbox_map_all_time')

        # Type of vehicle over time of day
        df = df_mapping.clone()  # copy df to manipulate for output

        # --- expand rows so each video becomes one row ---
        # Return type: List[Struct{vehicle_type: Utf8, time_of_day: Utf8}]
        pair_dtype = pl.List(
            pl.Struct([
                pl.Field("vehicle_type", pl.Utf8),
                pl.Field("time_of_day", pl.Utf8),
            ])
        )

        def expand_pairs(vs: str | None, ts: str | None) -> list[dict]:
            """Parse stringified lists (possibly nested) and emit expanded (vehicle_type,
               time_of_day) pairs as strings."""
            try:
                vehicle_types = ast.literal_eval(vs) if isinstance(vs, str) else None
                times_of_day = ast.literal_eval(ts) if isinstance(ts, str) else None
                if not (isinstance(vehicle_types, list) and isinstance(times_of_day, list)):
                    return []
            except Exception:
                return []

            out: list[dict] = []
            for v_type, tod in zip(vehicle_types, times_of_day):
                v_list = v_type if isinstance(v_type, list) else [v_type]
                t_list = tod if isinstance(tod, list) else [tod]
                for vt in v_list:
                    for t in t_list:
                        out.append({"vehicle_type": str(vt), "time_of_day": str(t)})
            return out

        def map_with_fallback(dct: dict, v):
            """
            Robust dict lookup for values that may arrive as str/int/float (or numeric strings).
            Tries:
              1) direct key
              2) string key (stripped)
              3) int key (from int(v) or int(float(v)) for "1.0")
              4) float key (rare, but safe)
            Returns None if no match.
            """
            if v is None:
                return None

            # 1) direct key
            if v in dct:
                return dct[v]

            # Normalize string form
            sv = v.strip() if isinstance(v, str) else str(v).strip()

            # 2) string key
            if sv in dct:
                return dct[sv]

            # 3) int key (handle "1" and "1.0")
            try:
                iv = int(sv)
                if iv in dct:
                    return dct[iv]
            except Exception:
                try:
                    iv = int(float(sv))
                    if iv in dct:
                        return dct[iv]
                except Exception:
                    pass

            # 4) float key (less common, but harmless)
            try:
                fv = float(sv)
                if fv in dct:
                    return dct[fv]
            except Exception:
                pass

            return None

        # --- expand rows ---
        df_expanded = (
            df.select(["vehicle_type", "time_of_day"])
              .with_columns(
                  pl.struct(["vehicle_type", "time_of_day"])
                  .map_elements(
                        lambda r: expand_pairs(r["vehicle_type"], r["time_of_day"]),
                        return_dtype=pair_dtype,
                    ).alias("pairs")
              ).select("pairs")               # avoid duplicate column name collisions
               .explode("pairs")
               .with_columns([
                  pl.col("pairs").struct.field("vehicle_type").alias("vehicle_type"),
                  pl.col("pairs").struct.field("time_of_day").alias("time_of_day"),
                  ]).drop("pairs")
        )

        # --- map to human-readable labels ---
        df_expanded = df_expanded.with_columns([
            pl.col("vehicle_type").map_elements(
                lambda x: map_with_fallback(analysis_class.vehicle_map, x),
                return_dtype=pl.Utf8,
            ).alias("vehicle_type_name"),
            pl.col("time_of_day").map_elements(
                lambda x: map_with_fallback(analysis_class.time_map, x),
                return_dtype=pl.Utf8,
            ).alias("time_of_day_name"),
        ])

        # drop rows where mapping failed
        df_expanded = df_expanded.filter(
            pl.col("vehicle_type_name").is_not_null() & pl.col("time_of_day_name").is_not_null()
        )

        # --- aggregate counts ---
        df_summary = (
            df_expanded
            .group_by(["vehicle_type_name", "time_of_day_name"])
            .len()
            .rename({"len": "count"})
        )

        # --- pivot into wide format for stacked bar plot ---
        df_pivot = df_summary.pivot(
            index="vehicle_type_name",
            on="time_of_day_name",      # renamed from `columns`
            values="count",
            aggregate_function="first",
        ).fill_null(0)

        # ensure consistent order of vehicle types
        vehicle_order = [
            "Car", "Bus", "Truck", "Two-wheeler", "Bicycle", "Automated car", "Automated bus", "Automated truck",
            "Automated two-wheeler", "Electric scooter"
        ]
        order_map = {name: i for i, name in enumerate(vehicle_order)}

        df_pivot = (
            df_pivot
            .with_columns(
                pl.col("vehicle_type_name")
                  .map_elements(lambda x: order_map.get(x, 10**9), return_dtype=pl.Int64)
                  .alias("_order")
            )
            .sort("_order")
            .drop("_order")
        )
        # --- plot ---
        distribution.bar(
            df=df_pivot.to_pandas(),
            x=df_pivot["vehicle_type_name"],
            y=[col for col in ["Day", "Night"] if col in df_pivot.columns],
            y_legend=["Day", "Night"],
            stacked=True,
            pretty_text=False,
            orientation="v",
            xaxis_title="Type of vehicle",
            yaxis_title="Number of segments",
            show_text_labels=False,
            save_file=True,
            save_final=True,
            name_file="bar_vehicle_type_time_of_day"
        )

        # Continent over time of day
        df = df_mapping.clone()  # copy df to manipulate for output

        # --- expand rows so each video becomes one row ---
        pair_dtype = pl.List(
            pl.Struct([
                pl.Field("continent", pl.Utf8),
                pl.Field("time_of_day", pl.Utf8),
            ])
        )

        def expand_continent_tod(continent: str | None, ts: str | None) -> list[dict]:
            try:
                times_of_day = ast.literal_eval(ts) if isinstance(ts, str) else None
                if not isinstance(times_of_day, list):
                    return []
            except Exception:
                return []

            cont = "" if continent is None else str(continent)

            out: list[dict] = []
            for tod in times_of_day:
                t_list = tod if isinstance(tod, list) else [tod]
                for t in t_list:
                    out.append({"continent": cont, "time_of_day": str(t)})
            return out

        # --- expand rows so each time-of-day entry becomes one row ---
        df_expanded = (
            df.select(["continent", "time_of_day"])
              .with_columns(
                  pl.struct(["continent", "time_of_day"])
                  .map_elements(
                        lambda r: expand_continent_tod(r["continent"], r["time_of_day"]),
                        return_dtype=pair_dtype,
                    ).alias("pairs")
              ).select("pairs").explode("pairs").with_columns([
                  pl.col("pairs").struct.field("continent").alias("continent"),
                  pl.col("pairs").struct.field("time_of_day").alias("time_of_day"),
              ]).drop("pairs")
        )

        # --- map to human-readable labels ---
        df_expanded = df_expanded.with_columns(
            pl.col("time_of_day").map_elements(
                lambda x: map_with_fallback(analysis_class.time_map, x),
                return_dtype=pl.Utf8,
            ).alias("time_of_day_name")
        )

        # drop rows where mapping failed
        df_expanded = df_expanded.filter(
            pl.col("time_of_day_name").is_not_null() & pl.col("continent").is_not_null() & (pl.col("continent") != "")
        )

        # --- aggregate counts ---
        df_summary = (
            df_expanded
            .group_by(["continent", "time_of_day_name"])
            .len()
            .rename({"len": "count"})
        )

        # --- pivot into wide format for stacked bar plot ---
        df_pivot = (
            df_summary
            .pivot(
                index="continent",
                on="time_of_day_name",   # Polars >= 1.0.0 uses `on` (not `columns`)
                values="count",
                aggregate_function="first",
            )
            .fill_null(0)
        )

        # ensure only expected columns (and ensure they exist)
        for col in ["Day", "Night"]:
            if col not in df_pivot.columns:
                df_pivot = df_pivot.with_columns(pl.lit(0).alias(col))

        # --- enforce alphabetical continent order ---
        df_pivot = df_pivot.sort("continent")

        time_columns = [col for col in ["Day", "Night"] if col in df_pivot.columns]

        # --- plot ---
        distribution.bar(
            df=df_pivot.to_pandas(),
            x=df_pivot["continent"],
            y=time_columns,
            y_legend=time_columns,
            stacked=True,
            pretty_text=False,
            orientation="v",
            xaxis_title="Continent",
            yaxis_title="Number of segments",
            show_text_labels=False,
            save_file=True,
            save_final=True,
            name_file="bar_continent_time_of_day"
        )

        total_duration = dataset_stats.calculate_total_seconds(df_mapping)

        # Displays values before applying filters
        logger.info(
            f"Duration of videos in seconds: {total_duration}, "
            f"in minutes: {total_duration/60:.2f}, "
            f"in hours: {total_duration/3600:.2f}, "
            f"in days: {total_duration/86400:.2f}, "
            f"in weeks: {total_duration/604800:.2f}, "
            f"in months: {total_duration/2629800:.2f}, "   # average month (30.44 days)
            f"in years: {total_duration/31557600:.2f} "    # average year (365.25 days)
            f"before filtering."
        )
        logger.info("Total number of videos before filtering: {}.",
                    dataset_stats.calculate_total_videos(df_mapping))

        country, number, _ = metrics_cache.get_unique_values(df_mapping, "iso3")
        logger.info(f"Total number of countries and territories before filtering: {number}.")

        locality_state_iso3, number, dup_report = metrics_cache.get_unique_values(
            df_mapping,
            ["locality", "state", "iso3"],
            return_duplicates=True,
        )

        logger.info(f"Total number of unique locality+state+iso3 keys: {number}.")

        if dup_report is not None and dup_report.height > 0:
            logger.warning(f"Duplicated keys:\n{dup_report}")

        # Limit countries if required
        countries_include = common.get_configs("countries_analyse")
        if countries_include:
            df_mapping = df_mapping.filter(pl.col("iso3").is_in(countries_include))
        df_mapping = _filter_cities(df_mapping)
        df_mapping = _select_cities_by_footage(
            df_mapping,
            city_limit,
        )
        if remote_detections.enabled():
            # Index the server's detection files for the chosen cities, so the
            # footage selection below sees every segment that has detections.
            remote_detections.build_index(remote_detections.mapping_segments(df_mapping))
            MetaData.clear_video_index_cache()
        log_rollups(df_mapping)

        # Make a dict for all columns
        locality_country_cols = {
            # Object columns
            'person': 0, 'bicycle': 0, 'car': 0, 'motorcycle': 0, 'airplane': 0, 'bus': 0, 'train': 0,
            'truck': 0, 'boat': 0, 'traffic_light': 0, 'fire_hydrant': 0, 'stop_sign': 0, 'parking_meter': 0,
            'bench': 0, 'bird': 0, 'cat': 0, 'dog': 0, 'horse': 0, 'sheep': 0, 'cow': 0, 'elephant': 0, 'bear': 0,
            'zebra': 0, 'giraffe': 0, 'backpack': 0, 'umbrella': 0, 'handbag': 0, 'tie': 0, 'suitcase': 0,
            'frisbee': 0, 'skis': 0, 'snowboard': 0, 'sports_ball': 0, 'kite': 0, 'baseball_bat': 0,
            'baseball_glove': 0, 'skateboard': 0, 'surfboard': 0, 'tennis_racket': 0, 'bottle': 0, 'wine_glass': 0,
            'cup': 0, 'fork': 0, 'knife': 0, 'spoon': 0, 'bowl': 0, 'banana': 0, 'apple': 0, 'sandwich': 0,
            'orange': 0, 'broccoli': 0, 'carrot': 0, 'hot_dog': 0, 'pizza': 0, 'donut': 0, 'cake': 0, 'chair': 0,
            'couch': 0, 'potted_plant': 0, 'bed': 0, 'dining_table': 0, 'toilet': 0, 'tv': 0, 'laptop': 0,
            'mouse': 0, 'remote': 0, 'keyboard': 0, 'cellphone': 0, 'microwave': 0, 'oven': 0, 'toaster': 0,
            'sink': 0, 'refrigerator': 0, 'book': 0, 'clock': 0, 'vase': 0, 'scissors': 0, 'teddy_bear': 0,
            'hair_drier': 0, 'toothbrush': 0,

            'total_time': 0,
            'total_crossing_detect': 0,

            # locality-level columns
            'speed_crossing_day_locality': math.nan,
            'speed_crossing_night_locality': math.nan,
            'speed_crossing_day_night_locality_avg': math.nan,
            'time_crossing_day_locality': math.nan,
            'time_crossing_night_locality': math.nan,
            'time_crossing_day_night_locality_avg': math.nan,
            'with_trf_light_day_locality': 0.0,
            'with_trf_light_night_locality': 0.0,
            'without_trf_light_day_locality': 0.0,
            'without_trf_light_night_locality': 0.0,
            'crossing_detected_locality': 0,
            'crossing_detected_locality_day': 0,
            'crossing_detected_locality_night': 0,
            'crossing_detected_locality_all': 0,
            'crossing_detected_locality_all_day': 0,
            'crossing_detected_locality_all_night': 0,

            # Country-level columns
            'speed_crossing_day_country': math.nan,
            'speed_crossing_night_country': math.nan,
            'speed_crossing_day_night_country_avg': math.nan,
            'time_crossing_day_country': math.nan,
            'time_crossing_night_country': math.nan,
            'time_crossing_day_night_country_avg': math.nan,
            'with_trf_light_day_country': 0.0,
            'with_trf_light_night_country': 0.0,
            'without_trf_light_day_country': 0.0,
            'without_trf_light_night_country': 0.0,
            'crossing_detected_country': 0,
            'crossing_detected_country_day': 0,
            'crossing_detected_country_night': 0,
            'crossing_detected_country_all': 0,
            'crossing_detected_country_all_day': 0,
            'crossing_detected_country_all_night': 0,
        }

        # Efficiently add all columns at once
        df_mapping = df_mapping.with_columns(
            [pl.lit(v).alias(k) for k, v in locality_country_cols.items()]
        )

        # Precompute fast lookup once (place this immediately before the loops)
        id_to_place: dict[int, tuple[str, str, str]] = {
            int(row_id): (locality, state, country)
            for row_id, locality, state, country in df_mapping.select(["id", "locality", "state", "country"]).iter_rows()  # noqa: E501
        }

        # Parse mapping segment metadata once. All CSV task lookups below are
        # direct dictionary accesses rather than row scans plus ast parsing.
        segment_lookup = metadata.segment_lookup(df_mapping)

        all_speed = {}
        all_time = {}

        logger.info("Processing detection files from the Parquet store.")
        pedestrian_crossing_count, data = {}, {}
        pedestrian_crossing_count_all = {}

        min_conf = common.get_configs("min_confidence")

        # ------------------------------------------------------------------
        # Concurrent bbox processing
        # ------------------------------------------------------------------
        csv_tasks = []

        boundary_left_default = float(common.get_configs("boundary_left"))
        boundary_right_default = float(common.get_configs("boundary_right"))

        for folder_path in configured_parquet_roots():
            bbox_path = os.path.join(folder_path, DETECTION_FOLDER)
            if not os.path.exists(bbox_path):
                logger.warning(f"Folder does not exist: {bbox_path}.")
                continue

            detection_files = analytics_IO.parquet_detection_files(bbox_path)
            for file_name in tqdm(
                detection_files,
                desc=f"Indexing detection files in {bbox_path}",
            ):
                filtered: Optional[str] = analytics_IO.filter_detection_file(
                    file=file_name,
                    df_mapping=df_mapping,
                )
                if filtered is None:
                    continue

                file_str = os.fspath(filtered)
                if file_str in MISC_FILES:
                    continue

                file_path = os.path.join(bbox_path, file_str)
                filename_no_ext = os.path.splitext(file_str)[0]

                try:
                    video_id, start_index_text, fps_text = (
                        filename_no_ext.rsplit("_", 2)
                    )
                    start_index = int(start_index_text)
                    fps = float(fps_text)
                except (TypeError, ValueError):
                    logger.warning(
                        f"Unexpected filename format: {filename_no_ext}"
                    )
                    continue

                segment_meta = segment_lookup.get(
                    (video_id, start_index)
                )
                video_locality_id = (
                    segment_meta[0]
                    if segment_meta is not None
                    else None
                )
                time_video = (
                    float(segment_meta[1])
                    if segment_meta is not None
                    else 0.0
                )

                place = (
                    id_to_place.get(int(video_locality_id))
                    if video_locality_id is not None
                    else None
                )
                if place is None:
                    logger.warning(
                        f"{file_str}: no mapping row found for "
                        f"id={video_locality_id}."
                    )
                    continue

                video_locality, video_state, video_country = place
                logger.debug(
                    f"{file_str}: found values {video_locality}, "
                    f"{video_state}, {video_country}."
                )

                csv_tasks.append(
                    {
                        "file_path": file_path,
                        "file_name": file_str,
                        "filename_no_ext": filename_no_ext,
                        "video_id": video_id,
                        "start_index": start_index,
                        "fps": fps,
                        "video_locality_id": int(video_locality_id),
                        "time_video": time_video,
                        "is_bbox_stream": True,
                    }
                )

        if remote_detections.enabled():
            # Segments whose detection file is only on the server: their task
            # points at where the Parquet file will be, and it is fetched just
            # before the segment is processed.
            local_stems = {task["filename_no_ext"] for task in csv_tasks}
            for (video_id, start_index), stem in sorted(remote_detections.indexed_segments().items()):
                if stem in local_stems:
                    continue
                if analytics_IO.filter_detection_file(file=f"{stem}.parquet", df_mapping=df_mapping) is None:
                    continue
                segment_meta = segment_lookup.get((video_id, start_index))
                if segment_meta is None or id_to_place.get(int(segment_meta[0])) is None:
                    continue
                csv_tasks.append(
                    {
                        "file_path": str(remote_detections.parquet_path(stem)),
                        "file_name": f"{stem}.parquet",
                        "filename_no_ext": stem,
                        "video_id": video_id,
                        "start_index": int(start_index),
                        "fps": float(stem.rsplit("_", 1)[1]),
                        "video_locality_id": int(segment_meta[0]),
                        "time_video": float(segment_meta[1]),
                        "is_bbox_stream": True,
                        "remote": True,
                    }
                )

        worker_env = os.environ.get("CROWD_CSV_WORKERS", "").strip()
        if worker_env:
            try:
                csv_workers = max(1, int(worker_env))
            except ValueError:
                logger.warning(
                    f"Invalid CROWD_CSV_WORKERS={worker_env!r}; using default."
                )
                csv_workers = min(common.get_configs("cpu_worker"), max(1, os.cpu_count() or 1))
        else:
            csv_workers = min(common.get_configs("cpu_worker"), max(1, os.cpu_count() or 1))

        csv_workers = min(csv_workers, len(csv_tasks)) if csv_tasks else 1

        logger.info(
            f"Analysing {len(csv_tasks)} detection files with "
            f"{csv_workers} worker process"
            f"{'es' if csv_workers != 1 else ''}."
        )

        aggregate_by_locality = {}
        # Rule-D candidates per detection segment, and the locality each
        # segment belongs to (used only when crossing_rule is road_crossing).
        road_candidates: Dict[str, dict] = {}
        stem_locality: Dict[str, int] = {}

        pipeline_model = dict(
            getattr(crossing_metrics_module, "_PIPELINE_MODEL", {}) or {}
        )
        speed_model = dict(
            getattr(crossing_metrics_module, "_SPEED_MODEL", {}) or {}
        )

        worker_init_args = (
            df_mapping,
            dict(waymo_crossing_parameters),
            float(min_conf),
            boundary_left_default,
            boundary_right_default,
            pipeline_model,
            speed_model,
        )

        def merge_worker_result(result):
            if result.get("status") != "ok":
                logger.error(
                    f"Detection analysis failed: "
                    f"{result.get('message', result.get('file_name', 'unknown'))}"
                )
                return

            metrics_cache.seed_file_metrics(
                result["file_name"],
                result["metric_counts"],
            )

            locality_id = int(result["video_locality_id"])
            bucket = aggregate_by_locality.setdefault(
                locality_id,
                {
                    "object_counts": {},
                    "total_time": 0.0,
                    "total_crossing_detect": 0,
                },
            )

            for class_name, value in result["object_counts"].items():
                bucket["object_counts"][class_name] = (
                    int(bucket["object_counts"].get(class_name, 0))
                    + int(value)
                )

            bucket["total_time"] += float(result["time_video"] or 0)

            if result["is_bbox_stream"]:
                filename_no_ext = result["filename_no_ext"]
                ids = result["ids"]
                all_ids = result["all_ids"]
                id_bounds = result.get("id_bounds") or {}
                temp_data = result["temp_data"]

                pedestrian_crossing_count[filename_no_ext] = {
                    "ids": ids,
                    "id_bounds": id_bounds,
                }
                pedestrian_crossing_count_all[filename_no_ext] = {
                    "ids": all_ids
                }
                data[filename_no_ext] = temp_data
                bucket["total_crossing_detect"] += len(ids)
                stem_locality[filename_no_ext] = locality_id
                if result.get("road_candidates") is not None:
                    road_candidates[filename_no_ext] = result["road_candidates"]

                speed_value = result["speed_value"]
                if speed_value is not None:
                    for outer_key, inner_dict in speed_value.items():
                        all_speed.setdefault(outer_key, {}).update(inner_dict)

                time_value = result["time_value"]
                if time_value is not None:
                    for outer_key, inner_dict in time_value.items():
                        all_time.setdefault(outer_key, {}).update(inner_dict)

        target_crossings = _target_crossings_per_city()
        road_selected: Dict[str, list] = {}
        # [failed, attempted] across rounds: the failure limit applies to the
        # whole run, not to each (possibly tiny) round.
        road_failures = [0, 0]

        def run_detection(tasks: list, executor) -> None:
            if executor is None:
                results = (process_csv_task(task) for task in tasks)
            else:
                results = executor.map(process_csv_task, tasks, chunksize=1)
            for result in tqdm(results, total=len(tasks), desc="Analysing detection files"):
                merge_worker_result(result)

        def select_crossings(tasks: list) -> None:
            # Rule D decides the crossings from the road surface under each
            # candidate's feet, so it runs before any count, speed or waiting
            # time is aggregated from the detector's picks.
            if crossing_rule != "road_crossing":
                return
            stems = {str(task["filename_no_ext"]) for task in tasks}
            road_selected.update(
                segmentation_pass.select_road_crossings(
                    df_mapping,
                    tasks,
                    {stem: road_candidates[stem] for stem in stems if stem in road_candidates},
                    waymo_crossing_parameters,
                    failure_totals=road_failures,
                )
            )

        def crossings_found(task: dict) -> int:
            stem = str(task["filename_no_ext"])
            if crossing_rule == "road_crossing":
                return len(road_selected.get(stem, []))
            return len((pedestrian_crossing_count.get(stem) or {}).get("ids") or [])

        if csv_workers == 1:
            initialise_csv_worker(*worker_init_args)
            executor_context = contextlib.nullcontext(None)
        else:
            executor_context = ProcessPoolExecutor(
                max_workers=csv_workers,
                mp_context=multiprocessing.get_context("spawn"),
                initializer=initialise_csv_worker,
                initargs=worker_init_args,
            )
        with executor_context as executor:
            if target_crossings is None:
                if remote_detections.enabled():
                    csv_tasks = remote_detections.fetch_tasks(csv_tasks)
                run_detection(csv_tasks, executor)
                select_crossings(csv_tasks)
            else:
                csv_tasks = _process_until_target(
                    csv_tasks,
                    MetaData.segment_draw_order(df_mapping),
                    int(target_crossings),
                    lambda tasks: (run_detection(tasks, executor), select_crossings(tasks)),
                    crossings_found,
                    prepare=remote_detections.fetch_tasks if remote_detections.enabled() else None,
                )

        if crossing_rule == "road_crossing":
            segmentation_pass.check_failure_share(*road_failures)
            all_speed, all_time = _apply_road_crossing_rule(
                road_selected,
                road_candidates,
                stem_locality,
                pedestrian_crossing_count,
                pedestrian_crossing_count_all,
                data,
                aggregate_by_locality,
            )

        if target_crossings is not None:
            # From here on every count is taken over the footage processed.
            MetaData.restrict_to_segments(
                {(str(task["video_id"]), int(task["start_index"])) for task in csv_tasks}
            )

        if aggregate_by_locality:
            update_rows = []

            for locality_id, bucket in aggregate_by_locality.items():
                row = {
                    "id": int(locality_id),
                    "_parallel_total_time": float(bucket["total_time"]),
                    "_parallel_total_crossing_detect": int(
                        bucket["total_crossing_detect"]
                    ),
                }

                object_counts = bucket["object_counts"]
                for class_name in PARALLEL_COCO_CLASSES:
                    row[f"_parallel_{class_name}"] = int(
                        object_counts.get(class_name, 0)
                    )

                update_rows.append(row)

            parallel_updates = pl.DataFrame(update_rows)

            df_mapping = df_mapping.join(
                parallel_updates,
                on="id",
                how="left",
            )

            df_mapping = df_mapping.with_columns(
                [
                    (
                        pl.col(class_name)
                        + pl.col(f"_parallel_{class_name}").fill_null(0)
                    ).alias(class_name)
                    for class_name in PARALLEL_COCO_CLASSES
                ]
                + [
                    (
                        pl.col("total_time")
                        + pl.col("_parallel_total_time").fill_null(0)
                    ).alias("total_time"),
                    (
                        pl.col("total_crossing_detect")
                        + pl.col(
                            "_parallel_total_crossing_detect"
                        ).fill_null(0)
                    ).alias("total_crossing_detect"),
                ]
            )

            df_mapping = df_mapping.drop(
                [
                    f"_parallel_{class_name}"
                    for class_name in PARALLEL_COCO_CLASSES
                ]
                + [
                    "_parallel_total_time",
                    "_parallel_total_crossing_detect",
                ]
            )

        person_counter = df_mapping.select(pl.col("person").sum()).item()
        bicycle_counter = df_mapping.select(pl.col("bicycle").sum()).item()
        car_counter = df_mapping.select(pl.col("car").sum()).item()
        motorcycle_counter = df_mapping.select(pl.col("motorcycle").sum()).item()
        bus_counter = df_mapping.select(pl.col("bus").sum()).item()
        truck_counter = df_mapping.select(pl.col("truck").sum()).item()
        cellphone_counter = df_mapping.select(pl.col("cellphone").sum()).item()
        traffic_light_counter = df_mapping.select(pl.col("traffic_light").sum()).item()
        stop_sign_counter = df_mapping.select(pl.col("stop_sign").sum()).item()

        # Record the average speed and time of crossing on country basis
        avg_speed_country, all_speed_country = metrics.avg_speed_of_crossing_country(df_mapping, all_speed)

        # Output in real world seconds
        avg_time_country, all_time_country = metrics.avg_time_to_start_cross_country(df_mapping, all_time)

        # Record the average speed and time of crossing on locality basis
        avg_speed_locality, all_speed_locality = metrics.avg_speed_of_crossing_locality(df_mapping, all_speed)
        avg_time_locality, all_time_locality = metrics.avg_time_to_start_cross_locality(df_mapping, all_time)
        # The bounding-box averages, kept before segmentation_is_primary may
        # replace the reported ones, for the *_bbox_* columns.
        bbox_averages = {
            "speed_locality": avg_speed_locality,
            "speed_country": avg_speed_country,
            "time_locality": avg_time_locality,
            "time_country": avg_time_country,
        }

        # Kill the program if there is no data to analyse
        if len(avg_time_locality) == 0 or len(avg_speed_locality) == 0:
            logger.error("No speed and time data to analyse.")
            exit()

        # ------------------------------------------------------------------
        # Road-surface segmentation pass
        #
        # The baseline metrics above infer both quantities from bounding-box
        # motion alone and have no notion of where the kerb is. This pass
        # segments the crossing windows with SegFormer Cityscapes, derives the
        # interval each pedestrian actually spends on the carriageway, and
        # recomputes both metrics against it. The results are written to
        # separate speed_crossing_seg / time_crossing_seg columns, and the
        # bounding-box values to speed_crossing_bbox / time_crossing_bbox,
        # so the two can be compared whichever one is reported.
        # ------------------------------------------------------------------
        segmentation_output = segmentation_pass.run_segmentation_pass(
            df_mapping=df_mapping,
            detection_tasks=csv_tasks,
            crossing_ids=pedestrian_crossing_count,
            crossing_parameters=waymo_crossing_parameters,
        )

        if segmentation_output.get("enabled"):
            seg_speed = segmentation_output["seg_speed"]
            seg_time = segmentation_output["seg_time"]

            seg_speed_min = float(common.get_configs("min_speed_limit"))
            seg_speed_max = float(common.get_configs("max_speed_limit"))
            seg_time_min = float(common.get_configs("min_waiting_time"))
            seg_time_max = float(common.get_configs("max_waiting_time"))

            avg_seg_speed_locality, all_seg_speed_locality = (
                segmentation_pass.average_by_locality(
                    seg_speed, seg_speed_min, seg_speed_max,
                )
            )
            avg_seg_speed_country, all_seg_speed_country = (
                segmentation_pass.average_by_country(
                    df_mapping, seg_speed, seg_speed_min, seg_speed_max,
                )
            )
            avg_seg_time_locality, all_seg_time_locality = (
                segmentation_pass.average_by_locality(
                    seg_time, seg_time_min, seg_time_max,
                )
            )
            avg_seg_time_country, all_seg_time_country = (
                segmentation_pass.average_by_country(
                    df_mapping, seg_time, seg_time_min, seg_time_max,
                )
            )

            segmentation_results = {
                "avg_speed_locality": avg_seg_speed_locality,
                "avg_speed_country": avg_seg_speed_country,
                "avg_time_locality": avg_seg_time_locality,
                "avg_time_country": avg_seg_time_country,
                "all_speed_locality": all_seg_speed_locality,
                "all_speed_country": all_seg_speed_country,
                "all_time_locality": all_seg_time_locality,
                "all_time_country": all_seg_time_country,
                "diagnostics": dict(segmentation_output["diagnostics"]),
            }

            logger.info(
                f"Segmentation metrics cover {len(avg_seg_speed_locality)} "
                f"locality-condition key(s) for speed and "
                f"{len(avg_seg_time_locality)} for hesitation time, against "
                f"{len(avg_speed_locality)} and {len(avg_time_locality)} "
                "respectively for the baseline."
            )

            # Which derivation the figures actually show.
            #
            # Every plot reads the crossing metrics out of results.pickle by
            # position, so making the segmentation values the reported ones is
            # a matter of substituting them here, before the pickle is
            # written. The bounding-box values stay available in the
            # speed_crossing_bbox and time_crossing_bbox columns either way,
            # and the segmentation values in speed_crossing_seg and
            # time_crossing_seg, so the two can still be compared.
            #
            # This is off by default because the segmentation metrics do not
            # cover every crossing the baseline covers: a track is dropped
            # when it never reaches the carriageway, when the frozen speed
            # model's reliability gates reject the shortened window, and, for
            # the hesitation time, whenever the pedestrian was already on the
            # road when first detected. A locality left with no qualifying
            # track disappears from the figures entirely.
            if _segmentation_is_primary():
                lost_speed = sorted(
                    set(avg_speed_locality) - set(avg_seg_speed_locality)
                )
                lost_time = sorted(
                    set(avg_time_locality) - set(avg_seg_time_locality)
                )
                logger.warning(
                    "segmentation_is_primary is enabled: the figures will "
                    "report the road-restricted crossing speed and the "
                    "kerb-anchored hesitation time. "
                    f"{len(lost_speed)} locality-condition key(s) lose their "
                    f"speed and {len(lost_time)} lose their hesitation time "
                    "relative to the baseline."
                )
                if lost_speed:
                    logger.warning(
                        f"No segmentation speed for: {', '.join(lost_speed[:20])}"
                        + (" ..." if len(lost_speed) > 20 else "")
                    )
                if lost_time:
                    logger.warning(
                        f"No segmentation hesitation time for: {', '.join(lost_time[:20])}"
                        + (" ..." if len(lost_time) > 20 else "")
                    )

                (
                    avg_speed_locality,
                    avg_speed_country,
                    all_speed_locality,
                    all_speed_country,
                ) = _primary_speed(
                    segmentation_results,
                    avg_speed_locality,
                    avg_speed_country,
                    all_speed_locality,
                    all_speed_country,
                )
                (
                    avg_time_locality,
                    avg_time_country,
                    all_time_locality,
                    all_time_country,
                ) = _primary_time(
                    segmentation_results,
                    avg_time_locality,
                    avg_time_country,
                    all_time_locality,
                    all_time_country,
                )
            else:
                logger.info(
                    "segmentation_is_primary is disabled: the figures continue "
                    "to report the baseline bounding-box metrics, and the "
                    "segmentation values are written to the speed_crossing_seg "
                    "and time_crossing_seg columns for comparison."
                )

        logger.info("Calculating counts of detected traffic signs.")
        traffic_sign_locality = metrics_cache.calculate_traffic_signs(df_mapping)

        logger.info("Calculating counts of detected mobile phones.")
        cellphone_locality = metrics_cache.calculate_cellphones(df_mapping)

        logger.info("Calculating counts of detected vehicles.")
        vehicle_locality = metrics_cache.calculate_traffic(df_mapping, motorcycle=True, car=True, bus=True, truck=True)

        logger.info("Calculating counts of detected bicycles.")
        bicycle_locality = metrics_cache.calculate_traffic(df_mapping, bicycle=True)

        logger.info("Calculating counts of detected cars (subset of vehicles).")
        car_locality = metrics_cache.calculate_traffic(df_mapping, car=True)

        logger.info("Calculating counts of detected motorcycles (subset of vehicles).")
        motorcycle_locality = metrics_cache.calculate_traffic(df_mapping, motorcycle=True)

        logger.info("Calculating counts of detected buses (subset of vehicles).")
        bus_locality = metrics_cache.calculate_traffic(df_mapping, bus=True)

        logger.info("Calculating counts of detected trucks (subset of vehicles).")
        truck_locality = metrics_cache.calculate_traffic(df_mapping, truck=True)

        logger.info("Calculating counts of detected persons.")
        person_locality = metrics_cache.calculate_traffic(df_mapping, person=True)

        logger.info("Calculating counts of detected crossing events with traffic lights.")
        cross_evnt_locality = events.crossing_event_wt_traffic_light(df_mapping, data)

        logger.info("Calculating counts of crossing events in cities.")

        pedestrian_cross_locality = dataset_stats.pedestrian_cross_per_locality(pedestrian_crossing_count, df_mapping)
        pedestrian_cross_locality_all = dataset_stats.pedestrian_cross_per_locality(pedestrian_crossing_count_all,
                                                                                    df_mapping)

        logger.info("Calculating counts of crossing events in countries.")
        pedestrian_cross_country = dataset_stats.pedestrian_cross_per_country(pedestrian_cross_locality, df_mapping)
        pedestrian_cross_country_all = dataset_stats.pedestrian_cross_per_country(pedestrian_cross_locality_all,
                                                                                  df_mapping)

        # Jaywalking data
        logger.info("Calculating parameters for detection of jaywalking.")

        (
            crossings_with_traffic_equipment_locality,
            crossings_without_traffic_equipment_locality,
            total_duration_by_locality,
            crossings_with_traffic_equipment_country,
            crossings_without_traffic_equipment_country,
            total_duration_by_country,
        ) = events.crossing_event_with_traffic_equipment(df_mapping, data)

        # ----------------------------------------------------------------------
        # Add locality-level crossing counts for with and without traffic equipment
        # ----------------------------------------------------------------------

        # Ensure target columns exist (init as 0.0 or null; choose 0.0 for counts)
        needed_cols = [
            "with_trf_light_day_locality",
            "with_trf_light_night_locality",
            "without_trf_light_day_locality",
            "without_trf_light_night_locality",
        ]
        for c in needed_cols:
            if c not in df_mapping.columns:
                df_mapping = df_mapping.with_columns(pl.lit(0.0).alias(c))

        # Helper: treat missing/unknown state consistently
        def _is_missing_state_expr(col: str) -> pl.Expr:
            return pl.col(col).is_null() | (pl.col(col).cast(pl.Utf8,
                                                             strict=False).str.to_lowercase().is_in(["nan",
                                                                                                     "na",
                                                                                                     "n/a",
                                                                                                     "unknown",
                                                                                                     ""]))

        # ---- WITH equipment ----
        with_rows = []
        for key, value in crossings_with_traffic_equipment_locality.items():
            parts = key.split("_")
            if len(parts) < 4:
                continue
            locality = parts[0]
            lat = float(parts[1])
            lon = float(parts[2])
            tod = int(parts[3])  # 0 day, 1 night
            with_rows.append({"locality": locality, "lat": lat, "lon": lon, "_tod": tod, "_val": float(value)})

        if with_rows:
            df_with = pl.DataFrame(with_rows)

            # bring in state from mapping (first non-null state for locality+lat+lon, else null)
            if "state" in df_mapping.columns:
                state_lookup = (
                    df_mapping
                    .select(["locality", "lat", "lon", "state"])
                    .group_by(["locality", "lat", "lon"])
                    .agg(pl.col("state").drop_nulls().first().alias("_state"))
                )
                df_with = df_with.join(state_lookup, on=["locality", "lat", "lon"], how="left")
            else:
                df_with = df_with.with_columns(pl.lit(None).cast(pl.Utf8).alias("_state"))

            df_with = df_with.with_columns([
                pl.when(pl.col("_tod") == 0).then(pl.col("_val")).otherwise(None).alias("_with_day"),
                pl.when(pl.col("_tod") == 1).then(pl.col("_val")).otherwise(None).alias("_with_night"),
            ]).group_by(["locality", "lat", "lon", "_state"]).agg([
                pl.col("_with_day").drop_nulls().last().alias("with_trf_light_day_locality"),
                pl.col("_with_night").drop_nulls().last().alias("with_trf_light_night_locality"),
            ])

            # Join updates to mapping by locality+lat+lon, then only apply if state matches OR both missing
            df_mapping = df_mapping.join(
                df_with.select(["locality", "lat", "lon", "_state",
                                "with_trf_light_day_locality", "with_trf_light_night_locality"]),
                on=["locality", "lat", "lon"],
                how="left",
                suffix="_upd",
            )

            if "state" in df_mapping.columns:
                state_match = (
                    (pl.col("state").cast(pl.Utf8, strict=False) == pl.col("_state").cast(pl.Utf8, strict=False))
                    | (_is_missing_state_expr("state") & _is_missing_state_expr("_state"))
                )
            else:
                state_match = pl.lit(True)

            df_mapping = df_mapping.with_columns([
                pl.when(state_match & pl.col("with_trf_light_day_locality_upd").is_not_null())
                  .then(pl.col("with_trf_light_day_locality_upd"))
                  .otherwise(pl.col("with_trf_light_day_locality"))
                  .alias("with_trf_light_day_locality"),
                pl.when(state_match & pl.col("with_trf_light_night_locality_upd").is_not_null())
                  .then(pl.col("with_trf_light_night_locality_upd"))
                  .otherwise(pl.col("with_trf_light_night_locality"))
                  .alias("with_trf_light_night_locality"),
            ]).drop(["with_trf_light_day_locality_upd", "with_trf_light_night_locality_upd", "_state"])

        # ---- WITHOUT equipment ----
        without_rows = []
        for key, value in crossings_without_traffic_equipment_locality.items():
            parts = key.split("_")
            if len(parts) < 4:
                continue
            locality = parts[0]
            lat = float(parts[1])
            lon = float(parts[2])
            tod = int(parts[3])
            without_rows.append({"locality": locality, "lat": lat, "lon": lon, "_tod": tod, "_val": float(value)})

        if without_rows:
            df_wo = pl.DataFrame(without_rows)

            if "state" in df_mapping.columns:
                state_lookup = (
                    df_mapping
                    .select(["locality", "lat", "lon", "state"])
                    .group_by(["locality", "lat", "lon"])
                    .agg(pl.col("state").drop_nulls().first().alias("_state"))
                )
                df_wo = df_wo.join(state_lookup, on=["locality", "lat", "lon"], how="left")
            else:
                df_wo = df_wo.with_columns(pl.lit(None).cast(pl.Utf8).alias("_state"))

            df_wo = df_wo.with_columns([
                pl.when(pl.col("_tod") == 0).then(pl.col("_val")).otherwise(None).alias("_wo_day"),
                pl.when(pl.col("_tod") == 1).then(pl.col("_val")).otherwise(None).alias("_wo_night"),
            ]).group_by(["locality", "lat", "lon", "_state"]).agg([
                pl.col("_wo_day").drop_nulls().last().alias("without_trf_light_day_locality"),
                pl.col("_wo_night").drop_nulls().last().alias("without_trf_light_night_locality"),
            ])

            df_mapping = df_mapping.join(
                df_wo.select(["locality",
                              "lat",
                              "lon",
                              "_state",
                              "without_trf_light_day_locality",
                              "without_trf_light_night_locality"]),

                on=["locality", "lat", "lon"],
                how="left",
                suffix="_upd",
            )

            if "state" in df_mapping.columns:
                state_match = (
                    (pl.col("state").cast(pl.Utf8, strict=False) == pl.col("_state").cast(pl.Utf8, strict=False))
                    | (_is_missing_state_expr("state") & _is_missing_state_expr("_state"))
                )
            else:
                state_match = pl.lit(True)

            df_mapping = df_mapping.with_columns([
                pl.when(state_match & pl.col("without_trf_light_day_locality_upd").is_not_null())
                  .then(pl.col("without_trf_light_day_locality_upd"))
                  .otherwise(pl.col("without_trf_light_day_locality"))
                  .alias("without_trf_light_day_locality"),
                pl.when(state_match & pl.col("without_trf_light_night_locality_upd").is_not_null())
                  .then(pl.col("without_trf_light_night_locality_upd"))
                  .otherwise(pl.col("without_trf_light_night_locality"))
                  .alias("without_trf_light_night_locality"),
            ]).drop(["without_trf_light_day_locality_upd", "without_trf_light_night_locality_upd", "_state"])

        # ----------------------------------------------------------------------
        # Add country-level crossing counts for with and without traffic equipment
        # ----------------------------------------------------------------------

        # Ensure target columns exist
        needed_country_cols = [
            "with_trf_light_day_country",
            "with_trf_light_night_country",
            "without_trf_light_day_country",
            "without_trf_light_night_country",
        ]
        for c in needed_country_cols:
            if c not in df_mapping.columns:
                df_mapping = df_mapping.with_columns(pl.lit(0.0).alias(c))

        # Build updates for "with"
        with_country_rows = []
        for key, value in crossings_with_traffic_equipment_country.items():
            parts = key.split("_")
            if len(parts) < 2:
                continue
            country = parts[0]
            tod = int(parts[1])  # 0 day, 1 night
            with_country_rows.append({"country": country, "_tod": tod, "_val": float(value)})

        if with_country_rows:
            df_with_country = (
                pl.DataFrame(with_country_rows)
                .with_columns([
                    pl.when(pl.col("_tod") == 0).then(pl.col("_val")).otherwise(None)
                    .alias("with_trf_light_day_country"),
                    pl.when(pl.col("_tod") == 1).then(pl.col("_val")).otherwise(None)
                    .alias("with_trf_light_night_country"),
                ])
                .group_by("country")
                .agg([
                    pl.col("with_trf_light_day_country").drop_nulls().last().alias("with_trf_light_day_country"),
                    pl.col("with_trf_light_night_country").drop_nulls().last().alias("with_trf_light_night_country"),
                ])
            )

            df_mapping = (
                df_mapping
                .join(df_with_country, on="country", how="left", suffix="_upd")
                .with_columns([
                    pl.coalesce([pl.col("with_trf_light_day_country_upd"), pl.col("with_trf_light_day_country")])
                      .alias("with_trf_light_day_country"),
                    pl.coalesce([pl.col("with_trf_light_night_country_upd"), pl.col("with_trf_light_night_country")])
                      .alias("with_trf_light_night_country"),
                ])
                .drop(["with_trf_light_day_country_upd", "with_trf_light_night_country_upd"])
            )

        # Build updates for "without"
        without_country_rows = []
        for key, value in crossings_without_traffic_equipment_country.items():
            parts = key.split("_")
            if len(parts) < 2:
                continue
            country = parts[0]
            tod = int(parts[1])
            without_country_rows.append({"country": country, "_tod": tod, "_val": float(value)})

        if without_country_rows:
            df_without_country = (
                pl.DataFrame(without_country_rows)
                .with_columns([
                    pl.when(pl.col("_tod") == 0).then(pl.col("_val")).otherwise(None)
                    .alias("without_trf_light_day_country"),
                    pl.when(pl.col("_tod") == 1).then(pl.col("_val")).otherwise(None)
                    .alias("without_trf_light_night_country"),
                ])
                .group_by("country")
                .agg([
                    pl.col("without_trf_light_day_country").drop_nulls().last()
                    .alias("without_trf_light_day_country"),
                    pl.col("without_trf_light_night_country").drop_nulls().last()
                    .alias("without_trf_light_night_country"),
                ])
            )

            df_mapping = (
                df_mapping
                .join(df_without_country, on="country", how="left", suffix="_upd")
                .with_columns([
                    pl.coalesce([pl.col("without_trf_light_day_country_upd"),
                                 pl.col("without_trf_light_day_country")])
                      .alias("without_trf_light_day_country"),
                    pl.coalesce([pl.col("without_trf_light_night_country_upd"),
                                 pl.col("without_trf_light_night_country")])
                      .alias("without_trf_light_night_country"),
                ])
                .drop(["without_trf_light_day_country_upd", "without_trf_light_night_country_upd"])
            )

        # ---------------------------------------
        # Add locality-level crossing counts detected
        # ---------------------------------------

        # Ensure locality-level columns exist
        needed_locality_cols = [
            "crossing_detected_locality_day",
            "crossing_detected_locality_night",
            "crossing_detected_locality_all_day",
            "crossing_detected_locality_all_night",
            "crossing_detected_locality",
            "crossing_detected_locality_all",
        ]
        for c in needed_locality_cols:
            if c not in df_mapping.columns:
                df_mapping = df_mapping.with_columns(pl.lit(0.0).alias(c))

        # locality day/night (filtered ids)
        rows_locality = []
        for locality_lat_long_cond, value in pedestrian_cross_locality.items():
            parts = locality_lat_long_cond.split("_")
            if len(parts) < 4:
                continue
            locality, lat_s, lon_s, cond = parts[0], parts[1], parts[2], parts[3]
            try:
                lat = float(lat_s)
            except Exception:
                continue
            if cond == "0":
                rows_locality.append({"locality": locality, "lat": lat, "crossing_detected_locality_day": float(value),
                                      "crossing_detected_locality_night": None})
            elif cond == "1":
                rows_locality.append({"locality": locality, "lat": lat, "crossing_detected_locality_day": None,
                                      "crossing_detected_locality_night": float(value)})

        if rows_locality:
            upd_locality = (
                pl.DataFrame(rows_locality)
                .group_by(["locality", "lat"])
                .agg([
                    pl.col("crossing_detected_locality_day").drop_nulls().last()
                    .alias("crossing_detected_locality_day"),
                    pl.col("crossing_detected_locality_night").drop_nulls().last()
                    .alias("crossing_detected_locality_night"),
                ])
            )

            df_mapping = (
                df_mapping
                .join(upd_locality, on=["locality", "lat"], how="left", suffix="_upd")
                .with_columns([
                    pl.coalesce([pl.col("crossing_detected_locality_day_upd"),
                                 pl.col("crossing_detected_locality_day")])
                      .alias("crossing_detected_locality_day"),
                    pl.coalesce([pl.col("crossing_detected_locality_night_upd"),
                                 pl.col("crossing_detected_locality_night")])
                      .alias("crossing_detected_locality_night"),
                ])
                .drop(["crossing_detected_locality_day_upd", "crossing_detected_locality_night_upd"])
            )

        # locality day/night (all ids)
        rows_locality_all = []
        for locality_lat_long_cond, value in pedestrian_cross_locality_all.items():
            parts = locality_lat_long_cond.split("_")
            if len(parts) < 4:
                continue
            locality, lat_s, lon_s, cond = parts[0], parts[1], parts[2], parts[3]
            try:
                lat = float(lat_s)
            except Exception:
                continue
            if cond == "0":
                rows_locality_all.append({"locality": locality, "lat": lat,
                                          "crossing_detected_locality_all_day": float(value),
                                          "crossing_detected_locality_all_night": None})
            elif cond == "1":
                rows_locality_all.append({"locality": locality, "lat": lat, "crossing_detected_locality_all_day": None,
                                          "crossing_detected_locality_all_night": float(value)})

        if rows_locality_all:
            upd_locality_all = (
                pl.DataFrame(rows_locality_all)
                .group_by(["locality", "lat"])
                .agg([
                    pl.col("crossing_detected_locality_all_day").drop_nulls().last()
                    .alias("crossing_detected_locality_all_day"),
                    pl.col("crossing_detected_locality_all_night").drop_nulls().last()
                    .alias("crossing_detected_locality_all_night"),
                ])
            )

            df_mapping = (
                df_mapping
                .join(upd_locality_all, on=["locality", "lat"], how="left", suffix="_upd")
                .with_columns([
                    pl.coalesce([pl.col("crossing_detected_locality_all_day_upd"),
                                 pl.col("crossing_detected_locality_all_day")])
                      .alias("crossing_detected_locality_all_day"),
                    pl.coalesce([pl.col("crossing_detected_locality_all_night_upd"),
                                 pl.col("crossing_detected_locality_all_night")])
                      .alias("crossing_detected_locality_all_night"),
                ])
                .drop(["crossing_detected_locality_all_day_upd", "crossing_detected_locality_all_night_upd"])
            )

        # Totals (same semantics as fillna(0) + fillna(0))
        df_mapping = df_mapping.with_columns([
            (pl.col("crossing_detected_locality_day").fill_null(0.0) + pl.col("crossing_detected_locality_night")
             .fill_null(0.0)).alias("crossing_detected_locality"),
            (pl.col("crossing_detected_locality_all_day").fill_null(0.0) + pl.col("crossing_detected_locality_all_night")  # noqa: E501
             .fill_null(0.0)).alias("crossing_detected_locality_all"),
        ])

        # ---------------------------------------
        # Add country-level crossing counts detected
        # ---------------------------------------

        # Ensure columns exist (init as 0.0)
        needed_country_detect_cols = [
            "crossing_detected_country_day",
            "crossing_detected_country_night",
            "crossing_detected_country_all_day",
            "crossing_detected_country_all_night",
            "crossing_detected_country",
            "crossing_detected_country_all",
        ]
        for c in needed_country_detect_cols:
            if c not in df_mapping.columns:
                df_mapping = df_mapping.with_columns(pl.lit(0.0).alias(c))

        # Updates: filtered ids
        rows_country = []
        for country_cond, value in pedestrian_cross_country.items():
            try:
                country, cond = country_cond.rsplit("_", 1)
            except ValueError:
                continue

            if cond == "0":
                rows_country.append({"country": country, "crossing_detected_country_day": float(value),
                                     "crossing_detected_country_night": None})
            elif cond == "1":
                rows_country.append({"country": country, "crossing_detected_country_day": None,
                                     "crossing_detected_country_night": float(value)})

        if rows_country:
            upd_country = (
                pl.DataFrame(rows_country)
                .group_by("country")
                .agg([
                    pl.col("crossing_detected_country_day").drop_nulls().last()
                    .alias("crossing_detected_country_day"),
                    pl.col("crossing_detected_country_night").drop_nulls().last()
                    .alias("crossing_detected_country_night"),
                ])
            )

            df_mapping = (
                df_mapping
                .join(upd_country, on="country", how="left", suffix="_upd")
                .with_columns([
                    pl.coalesce([pl.col("crossing_detected_country_day_upd"),
                                 pl.col("crossing_detected_country_day")])
                      .alias("crossing_detected_country_day"),
                    pl.coalesce([pl.col("crossing_detected_country_night_upd"),
                                 pl.col("crossing_detected_country_night")])
                      .alias("crossing_detected_country_night"),
                ])
                .drop(["crossing_detected_country_day_upd", "crossing_detected_country_night_upd"])
            )

        # Updates: all ids
        rows_country_all = []
        for country_cond, value in pedestrian_cross_country_all.items():
            try:
                country, cond = country_cond.rsplit("_", 1)
            except ValueError:
                continue

            if cond == "0":
                rows_country_all.append({"country": country, "crossing_detected_country_all_day": float(value),
                                         "crossing_detected_country_all_night": None})
            elif cond == "1":
                rows_country_all.append({"country": country, "crossing_detected_country_all_day": None,
                                         "crossing_detected_country_all_night": float(value)})

        if rows_country_all:
            upd_country_all = (
                pl.DataFrame(rows_country_all)
                .group_by("country")
                .agg([
                    pl.col("crossing_detected_country_all_day").drop_nulls().last()
                    .alias("crossing_detected_country_all_day"),
                    pl.col("crossing_detected_country_all_night").drop_nulls().last()
                    .alias("crossing_detected_country_all_night"),
                ])
            )

            df_mapping = (
                df_mapping
                .join(upd_country_all, on="country", how="left", suffix="_upd")
                .with_columns([
                    pl.coalesce([pl.col("crossing_detected_country_all_day_upd"),
                                 pl.col("crossing_detected_country_all_day")])
                      .alias("crossing_detected_country_all_day"),
                    pl.coalesce([pl.col("crossing_detected_country_all_night_upd"),
                                 pl.col("crossing_detected_country_all_night")])
                      .alias("crossing_detected_country_all_night"),
                ])
                .drop(["crossing_detected_country_all_day_upd", "crossing_detected_country_all_night_upd"])
            )

        # Totals
        df_mapping = df_mapping.with_columns([
            (pl.col("crossing_detected_country_day").fill_null(0.0) + pl.col("crossing_detected_country_night")
             .fill_null(0.0)).alias("crossing_detected_country"),
            (pl.col("crossing_detected_country_all_day").fill_null(0.0) + pl.col("crossing_detected_country_all_night")
             .fill_null(0.0)).alias("crossing_detected_country_all"),
        ])

        # Add column with count of videos
        # Same semantics as: len(x.strip("[]").split(",")) if x.strip("[]") else 0
        videos_stripped = pl.col("videos").cast(pl.Utf8, strict=False).str.strip_chars("[]").str.strip_chars()
        df_mapping = df_mapping.with_columns(
            pl.when(pl.col("videos").is_null() | (videos_stripped == ""))
              .then(0)
              .otherwise(videos_stripped.str.split(",").list.len())
              .alias("total_videos")
        )

        # Get lat and lon for cities (row-wise; geopy is inherently scalar)
        logger.info("Fetching lat and lon coordinates for cities.")

        # Ensure columns exist
        for c in ["lat", "lon"]:
            if c not in df_mapping.columns:
                df_mapping = df_mapping.with_columns(pl.lit(None).cast(pl.Float64).alias(c))

        lat_list = df_mapping.get_column("lat").to_list()
        lon_list = df_mapping.get_column("lon").to_list()

        locality_list = df_mapping.get_column("locality").to_list()
        state_list = df_mapping.get_column("state").to_list() if "state" in df_mapping.columns else [None] * df_mapping.height  # noqa: E501
        country_list = df_mapping.get_column("country").to_list()

        for i in tqdm(range(df_mapping.height), total=df_mapping.height):
            lat = lat_list[i]
            lon = lon_list[i]
            # Treat None as missing; also treat NaN floats as missing
            lat_missing = (lat is None) or (isinstance(lat, float) and lat != lat)
            lon_missing = (lon is None) or (isinstance(lon, float) and lon != lon)

            if lat_missing or lon_missing:
                new_lat, new_lon = geo.get_coordinates(locality_list[i], state_list[i], common.correct_country(country_list[i]))  # type: ignore  # noqa: E501
                lat_list[i] = new_lat
                lon_list[i] = new_lon

        df_mapping = df_mapping.with_columns([
            pl.Series("lat", lat_list).cast(pl.Float64, strict=False),
            pl.Series("lon", lon_list).cast(pl.Float64, strict=False),
        ])

        # Save the raw file for further investigation
        df_mapping_raw = df_mapping.clone()

        # ----------------------------------------------------------------------
        # df_mapping_raw cleanup + save
        # ----------------------------------------------------------------------

        drop_cols = ['gmp', 'population_locality', 'population_country', 'traffic_mortality',
                     'literacy_rate', 'avg_height', 'med_age', 'gini', 'traffic_index', 'videos',
                     'time_of_day', 'start_time', 'end_time', 'vehicle_type', 'upload_date']

        df_mapping_raw = df_mapping_raw.drop([c for c in drop_cols if c in df_mapping_raw.columns])

        # apply tools.count_unique_channels to channel column
        if "channel" in df_mapping_raw.columns:
            df_mapping_raw = df_mapping_raw.with_columns(
                pl.col("channel").map_elements(
                    tools.count_unique_channels,
                    return_dtype=pl.Int64,  # adjust if your function returns float/str
                ).alias("channel")
            )

        df_mapping_raw.write_csv(os.path.join(common.output_dir, "mapping_locality_raw.csv"))

        # ----------------------------------------------------------------------
        # Filters (population thresholds)
        # ----------------------------------------------------------------------

        population_threshold = common.get_configs("population_threshold")
        min_percentage = common.get_configs("min_locality_population_percentage")

        # Ensure numeric types (coerce invalid to null)
        df_mapping = df_mapping.with_columns([
            pl.col("population_locality").cast(pl.Float64, strict=False).alias("population_locality"),
            pl.col("population_country").cast(pl.Float64, strict=False).alias("population_country"),
        ])

        df_mapping = df_mapping.filter(
            (pl.col("population_locality") >= float(population_threshold))
            | (
                pl.col("population_locality")
                >= float(min_percentage) * pl.col("population_country")
            )
        )

        # Remove the rows of the cities where the footage recorded is less than threshold
        df_mapping = dataset_stats.remove_columns_below_threshold(df_mapping, common.get_configs("footage_threshold"))

        # Limit countries if required
        countries_include = common.get_configs("countries_analyse")
        if countries_include:
            df_mapping = df_mapping.filter(pl.col("iso3").is_in(countries_include))
        df_mapping = _filter_cities(df_mapping, require_match=False)
        log_rollups(df_mapping)

        total_duration = dataset_stats.calculate_total_seconds(df_mapping)

        logger.info(
            f"Duration of videos in seconds after filtering: {total_duration}, in"
            f" minutes after filtering: {total_duration/60:.2f}, in "
            f"hours: {total_duration/60/60:.2f}."
        )

        logger.info("Total number of videos after filtering: {}.",
                    dataset_stats.calculate_total_videos(df_mapping))

        country, number, _ = metrics_cache.get_unique_values(df_mapping, "iso3")
        logger.info(f"Total number of countries and territories after filtering: {number}.")

        locality_state_iso3, number, dup_report = metrics_cache.get_unique_values(
            df_mapping,
            ["locality", "state", "iso3"],
            return_duplicates=True,
        )

        logger.info(f"Total number of unique locality+state+iso3 keys after filtering: {number}.")

        if dup_report is not None and dup_report.height > 0:
            logger.warning(f"Duplicated keys:\n{dup_report}")

        df_mapping = mapping_enrich.add_speed_and_time_to_mapping(
            df_mapping=df_mapping,
            avg_speed_locality=avg_speed_locality,
            avg_speed_country=avg_speed_country,
            avg_time_locality=avg_time_locality,
            avg_time_country=avg_time_country,
            pedestrian_cross_locality=pedestrian_cross_locality,
            pedestrian_cross_country=pedestrian_cross_country,
        )
        df_mapping = _write_bbox_columns(
            df_mapping,
            bbox_averages["speed_locality"],
            bbox_averages["speed_country"],
            bbox_averages["time_locality"],
            bbox_averages["time_country"],
            pedestrian_cross_locality,
            pedestrian_cross_country,
        )

        min_max_speed = duration.get_duration_segment(all_speed, df_mapping, name="speed", duration=None)
        min_max_time = duration.get_duration_segment(all_time, df_mapping, name="time", duration=None)

        # Save the results to a pickle file
        logger.info("Saving results to a pickle file {}.", file_results)
        with open(file_results, "wb") as file:
            pickle.dump(
                (
                    data,                                         # 0
                    person_counter,                               # 1
                    bicycle_counter,                              # 2
                    car_counter,                                  # 3
                    motorcycle_counter,                           # 4
                    bus_counter,                                  # 5
                    truck_counter,                                # 6
                    cellphone_counter,                            # 7
                    traffic_light_counter,                        # 8
                    stop_sign_counter,                            # 9
                    pedestrian_cross_locality,                        # 10
                    pedestrian_crossing_count,                    # 11
                    person_locality,                                  # 12
                    bicycle_locality,                                 # 13
                    car_locality,                                     # 14
                    motorcycle_locality,                              # 15
                    bus_locality,                                     # 16
                    truck_locality,                                   # 17
                    cross_evnt_locality,                              # 18
                    vehicle_locality,                                 # 19
                    cellphone_locality,                               # 20
                    traffic_sign_locality,                            # 21
                    all_speed,                                    # 22
                    all_time,                                     # 23
                    avg_time_locality,                                # 24
                    avg_speed_locality,                               # 25
                    df_mapping,                                   # 26
                    avg_speed_country,                            # 27
                    avg_time_country,                             # 28
                    crossings_with_traffic_equipment_locality,        # 29
                    crossings_without_traffic_equipment_locality,     # 30
                    crossings_with_traffic_equipment_country,     # 31
                    crossings_without_traffic_equipment_country,  # 32
                    min_max_speed,                                # 33
                    min_max_time,                                 # 34
                    pedestrian_cross_country,                     # 35
                    all_speed_locality,                               # 36
                    all_time_locality,                                # 37
                    all_speed_country,                            # 38
                    all_time_country,                             # 39
                    df_mapping_raw,                               # 40
                    pedestrian_cross_locality_all,                    # 41
                    pedestrian_cross_country_all,                 # 42
                ) + (_build_cache_metadata(current_cache_config, segmentation_results),),
                file,
            )

        logger.info("Analysis results saved to pickle file.")

    df_mapping = _apply_segmentation_columns(
        df_mapping,
        segmentation_results,
        pedestrian_cross_locality,
        pedestrian_cross_country,
    )

    # Set index as ID  (Polars has no index; keep semantics by ensuring `id` exists and is first)
    if "id" in df_mapping.columns:
        df_mapping = df_mapping.select(["id"] + [c for c in df_mapping.columns if c != "id"])

        # --- Check if reanalysis of speed is required ---
    if common.get_configs("reanalyse_speed"):
        # Re-derive the per-track speeds when the cached values carry a
        # different unit from the one the installed Waymo model produces.
        # Without this the aggregates below would re-average stale values and
        # the pickle would keep the old unit under a new label.
        target_unit = speed_recompute.metric_speed_unit()
        cached_unit = os.environ.get("CROWD_CROSSING_SPEED_UNIT", "relative")
        if cached_unit != target_unit:
            logger.info(
                "Cached crossing speed is in '{}' but the installed model "
                "produces '{}'; recomputing per-track speeds from the "
                "detection files.",
                cached_unit, target_unit,
            )
            recomputed, track_count, failures = speed_recompute.recompute_all_speed(
                df_mapping=df_mapping,
                wanted=set(pedestrian_crossing_count.keys()),
            )
            if recomputed and not failures:
                all_speed = recomputed
                os.environ["CROWD_CROSSING_SPEED_UNIT"] = target_unit
                logger.info(
                    "Crossing speed recomputed in {} for {} tracks.",
                    target_unit, track_count,
                )
            else:
                logger.error(
                    "Crossing speed recomputation produced {} values with {} "
                    "failed files; keeping the cached '{}' values so the saved "
                    "unit still matches the saved data.",
                    len(recomputed), failures, cached_unit,
                )

        # NOTE: if your Metrics/Mapping_Enrich now expect polars, keep as-is;
        # if any still expects pandas, convert inside those functions (not here).
        avg_speed_country, all_speed_country = metrics.avg_speed_of_crossing_country(df_mapping, all_speed)
        avg_speed_locality, all_speed_locality = metrics.avg_speed_of_crossing_locality(df_mapping, all_speed)
        df_mapping = _write_bbox_columns(
            df_mapping, avg_speed_locality, avg_speed_country, None, None,
            pedestrian_cross_locality, pedestrian_cross_country,
        )
        (
            avg_speed_locality,
            avg_speed_country,
            all_speed_locality,
            all_speed_country,
        ) = _primary_speed(
            segmentation_results,
            avg_speed_locality,
            avg_speed_country,
            all_speed_locality,
            all_speed_country,
        )
        # Re-deriving the per-track speeds replaces every average wholesale,
        # and the metric speed model rejects tracks that the relative index
        # accepted. A locality left without a single valid track is simply
        # absent from the new averages, so its cached column would silently
        # keep the old relative value under the new label. Clear the speed
        # columns first and let the enrichment repopulate only what the
        # recomputed data actually supports.
        speed_columns = [
            column
            for column in (
                "speed_crossing_day_locality",
                "speed_crossing_night_locality",
                "speed_crossing_day_night_locality_avg",
                "speed_crossing_day_country",
                "speed_crossing_night_country",
                "speed_crossing_day_night_country_avg",
            )
            if column in df_mapping.columns
        ]
        if speed_columns:
            df_mapping = df_mapping.with_columns(
                [
                    pl.lit(None).cast(pl.Float64).alias(column)
                    for column in speed_columns
                ]
            )

        df_mapping = mapping_enrich.add_speed_and_time_to_mapping(
            df_mapping=df_mapping,
            avg_speed_locality=avg_speed_locality,
            avg_time_locality=None,
            avg_speed_country=avg_speed_country,
            avg_time_country=None,
            pedestrian_cross_locality=pedestrian_cross_locality,
            pedestrian_cross_country=pedestrian_cross_country,
        )
        # --- Update avg speed values in the pickle file ---
        with open(file_results, "rb") as file:
            results = pickle.load(file)
        results_list = list(results)
        results_list[22] = all_speed          # Update per-track speeds
        results_list[25] = avg_speed_locality     # Update locality speed
        results_list[27] = avg_speed_country  # Update country speed
        results_list[36] = all_speed_locality  # Keep per-track speeds consistent
        results_list[38] = all_speed_country   # Keep per-track speeds consistent
        results_list[26] = df_mapping         # Update mapping (polars)
        with open(file_results, "wb") as file:
            pickle.dump(
                tuple(results_list[:CACHE_RESULTS_COUNT])
                + (_build_cache_metadata(current_cache_config, segmentation_results),),
                file,
            )
        logger.info("Updated speed values in the pickle file.")

    # --- Check if reanalysis of waiting time is required ---
    if common.get_configs("reanalyse_waiting_time"):
        avg_time_country, all_time_country = metrics.avg_time_to_start_cross_country(df_mapping, all_time)
        avg_time_locality, all_time_locality = metrics.avg_time_to_start_cross_locality(df_mapping, all_time)
        df_mapping = _write_bbox_columns(
            df_mapping, None, None, avg_time_locality, avg_time_country,
            pedestrian_cross_locality, pedestrian_cross_country,
        )
        (
            avg_time_locality,
            avg_time_country,
            all_time_locality,
            all_time_country,
        ) = _primary_time(
            segmentation_results,
            avg_time_locality,
            avg_time_country,
            all_time_locality,
            all_time_country,
        )
        df_mapping = mapping_enrich.add_speed_and_time_to_mapping(
            df_mapping=df_mapping,
            avg_time_locality=avg_time_locality,
            avg_speed_locality=avg_speed_locality,
            avg_time_country=avg_time_country,
            avg_speed_country=avg_speed_country,
            pedestrian_cross_locality=pedestrian_cross_locality,
            pedestrian_cross_country=pedestrian_cross_country,
        )
        # --- Update avg time values in the pickle file ---
        with open(file_results, "rb") as file:
            results = pickle.load(file)
        results_list = list(results)
        results_list[24] = avg_time_locality     # Update locality waiting time
        results_list[28] = avg_time_country  # Update country waiting time
        results_list[37] = all_time_locality  # Keep per-track times consistent
        results_list[39] = all_time_country   # Keep per-track times consistent
        results_list[26] = df_mapping        # Update mapping
        with open(file_results, "wb") as file:
            pickle.dump(
                tuple(results_list[:CACHE_RESULTS_COUNT])
                + (_build_cache_metadata(current_cache_config, segmentation_results),),
                file,
            )
        logger.info("Updated time values in the pickle file.")

    # --- Remove countries/cities with insufficient crossing detections ---
    if common.get_configs("min_crossing_detect") != 0:
        threshold: float = float(common.get_configs("min_crossing_detect"))
        country_detect: Dict[str, Dict[str, float]] = {}
        for key, value in pedestrian_cross_country.items():
            country, cond = key.rsplit("_", 1)
            val_f = float(value)
            if country not in country_detect:
                country_detect[country] = {}
            country_detect[country][cond] = val_f

        # Find countries where BOTH conditions are below threshold
        keep_countries: Set[str] = {
            country for country, vals in country_detect.items()
            if (("0" in vals or "1" in vals) and (vals.get("0", 0.0) + vals.get("1", 0.0) >= threshold))
        }

        if keep_countries:
            df_mapping = df_mapping.filter(pl.col("country").is_in(list(keep_countries)))
        else:
            df_mapping = df_mapping.head(0)  # no countries meet threshold

    # Sort by continent and locality, both in ascending order
    df_mapping = df_mapping.sort(["continent", "locality"])

    # Save updated mapping file in output
    os.makedirs(common.output_dir, exist_ok=True)
    df_mapping.write_csv(os.path.join(common.output_dir, "mapping_updated.csv"))

    # The main crossing speed values are already stored in results.pickle.
    # On a cache hit, reuse all_speed and its derived aggregates instead of
    # reopening every bbox file solely for the integrated diagnostic report.
    # Set CROWD_SPEED_REPORT_FORCE=true to explicitly request that expensive
    # diagnostic pass even when the main analysis cache is reused.
    if use_cached_results and not _environment_truthy(
        "CROWD_SPEED_REPORT_FORCE",
        False,
    ):
        cached_speed_tracks = sum(
            len(tracks)
            for videos in (all_speed or {}).values()
            if isinstance(videos, dict)
            for tracks in videos.values()
            if isinstance(tracks, dict)
        )
        logger.info(
            "Using %s cached crossing speed values from results.pickle; "
            "skipping the integrated raw bbox speed report scan.",
            cached_speed_tracks,
        )
    else:
        _run_integrated_speed_reporting(
            df_mapping,
            pedestrian_crossing_count,
        )

    logger.info("Detected:")
    logger.info(f"person: {person_counter}; bicycle: {bicycle_counter}; car: {car_counter}")
    logger.info(f"motorcycle: {motorcycle_counter}; bus: {bus_counter}; truck: {truck_counter}")
    logger.info(f"cellphone: {cellphone_counter}; traffic light: {traffic_light_counter}; " +
                f"traffic sign: {stop_sign_counter}")
    logger.info("Producing output.")

    # Data to avoid showing on hover in scatter plots
    columns_remove = ['videos', 'time_of_day', 'start_time', 'end_time', 'upload_date', 'fps_list', 'vehicle_type',
                      'channel']

    hover_data = [c for c in df_mapping.columns if c not in set(columns_remove)]
    df = df_mapping.clone()  # copy df to manipulate for output

    # Set state to NA (Polars nulls)
    if "state" in df.columns:
        df = df.with_columns(pl.col("state").fill_null("NA").alias("state"))

    # Maps with filtered data
    maps.mapbox_map(df=df.to_pandas(), hover_data=hover_data, file_name='mapbox_map')

    maps.mapbox_map(df=df.to_pandas(),
                    hover_data=hover_data,
                    density_col='total_time',
                    density_radius=10,
                    file_name='mapbox_map_time')

    maps.world_map(df_mapping=df.to_pandas())  # map with countries

    distribution.violin_plot(data_index=22,
                             name="speed",
                             min_threshold=common.get_configs("min_speed_limit"),
                             max_threshold=common.get_configs("max_speed_limit"), df_mapping=df_mapping,
                             xaxis_title=crossing_metric_label(),
                             save_file=True, data_file=file_results)

    # ------------All values----------------- #
    distribution.hist(data_index=22,
                      name="speed",
                      xaxis_title=crossing_metric_label(),
                      marginal="violin",
                      nbins=100,
                      raw=True,
                      min_threshold=common.get_configs("min_speed_limit"),
                      max_threshold=common.get_configs("max_speed_limit"),
                      font_size=common.get_configs("font_size") + 4,
                      fig_save_height=650,
                      save_file=True,
                      data_file=file_results)

    distribution.hist(data_index=39,
                      name="time",
                      xaxis_title="Crossing initiation time (s)",
                      marginal="violin",
                      # nbins=100,
                      raw=True,
                      min_threshold=None,
                      max_threshold=None,
                      font_size=common.get_configs("font_size") + 4,
                      fig_save_height=650,
                      save_file=True,
                      data_file=file_results)

    # ------------Filtered values----------------- #
    distribution.hist(data_index=38,
                      name="speed_filtered",
                      xaxis_title=crossing_metric_label(),
                      marginal="violin",
                      nbins=100,
                      raw=False,
                      min_threshold=common.get_configs("min_speed_limit"),
                      max_threshold=common.get_configs("max_speed_limit"),
                      font_size=common.get_configs("font_size") + 4,
                      fig_save_height=650,
                      save_file=True,
                      data_file=file_results)

    distribution.hist(data_index=37,
                      name="time_filtered",
                      xaxis_title="Crossing initiation time (s)",
                      marginal="violin",
                      # nbins=100,
                      raw=False,
                      min_threshold=None,
                      max_threshold=None,
                      font_size=common.get_configs("font_size") + 4,
                      df_mapping=df_mapping,
                      fig_save_height=650,
                      save_file=True,
                      data_file=file_results)

    analysis_level = str(common.get_configs("analysis_level")).strip().lower()
    if analysis_level in {"city", "locality"}:

        # Amount of footage
        bivariate.scatter(df=df,
                          x="total_time",
                          y="person",
                          color="continent",
                          text="locality",
                          xaxis_title='Total time of footage (s)',
                          yaxis_title='Number of detected pedestrians',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="locality",
                          legend_title="",
                          legend_x=0.01,
                          legend_y=1.0,
                          label_distance_factor=5.0,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # todo: ISO-3 codes next to figures shift. need to correct once "final" dataset is online
        crossing.speed_and_time_to_start_cross(df_mapping,
                                               x_axis_title_height=110,
                                               font_size_captions=common.get_configs("font_size") + 8,
                                               legend_x=0.9,
                                               legend_y=0.01,
                                               legend_spacing=0.0026)

        stacked.stack_plot(df,
                           order_by="alphabetical",
                           metric="time",
                           data_view="combined",
                           title_text="Crossing initiation time (s)",
                           filename="time_crossing_alphabetical",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=80,
                           right_margin=80
                           )

        stacked.stack_plot(df,
                           order_by="alphabetical",
                           metric="time",
                           data_view="day",
                           title_text="Crossing initiation time (s)",
                           filename="time_crossing_alphabetical_day",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=80,
                           right_margin=80
                           )

        stacked.stack_plot(df,
                           order_by="alphabetical",
                           metric="time",
                           data_view="night",
                           title_text="Crossing initiation time (s)",
                           filename="time_crossing_alphabetical_night",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=80,
                           right_margin=80
                           )

        stacked.stack_plot(df,
                           order_by="average",
                           metric="time",
                           data_view="combined",
                           title_text="Crossing initiation time (s)",
                           filename="time_crossing_avg",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=10,
                           right_margin=10
                           )

        stacked.stack_plot(df,
                           order_by="average",
                           metric="time",
                           data_view="day",
                           title_text="Crossing initiation time (s)",
                           filename="time_crossing_avg_day",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=10,
                           right_margin=10
                           )

        stacked.stack_plot(df,
                           order_by="average",
                           metric="time",
                           data_view="night",
                           title_text="Crossing initiation time (s)",
                           filename="time_crossing_avg_night",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=10,
                           right_margin=10
                           )

        stacked.stack_plot(df,
                           order_by="alphabetical",
                           metric="speed",
                           data_view="combined",
                           title_text=crossing_metric_label(mean=True),
                           filename="speed_crossing_alphabetical",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=80,
                           right_margin=80
                           )

        stacked.stack_plot(df,
                           order_by="alphabetical",
                           metric="speed",
                           data_view="day",
                           title_text=crossing_metric_label(mean=True),
                           filename="speed_crossing_alphabetical_day",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=80,
                           right_margin=80
                           )

        stacked.stack_plot(df,
                           order_by="alphabetical",
                           metric="speed",
                           data_view="night",
                           title_text=crossing_metric_label(mean=True),
                           filename="speed_crossing_alphabetical_night",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=80,
                           right_margin=80
                           )

        stacked.stack_plot(df,
                           order_by="average",
                           metric="speed",
                           data_view="combined",
                           title_text=crossing_metric_label(mean=True),
                           filename="speed_crossing_avg",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=10,
                           right_margin=10
                           )

        stacked.stack_plot(df,
                           order_by="average",
                           metric="speed",
                           data_view="day",
                           title_text=crossing_metric_label(mean=True),
                           filename="speed_crossing_avg_day",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=10,
                           right_margin=10
                           )

        stacked.stack_plot(df,
                           order_by="average",
                           metric="speed",
                           data_view="night",
                           title_text=crossing_metric_label(mean=True),
                           filename="speed_crossing_avg_night",
                           font_size_captions=common.get_configs("font_size") + 8,
                           left_margin=10,
                           right_margin=10
                           )

        correlation.correlation_matrix(df_mapping, pedestrian_cross_locality, person_locality, bicycle_locality,
                                       car_locality, motorcycle_locality, bus_locality, truck_locality,
                                       cross_evnt_locality, vehicle_locality, cellphone_locality,
                                       traffic_sign_locality, all_speed, all_time, avg_time_locality,
                                       avg_speed_locality)

        # ---------------------------------------------------------------------
        # City-level structure diagnostics
        #
        # A k-means partition always returns k groups, so before any grouping
        # is reported the data are tested for cluster tendency (Hopkins, gap
        # statistic, silhouette). The same pass reports the gradient
        # correlations, the split-half reliability of each city estimate, and
        # the two measurement artefacts that constrain interpretation: the
        # initiation-time floor and the within-video normalisation of the
        # crossing speed. Every value is logged as well as plotted.
        # ---------------------------------------------------------------------
        try:
            structure_result = analyse_structure(
                df_mapping=df_mapping,
                avg_speed_locality=avg_speed_locality,
                avg_time_locality=avg_time_locality,
                pedestrian_cross_locality=pedestrian_cross_locality,
                crossings_with_traffic_equipment_locality=(
                    crossings_with_traffic_equipment_locality
                ),
                crossings_without_traffic_equipment_locality=(
                    crossings_without_traffic_equipment_locality
                ),
                all_speed=all_speed,
                all_time=all_time,
                all_speed_locality=all_speed_locality,
                all_time_locality=all_time_locality,
                checks_per_second=float(common.get_configs("check_per_sec_time")),
            )
            if structure_result.get("status") == "complete":
                structure_result["city_table"].write_csv(
                    os.path.join(common.output_dir, "structure_city_level.csv")
                )
                structure_plots.plot_all(structure_result)
        except Exception as error:
            logger.error(f"Structure diagnostics failed: {error}")

        # ---------------------------------------------------------------------
        # Locality-level bivariate plots
        #
        # The mapping enrichment pipeline now creates explicit locality-level
        # columns such as:
        #   speed_crossing_day_locality
        #   speed_crossing_night_locality
        #   speed_crossing_day_night_locality_avg
        #   time_crossing_day_locality
        #   time_crossing_night_locality
        #   time_crossing_day_night_locality_avg
        #
        # This block intentionally stays Polars-native. Older versions of this
        # section used legacy column names and Pandas-style DataFrame operations.
        # ---------------------------------------------------------------------

        LOCALITY_SPEED = "speed_crossing_day_night_locality_avg"
        LOCALITY_TIME = "time_crossing_day_night_locality_avg"
        LOCALITY_SPEED_DAY = "speed_crossing_day_locality"
        LOCALITY_SPEED_NIGHT = "speed_crossing_night_locality"
        LOCALITY_TIME_DAY = "time_crossing_day_locality"
        LOCALITY_TIME_NIGHT = "time_crossing_night_locality"

        def _locality_scatter(
            x: str,
            y: str,
            *,
            xaxis_title: str,
            yaxis_title: str,
            legend_x: float = 0.87,
            legend_y: float = 1.0,
            label_distance_factor: float = 3.0,
            source_df: Optional[pl.DataFrame] = None,
        ) -> None:
            # Create a locality scatter plot using current Polars columns.
            base = df_mapping if source_df is None else source_df

            required = [x, y]
            missing = [column for column in required if column not in base.columns]
            if missing:
                logger.warning(
                    "Skipping locality scatter x={} y={}; missing columns: {}.",
                    x,
                    y,
                    missing,
                )
                return

            # Cast only the plotted values so numeric strings and nullable
            # numeric columns are handled consistently.
            plot_df = base.with_columns(
                [
                    pl.col(x).cast(pl.Float64, strict=False).alias(x),
                    pl.col(y).cast(pl.Float64, strict=False).alias(y),
                ]
            )

            plot_df = plot_df.filter(
                pl.col(x).is_not_null()
                & pl.col(y).is_not_null()
                & pl.col(x).is_finite()
                & pl.col(y).is_finite()
                & (pl.col(x) != 0)
                & (pl.col(y) != 0)
            )

            if "state" in plot_df.columns:
                plot_df = plot_df.with_columns(
                    pl.col("state").fill_null("NA").alias("state")
                )

            bivariate.scatter(
                df=plot_df,
                x=x,
                y=y,
                color="continent",
                text="locality",
                xaxis_title=xaxis_title,
                yaxis_title=yaxis_title,
                pretty_text=False,
                marker_size=10,
                save_file=True,
                hover_data=hover_data,
                hover_name="locality",
                legend_title="",
                legend_x=legend_x,
                legend_y=legend_y,
                label_distance_factor=label_distance_factor,
                marginal_x=None,  # type: ignore
                marginal_y=None,  # type: ignore
            )

        # Mean crossing speed vs crossing initiation time
        _locality_scatter(
            LOCALITY_SPEED,
            LOCALITY_TIME,
            xaxis_title=crossing_metric_label(),
            yaxis_title="Crossing initiation time (in s)",
            legend_x=0.01,
            legend_y=1.0,
            label_distance_factor=3.0,
        )

        # Daytime crossing speed vs daytime crossing initiation time
        _locality_scatter(
            LOCALITY_SPEED_DAY,
            LOCALITY_TIME_DAY,
            xaxis_title=f"{crossing_metric_label()} during daytime",
            yaxis_title="Crossing initiation time during daytime (in s)",
            legend_x=0.01,
            legend_y=1.0,
            label_distance_factor=3.0,
        )

        # Night-time crossing speed vs night-time crossing initiation time
        _locality_scatter(
            LOCALITY_SPEED_NIGHT,
            LOCALITY_TIME_NIGHT,
            xaxis_title=f"{crossing_metric_label()} during night time",
            yaxis_title="Crossing initiation time during night time (in s)",
            legend_x=0.87,
            legend_y=1.0,
            label_distance_factor=0.8,
        )

        # Crossing initiation time vs locality population
        _locality_scatter(
            LOCALITY_TIME,
            "population_locality",
            xaxis_title="Crossing initiation time (in s)",
            yaxis_title="Population of locality",
            label_distance_factor=2.0,
        )

        # Crossing motion vs locality population
        _locality_scatter(
            LOCALITY_SPEED,
            "population_locality",
            xaxis_title=crossing_metric_label(mean=True),
            yaxis_title="Population of locality",
            label_distance_factor=3.0,
        )

        # Crossing initiation time vs national traffic mortality
        _locality_scatter(
            LOCALITY_TIME,
            "traffic_mortality",
            xaxis_title="Crossing initiation time (in s)",
            yaxis_title="National traffic mortality rate (per 100,000 of population)",
            label_distance_factor=3.0,
        )

        # Crossing motion vs national traffic mortality
        _locality_scatter(
            LOCALITY_SPEED,
            "traffic_mortality",
            xaxis_title=crossing_metric_label(mean=True),
            yaxis_title="National traffic mortality rate (per 100,000 of population)",
            label_distance_factor=2.0,
        )

        # Crossing initiation time vs literacy rate
        _locality_scatter(
            LOCALITY_TIME,
            "literacy_rate",
            xaxis_title="Crossing initiation time (in s)",
            yaxis_title="Literacy rate",
            legend_x=0.87,
            legend_y=0.01,
            label_distance_factor=3.0,
        )

        # Crossing motion vs literacy rate
        _locality_scatter(
            LOCALITY_SPEED,
            "literacy_rate",
            xaxis_title=crossing_metric_label(mean=True),
            yaxis_title="Literacy rate",
            legend_x=0.87,
            legend_y=0.01,
            label_distance_factor=3.0,
        )

        # Crossing initiation time vs Gini coefficient
        _locality_scatter(
            LOCALITY_TIME,
            "gini",
            xaxis_title="Crossing initiation time (in s)",
            yaxis_title="Gini coefficient",
            label_distance_factor=3.0,
        )

        # Crossing motion vs Gini coefficient
        _locality_scatter(
            LOCALITY_SPEED,
            "gini",
            xaxis_title=crossing_metric_label(mean=True),
            yaxis_title="Gini coefficient",
            label_distance_factor=2.0,
        )

        # Crossing initiation time vs traffic index
        _locality_scatter(
            LOCALITY_TIME,
            "traffic_index",
            xaxis_title="Crossing initiation time (in s)",
            yaxis_title="Traffic index",
            label_distance_factor=2.0,
        )

        # Crossing motion vs traffic index
        _locality_scatter(
            LOCALITY_SPEED,
            "traffic_index",
            xaxis_title=crossing_metric_label(mean=True),
            yaxis_title="Traffic index",
            label_distance_factor=2.0,
        )

        # Mobile-phone detections normalised by footage time
        if {"cellphone", "total_time"} <= set(df_mapping.columns):
            cellphone_df = df_mapping.with_columns(
                pl.when(
                    pl.col("total_time").cast(pl.Float64, strict=False) > 0
                )
                .then(
                    pl.col("cellphone").cast(pl.Float64, strict=False)
                    / pl.col("total_time").cast(pl.Float64, strict=False)
                )
                .otherwise(None)
                .alias("cellphone_normalised")
            )

            _locality_scatter(
                LOCALITY_TIME,
                "cellphone_normalised",
                xaxis_title="Crossing initiation time (in s)",
                yaxis_title="Mobile phones detected (normalised over time)",
                label_distance_factor=3.0,
                source_df=cellphone_df,
            )

            _locality_scatter(
                LOCALITY_SPEED,
                "cellphone_normalised",
                xaxis_title=crossing_metric_label(mean=True),
                yaxis_title="Mobile phones detected (normalised over time)",
                label_distance_factor=3.0,
                source_df=cellphone_df,
            )
        else:
            logger.warning(
                "Skipping cellphone-normalised locality plots because "
                "'cellphone' or 'total_time' is missing."
            )

        # Jaywalking / traffic-light plots
        crossing.plot_crossing_without_traffic_light(
            df_mapping,
            x_axis_title_height=60,
            font_size_captions=common.get_configs("font_size"),
            legend_x=0.97,
            legend_y=1.0,
            legend_spacing=0.004,
        )
        crossing.plot_crossing_with_traffic_light(
            df_mapping,
            x_axis_title_height=60,
            font_size_captions=common.get_configs("font_size"),
            legend_x=0.97,
            legend_y=1.0,
            legend_spacing=0.004,
        )

        # Crossing events with and without traffic lights, normalised by
        # footage time and locality population.
        traffic_light_columns = {
            "with_trf_light_day_locality",
            "with_trf_light_night_locality",
            "without_trf_light_day_locality",
            "without_trf_light_night_locality",
            "total_time",
            "population_locality",
        }

        missing_traffic_light_columns = sorted(
            traffic_light_columns - set(df_mapping.columns)
        )

        if not missing_traffic_light_columns:
            total_time_expr = pl.col("total_time").cast(
                pl.Float64, strict=False
            )
            population_expr = pl.col("population_locality").cast(
                pl.Float64, strict=False
            )
            denominator = total_time_expr * population_expr

            traffic_light_df = df_mapping.with_columns(
                [
                    pl.when(denominator > 0)
                    .then(
                        (
                            pl.col("with_trf_light_day_locality")
                            .cast(pl.Float64, strict=False)
                            .fill_null(0)
                            + pl.col("with_trf_light_night_locality")
                            .cast(pl.Float64, strict=False)
                            .fill_null(0)
                        )
                        / denominator
                    )
                    .otherwise(None)
                    .alias("with_trf_light_norm"),
                    pl.when(denominator > 0)
                    .then(
                        (
                            pl.col("without_trf_light_day_locality")
                            .cast(pl.Float64, strict=False)
                            .fill_null(0)
                            + pl.col("without_trf_light_night_locality")
                            .cast(pl.Float64, strict=False)
                            .fill_null(0)
                        )
                        / denominator
                    )
                    .otherwise(None)
                    .alias("without_trf_light_norm"),
                ]
            )

            _locality_scatter(
                "with_trf_light_norm",
                "without_trf_light_norm",
                xaxis_title="Crossing events with traffic lights (normalised)",
                yaxis_title="Crossing events without traffic lights (normalised)",
                label_distance_factor=3.0,
                source_df=traffic_light_df,
            )
        else:
            logger.warning(
                "Skipping locality traffic-light normalisation plot; "
                "missing columns: {}.",
                missing_traffic_light_columns,
            )

    if analysis_level == "country":
        df_countries = aggregation.aggregate_by_iso3(df_mapping)
        df_countries_raw = aggregation.aggregate_by_iso3(df_mapping_raw)

        # ---------------------------------------------------------------------
        # Hover columns (keep order stable; exclude configured columns)
        # ---------------------------------------------------------------------
        columns_remove = ['videos', 'time_of_day', 'start_time', 'end_time', 'upload_date', 'fps_list', 'vehicle_type']
        remove_set = set(columns_remove)
        hover_data = [c for c in df_countries.columns if c not in remove_set]

        columns_remove_raw = ['gini', 'traffic_mortality', 'avg_height', 'population_country', 'population_locality',
                              'med_age', 'literacy_rate']
        remove_raw_set = remove_set | set(columns_remove_raw)
        hover_data_raw = [c for c in df_countries.columns if c not in remove_raw_set]

        # ---------------------------------------------------------------------
        # Save via pandas (handles nested data by serializing to string/object)
        # ---------------------------------------------------------------------
        os.makedirs(common.output_dir, exist_ok=True)
        df_countries.to_pandas().to_csv(os.path.join(common.output_dir, "mapping_countries.csv"), index=False)

        maps.map_world(df=df_countries_raw.to_pandas(),
                       color="continent",                # same default as map_political
                       show_cities=True,
                       df_cities=df_mapping.to_pandas(),
                       show_images=True,
                       hover_data=hover_data_raw,
                       save_file=True,
                       save_final=False,
                       file_name="raw_map")

        # Map with screenshots and countries colours by continent
        maps.map_world(df=df_countries.to_pandas(),
                       color="continent",
                       show_cities=True,
                       df_cities=df_mapping.to_pandas(),
                       show_images=True,
                       hover_data=hover_data,
                       save_file=False,
                       save_final=False,
                       file_name="map_screenshots",
                       show_colorbar=True,
                       colorbar_title="Continent",
                       colorbar_kwargs=dict(y=0.035, len=0.55, bgcolor="rgba(255,255,255,0.9)"))

        # Map with screenshots and countries colours by amount of footage
        remove_set = set(columns_remove)
        hover_data = [c for c in df_countries_raw.columns if c not in remove_set]

        # log(1 + x) to avoid -inf for zero
        df_countries_raw = df_countries_raw.with_columns(
            pl.col("total_time")
              .fill_null(0)
              .cast(pl.Float64)
              .log1p()
              .alias("log_total_time")
        )

        # Produce map with all data
        df = df_mapping_raw.clone()  # copy df to manipulate for output
        df = df.with_columns(
            pl.col("state").fill_null("NA").alias("state")
        )

        # Sort by continent and locality, both in ascending order
        df = df.sort(["continent", "locality"])

        maps.map_world(df=df_countries_raw.to_pandas(),
                       color="log_total_time",
                       show_cities=True,
                       df_cities=df_mapping.to_pandas(),
                       show_images=True,
                       hover_data=hover_data,
                       show_colorbar=True,
                       colorbar_title="Footage (log)",
                       save_file=True,
                       save_final=False,
                       file_name="map_screenshots_total_time")

        # Drop columns (only if present)
        drop_cols = [
            "speed_crossing_day_country",
            "speed_crossing_night_country",
            "speed_crossing_day_night_country_avg",
            "time_crossing_day_country",
            "time_crossing_night_country",
            "time_crossing_day_night_country_avg",
        ]
        df_countries_raw = df_countries_raw.drop([c for c in drop_cols if c in df_countries_raw.columns])

        # Save via pandas (CSV-friendly for nested columns)
        os.makedirs(common.output_dir, exist_ok=True)
        df_countries_raw.to_pandas().to_csv(
            os.path.join(common.output_dir, "mapping_countries_raw.csv"),
            index=False
        )

        # Amount of footage
        bivariate.scatter(df=df_countries,
                          x="total_time",
                          y="person",
                          extension=common.get_configs("analysis_level"),
                          color="continent",
                          text="iso3",
                          xaxis_title='Total time of footage (s)',
                          yaxis_title='Number of detected pedestrians',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.01,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Amount of bicycle footage normalised
        df = (df_countries.filter(pl.col("person") != 0).with_columns(
                (pl.col("person") / pl.col("total_time")).alias("person_norm")))

        bivariate.scatter(df=df,
                          x="total_time",
                          y="person_norm",
                          color="continent",
                          text="iso3",
                          xaxis_title='Total time of footage (s)',
                          yaxis_title='Number of detected pedestrians (normalised over amount of footage)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.94,
                          legend_y=1.0,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Amount of bicycle footage normalised
        df = (df_countries.filter(pl.col("bicycle") != 0).with_columns(
                (pl.col("bicycle") / pl.col("total_time")).alias("bicycle_norm")))

        bivariate.scatter(df=df,
                          x="total_time",
                          y="bicycle_norm",
                          color="continent",
                          text="iso3",
                          xaxis_title='Total time of footage (s)',
                          yaxis_title='Number of detected bicycle (normalised over amount of footage)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.94,
                          legend_y=1.0,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        stacked.stack_plot_country(df_countries,
                                   order_by="average",
                                   metric="time",
                                   data_view="combined",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_avg_country",
                                   font_size_captions=common.get_configs("font_size") + 8,
                                   legend_x=0.95,
                                   legend_y=0.04,
                                   legend_spacing=0.02,
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="condition",
                                   metric="speed",
                                   data_view="combined",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_combined_country",
                                   font_size_captions=common.get_configs("font_size") + 28,
                                   legend_x=0.92,
                                   legend_y=0.04,
                                   legend_spacing=0.02,
                                   top_margin=150,
                                   height=2450,
                                   width=2480)

        stacked.stack_plot_country(df_countries,
                                   order_by="condition",
                                   metric="time",
                                   data_view="combined",
                                   title_text="Mean crossing initiation time (in s)",
                                   filename="time_crossing_combined_country",
                                   font_size_captions=common.get_configs("font_size") + 28,
                                   legend_x=0.92,
                                   legend_y=0.04,
                                   legend_spacing=0.02,
                                   top_margin=150,
                                   height=2400,
                                   width=2480)

        stacked.stack_plot_country(df_countries_raw,
                                   order_by="condition",
                                   metric="speed",
                                   data_view="combined",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_combined_country_raw",
                                   font_size_captions=common.get_configs("font_size") + 28,
                                   raw=True,
                                   legend_x=0.92,
                                   legend_y=0.04,
                                   legend_spacing=0.02,
                                   top_margin=150,
                                   height=2400,
                                   width=2480)

        stacked.stack_plot_country(df_countries_raw,
                                   order_by="condition",
                                   metric="time",
                                   data_view="combined",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_combined_country_raw",
                                   font_size_captions=common.get_configs("font_size") + 28,
                                   raw=True,
                                   legend_x=0.92,
                                   legend_y=0.04,
                                   legend_spacing=0.02,
                                   top_margin=150,
                                   height=2400,
                                   width=2480)

        stacked.stack_plot_country(df_countries,
                                   order_by="alphabetical",
                                   metric="time",
                                   data_view="combined",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_alphabetical_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   legend_x=0.94,
                                   legend_y=0.03,
                                   legend_spacing=0.02,
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="average",
                                   metric="speed",
                                   data_view="combined",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_avg_country",
                                   font_size_captions=common.get_configs("font_size") + 8,
                                   legend_x=0.87,
                                   legend_y=0.04,
                                   legend_spacing=0.02,
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="alphabetical",
                                   metric="speed",
                                   data_view="combined",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_alphabetical_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   legend_x=0.94,
                                   legend_y=0.03,
                                   legend_spacing=0.02,
                                   top_margin=100)

        # Plotting stacked plot during day
        stacked.stack_plot_country(df_countries,
                                   order_by="average",
                                   metric="time",
                                   data_view="day",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_avg_day_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="alphabetical",
                                   metric="time",
                                   data_view="day",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_alphabetical_day_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="average",
                                   metric="speed",
                                   data_view="day",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_avg_day_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="alphabetical",
                                   metric="speed",
                                   data_view="day",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_alphabetical_day_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        # Plotting stacked plot during night
        stacked.stack_plot_country(df_countries,
                                   order_by="average",
                                   metric="time",
                                   data_view="night",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_avg_night_country",
                                   font_size_captions=common.get_configs("font_size"))

        stacked.stack_plot_country(df_countries,
                                   order_by="alphabetical",
                                   metric="time",
                                   data_view="night",
                                   title_text="Crossing initiation time (s)",
                                   filename="time_crossing_alphabetical_night_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="average",
                                   metric="speed",
                                   data_view="night",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_avg_night_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        stacked.stack_plot_country(df_countries,
                                   order_by="alphabetical",
                                   metric="speed",
                                   data_view="night",
                                   title_text=crossing_metric_label(mean=True),
                                   filename="crossing_speed_alphabetical_night_country",
                                   font_size_captions=common.get_configs("font_size"),
                                   top_margin=100)

        crossing.speed_and_time_to_start_cross_country(df_countries,
                                                       x_axis_title_height=110,
                                                       font_size_captions=common.get_configs("font_size") + 8,
                                                       legend_x=0.87,
                                                       legend_y=0.04,
                                                       legend_spacing=0.01)

        correlation.correlation_matrix_country(df_mapping, df_countries, pedestrian_cross_locality, person_locality,
                                               bicycle_locality, car_locality, motorcycle_locality, bus_locality,
                                               truck_locality, cross_evnt_locality, vehicle_locality,
                                               cellphone_locality, traffic_sign_locality, avg_speed_country,
                                               avg_time_country, crossings_without_traffic_equipment_country)

        # Speed of crossing vs Crossing initiation time
        df = (
            df_countries
            .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
            .filter(pl.col("time_crossing_day_night_country_avg") != 0)
        )
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="time_crossing_day_night_country_avg",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='Crossing initiation time (s)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing during daytime vs time to start crossing during daytime
        df = (df_countries
              .filter(pl.col("speed_crossing_day_country") != 0)
              .filter(pl.col("time_crossing_day_country") != 0))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_country",
                          y="time_crossing_day_country",
                          color="continent",
                          text="iso3",
                          xaxis_title=f"{crossing_metric_label()} during daytime",
                          yaxis_title='Crossing initiation time during daytime (in s)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing during night time vs time to start crossing during night time
        df = (df_countries
              .filter(pl.col("speed_crossing_night_country") != 0)
              .filter(pl.col("time_crossing_night_country") != 0))
        bivariate.scatter(df=df,
                          x="speed_crossing_night_country",
                          y="time_crossing_night_country",
                          color="continent",
                          text="iso3",
                          xaxis_title=f"{crossing_metric_label()} during night time",
                          yaxis_title='Crossing initiation time during night time (in s)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Time to start crossing vs population of locality
        df = (df_countries
              .filter(pl.col("time_crossing_day_night_country_avg") != 0)
              .filter(pl.col("population_country").is_not_null() & (pl.col("population_country") != 0)))
        bivariate.scatter(df=df,
                          x="time_crossing_day_night_country_avg",
                          y="population_country",
                          color="continent",
                          text="iso3",
                          xaxis_title='Crossing initiation time (s)',
                          yaxis_title='Population of country',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs population of country
        df = (df_countries
              .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
              .filter(pl.col("population_country").is_not_null() & (pl.col("population_country") != 0)))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="population_country",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='Population of country',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.2,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Time to start crossing vs population of locality
        df = (df_countries
              .filter(pl.col("time_crossing_day_night_country_avg") != 0)
              .filter(pl.col("traffic_mortality").is_not_null() & (pl.col("traffic_mortality") != 0)))
        bivariate.scatter(df=df,
                          x="time_crossing_day_night_country_avg",
                          y="traffic_mortality",
                          color="continent",
                          text="iso3",
                          xaxis_title='Crossing initiation time (in s)',
                          yaxis_title='National traffic mortality rate (per 100,000 of population)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs population of locality
        df = (df_countries
              .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
              .filter(pl.col("traffic_mortality").is_not_null() & (pl.col("traffic_mortality") != 0)))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="traffic_mortality",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='National traffic mortality rate (per 100,000 of population)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.3,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Time to start crossing vs population of locality
        df = (df_countries
              .filter(pl.col("time_crossing_day_night_country_avg") != 0)
              .filter(pl.col("literacy_rate").is_not_null() & (pl.col("literacy_rate") != 0)))
        bivariate.scatter(df=df,
                          x="time_crossing_day_night_country_avg",
                          y="literacy_rate",
                          color="continent",
                          text="iso3",
                          xaxis_title='Crossing initiation time (in s)',
                          yaxis_title='Literacy rate',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=0.01,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs population of locality
        df = (df_countries
              .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
              .filter(pl.col("literacy_rate").is_not_null() & (pl.col("literacy_rate") != 0)))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="literacy_rate",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='Literacy rate',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=0.01,
                          label_distance_factor=0.4,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Time to start crossing vs population of locality
        df = (df_countries
              .filter(pl.col("time_crossing_day_night_country_avg") != 0)
              .filter(pl.col("gini").is_not_null() & (pl.col("gini") != 0)))
        bivariate.scatter(df=df,
                          x="time_crossing_day_night_country_avg",
                          y="gini",
                          color="continent",
                          text="iso3",
                          xaxis_title='Crossing initiation time (in s)',
                          yaxis_title='Gini coefficient',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs population of locality
        df = (df_countries
              .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
              .filter(pl.col("gini").is_not_null() & (pl.col("gini") != 0)))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="gini",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='Gini coefficient',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Time to start crossing vs population of locality
        df = (df_countries
              .filter(pl.col("time_crossing_day_night_country_avg") != 0)
              .filter(pl.col("med_age").is_not_null() & (pl.col("med_age") != 0)))
        bivariate.scatter(df=df,
                          x="time_crossing_day_night_country_avg",
                          y="med_age",
                          color="continent",
                          text="iso3",
                          # size="gmp",
                          xaxis_title='Crossing initiation time (in s)',
                          yaxis_title='Median age (in years)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs population of locality
        df = (df_countries
              .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
              .filter(pl.col("med_age") != 0))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="med_age",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='Median age (in years)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.4,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs detected mobile phones
        df = (df_countries
              .filter(pl.col("time_crossing_day_night_country_avg") != 0)
              .with_columns(
                (pl.col("cellphone") / pl.col("total_time")).alias("cellphone_normalised")))
        bivariate.scatter(df=df,
                          x="time_crossing_day_night_country_avg",
                          y="cellphone_normalised",
                          color="continent",
                          text="iso3",
                          xaxis_title='Crossing initiation time (in s)',
                          yaxis_title='Mobile phones detected (normalised over time)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Speed of crossing vs detected mobile phones
        df = (df_countries
              .filter(pl.col("speed_crossing_day_night_country_avg") != 0)
              .with_columns(
                (pl.col("cellphone") / pl.col("total_time")).alias("cellphone_normalised")))
        bivariate.scatter(df=df,
                          x="speed_crossing_day_night_country_avg",
                          y="cellphone_normalised",
                          color="continent",
                          text="iso3",
                          xaxis_title=crossing_metric_label(mean=True),
                          yaxis_title='Mobile phones detected (normalised over time)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        # Mean speed of crossing (used to be plots_class.map)
        maps.map_world(df=df_countries.to_pandas(),
                       color="speed_crossing_day_night_country_avg",
                       title=crossing_metric_label(mean=True),
                       show_colorbar=True,
                       colorbar_title="",                 # keep your empty title behavior
                       filter_zero_nan=True,              # preserves old map() filtering
                       save_file=True,
                       file_name="map_speed_crossing"
                       )

        # Crossing initiation time (used to be plots_class.map)
        maps.map_world(df=df_countries.to_pandas(),
                       color="time_crossing_day_night_country_avg",
                       title="Crossing initiation time (in s)",
                       show_colorbar=True,
                       colorbar_title="",
                       filter_zero_nan=True,
                       save_file=True,
                       file_name="map_crossing_time"
                       )

        # Crossing with and without traffic lights
        df = (df_countries.with_columns([
                ((pl.col("with_trf_light_day_country") + pl.col("with_trf_light_night_country"))
                    / pl.col("total_time")
                    / pl.col("population_country")).alias("with_trf_light_norm"),
                ((pl.col("without_trf_light_day_country") + pl.col("without_trf_light_night_country"))
                    / pl.col("total_time")
                    / pl.col("population_country")).alias("without_trf_light_norm"),
                pl.col("country").str.to_titlecase().alias("country")]))
        bivariate.scatter(df=df,
                          x="with_trf_light_norm",
                          y="without_trf_light_norm",
                          color="continent",
                          text="iso3",
                          xaxis_title='Crossing events with traffic lights (normalised)',
                          yaxis_title='Crossing events without traffic lights (normalised)',
                          pretty_text=False,
                          marker_size=10,
                          save_file=True,
                          hover_data=hover_data,
                          hover_name="country",
                          legend_title="",
                          legend_x=0.87,
                          legend_y=1.0,
                          label_distance_factor=0.5,
                          marginal_x=None,  # type: ignore
                          marginal_y=None)  # type: ignore

        SPEED_COL = "speed_crossing_day_night_country_avg"
        TIME_COL = "time_crossing_day_night_country_avg"

        def _scalar_select(df: pl.DataFrame, expr: pl.Expr):
            """Return the single scalar from df.select(expr), or None if empty."""
            out = df.select(expr)
            if out.height == 0:
                return None
            s = out.to_series(0)
            return s[0] if len(s) else None

        def _row_value(df: pl.DataFrame, row_idx: int, col: str):
            """Get a value by row index; assumes row_idx is valid."""
            return df.row(row_idx, named=True)[col]

        if df_countries.height == 0:
            logger.error("df_countries is empty; cannot compute stats.")
        else:
            # Exclude zero values before finding min
            nonzero_speed = df_countries.filter(pl.col(SPEED_COL) > 0)
            nonzero_time = df_countries.filter(pl.col(TIME_COL) > 0)

            # Row indices (None if not computable)
            max_speed_idx = _scalar_select(df_countries, pl.col(SPEED_COL).arg_max())
            max_time_idx = _scalar_select(df_countries, pl.col(TIME_COL).arg_max())

            min_speed_idx = None
            if nonzero_speed.height > 0:
                min_speed_idx = (
                    nonzero_speed
                    .with_row_index("idx")
                    .sort(SPEED_COL)
                    .select("idx")
                    .to_series(0)[0]
                )

            min_time_idx = None
            if nonzero_time.height > 0:
                min_time_idx = (
                    nonzero_time
                    .with_row_index("idx")
                    .sort(TIME_COL)
                    .select("idx")
                    .to_series(0)[0]
                )

            # Mean and standard deviation (non-zero only)
            speed_mean = _scalar_select(nonzero_speed, pl.col(SPEED_COL).mean())
            speed_std = _scalar_select(nonzero_speed, pl.col(SPEED_COL).std())
            time_mean = _scalar_select(nonzero_time,  pl.col(TIME_COL).mean())
            time_std = _scalar_select(nonzero_time,  pl.col(TIME_COL).std())

            # Logging (guard against None indices)
            if max_speed_idx is not None:
                max_speed_country = _row_value(df_countries, int(max_speed_idx), "country")
                max_speed_value = float(_row_value(df_countries, int(max_speed_idx), SPEED_COL))
                logger.info(
                    f"Country with the highest average speed while crossing: {max_speed_country} "
                    f"({max_speed_value:.2f})"
                )
            else:
                logger.info("No max speed could be computed (empty or invalid speed column).")

            if min_speed_idx is not None:
                min_speed_country = _row_value(df_countries, int(min_speed_idx), "country")
                min_speed_value = float(_row_value(df_countries, int(min_speed_idx), SPEED_COL))
                logger.info(
                    f"Country with the lowest non-zero average speed while crossing: {min_speed_country} "
                    f"({min_speed_value:.2f})"
                )
            else:
                logger.info("No non-zero speed rows found; cannot compute min non-zero speed.")

            logger.info(
                f"Mean {crossing_metric_label().lower()} (non-zero): "
                f"{float(speed_mean) if speed_mean is not None else float('nan'):.2f}"
            )
            logger.info(
                f"Standard deviation of {crossing_metric_label().lower()} (non-zero): "
                f"{float(speed_std) if speed_std is not None else float('nan'):.2f}"
            )

            if max_time_idx is not None:
                max_time_country = _row_value(df_countries, int(max_time_idx), "country")
                max_time_value = float(_row_value(df_countries, int(max_time_idx), TIME_COL))
                logger.info(
                    f"Country with the highest average crossing time: {max_time_country} "
                    f"({max_time_value:.2f})"
                )
            else:
                logger.info("No max time could be computed (empty or invalid time column).")

            if min_time_idx is not None:
                min_time_country = _row_value(df_countries, int(min_time_idx), "country")
                min_time_value = float(_row_value(df_countries, int(min_time_idx), TIME_COL))
                logger.info(
                    f"Country with the lowest non-zero average crossing time: {min_time_country} "
                    f"({min_time_value:.2f})"
                )
            else:
                logger.info("No non-zero time rows found; cannot compute min non-zero time.")

            logger.info(f"Mean crossing time (non-zero): {float(time_mean) if time_mean is not None else float('nan'):.2f}")  # noqa: E501
            logger.info(f"Standard deviation of crossing time (non-zero): {float(time_std) if time_std is not None else float('nan'):.2f}")  # noqa: E501

            # Equivalent of: df_countries[['total_time','total_videos']].agg(['mean','std','sum'])
            stats = df_countries.select([
                pl.col("total_time").mean().alias("total_time_mean"),
                pl.col("total_time").std().alias("total_time_std"),
                pl.col("total_time").sum().alias("total_time_sum"),
                pl.col("total_videos").mean().alias("total_videos_mean"),
                pl.col("total_videos").std().alias("total_videos_std"),
                pl.col("total_videos").sum().alias("total_videos_sum"),
            ])
            if stats.height > 0:
                st = stats.row(0, named=True)
                logger.info(
                    f"Average total_time: {float(st['total_time_mean']):.2f}, "
                    f"Standard deviation: {float(st['total_time_std']):.2f}, "
                    f"Sum: {float(st['total_time_sum']):.2f}"
                )
                logger.info(
                    f"Average total_videos: {float(st['total_videos_mean']):.2f}, "
                    f"Standard deviation: {float(st['total_videos_std']):.2f}, "
                    f"Sum: {float(st['total_videos_sum']):.2f}"
                )

            # Max/min total_time rows (guarded)
            max_total_time_idx = _scalar_select(df_countries, pl.col("total_time").arg_max())
            if max_total_time_idx is not None:
                max_row = df_countries.row(int(max_total_time_idx), named=True)
                logger.info(
                    f"Country with maximum total_time: {max_row['country']}, "
                    f"total_time: {max_row['total_time']}, "
                    f"total_videos: {max_row['total_videos']}"
                )

            min_total_time_idx = _scalar_select(df_countries, pl.col("total_time").arg_min())
            if min_total_time_idx is not None:
                min_row = df_countries.row(int(min_total_time_idx), named=True)
                logger.info(
                    f"Country with minimum total_time: {min_row['country']}, "
                    f"total_time: {min_row['total_time']}, "
                    f"total_videos: {min_row['total_videos']}"
                )
        logger.info("Analysis complete.")