from _collections_abc import dict_items
import os
import re
from typing import Any, cast, Tuple, TYPE_CHECKING
if TYPE_CHECKING:
    from omrat import OMRAT

import numpy as np
from qgis.core import QgsProject
from qgis.PyQt.QtCore import QSettings
from qgis.PyQt.QtWidgets import QMessageBox, QTableWidget
from shapely import wkt
from shapely.geometry import Point
from shapely.geometry.base import BaseGeometry

from compute.database import DB
from omrat_utils.vessel_lookup import VesselLookupConfig
from ui.ais_connection_widget import AISConnectionWidget


_SQL_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


def _validate_sql_identifier(name: str) -> str:
    """Reject anything that isn't a plain SQL identifier.

    Used so that the AIS schema name (which the user types in the GUI) can be
    safely interpolated into table references — table/schema names cannot be
    bound as query parameters, so we whitelist instead.
    """
    if not isinstance(name, str) or not _SQL_IDENT_RE.match(name):
        raise ValueError(f"Invalid SQL identifier: {name!r}")
    return name

def update_ais_settings_file(db_host:str, db_user:str, db_pass:str, db_name:str):
    ais_settings_path = os.path.join(os.path.dirname(__file__), '..', 'ui', 'ais_settings.py')
    with open(ais_settings_path, 'w', encoding='utf-8') as f:
        f.write(f"""db_host = '{db_host}'
db_user = '{db_user}'
db_password = '{db_pass}'
db_name = '{db_name}'
""")


def get_pl(db:DB, lat1:float, lat2:float, lon1:float, lon2:float, l_width:float) -> str:
    """Collects the passage line as text"""
    sql = f"""SELECT st_astext(st_makeline(ST_Project(ST_Centroid(ST_GeomFromText('LINESTRING({lon1} {lat1}, {lon2} {lat2})', 4326))::geography, 
    {l_width/2}, ST_Azimuth(ST_Point({lon1}, {lat1})::geography, ST_Point({lon2}, {lat2})::geography) + radians(90))::geometry,
    ST_Project(ST_Centroid(ST_GeomFromText('LINESTRING({lon1} {lat1}, {lon2} {lat2})', 4326))::geography, 
    {l_width/2}, ST_Azimuth(ST_Point({lon1}, {lat1})::geography, ST_Point({lon2}, {lat2})::geography) + radians(270))::geometry
    ))"""
    ok, res = cast(tuple[bool, list[list[Any]]], db.execute_and_return(sql, return_error=True))
    if ok:
        pl: str = res[0][0]
    else:
        pl = ""
    return pl

def get_type(toc: float) -> int:
    """Return ship type index matching the UI ship category list.

    Maps AIS Type-of-Cargo (TOC) codes to indices 0-20 corresponding to:
        0: Fishing (TOC 30)
        1: Towing (TOC 31-32)
        2: Dredging or underwater ops (TOC 33)
        3: Diving ops (TOC 34)
        4: Military ops (TOC 35)
        5: Sailing (TOC 36)
        6: Pleasure Craft (TOC 37)
        7: High speed craft (TOC 40-49)
        8: Pilot Vessel (TOC 50)
        9: Search and Rescue vessel (TOC 51)
        10: Tug (TOC 52)
        11: Port Tender (TOC 53)
        12: Anti-pollution equipment (TOC 54)
        13: Law Enforcement (TOC 55)
        14: Spare (TOC 56-57)
        15: Medical Transport (TOC 58)
        16: Noncombatant ship (TOC 59)
        17: Passenger (TOC 60-69)
        18: Cargo (TOC 70-79)
        19: Tanker (TOC 80-89)
        20: Other Type (everything else)
    """
    # NULL / unparseable type_and_cargo is common — any MMSI whose Type-5
    # statics never came through arrives here as None.  Bucket those into
    # "Other Type" rather than crashing the whole leg's traffic build.
    if toc is None:
        return 20
    try:
        toc_int = int(toc)
    except (TypeError, ValueError):
        return 20
    _TOC_MAP = {
        30: 0, 31: 1, 32: 1, 33: 2, 34: 3, 35: 4, 36: 5, 37: 6,
        50: 8, 51: 9, 52: 10, 53: 11, 54: 12, 55: 13,
        56: 14, 57: 14, 58: 15, 59: 16,
    }
    if toc_int in _TOC_MAP:
        return _TOC_MAP[toc_int]
    if 40 <= toc_int <= 49:
        return 7
    if 60 <= toc_int <= 69:
        return 17
    if 70 <= toc_int <= 79:
        return 18
    if 80 <= toc_int <= 89:
        return 19
    return 20

def close_to_line(bearing:float, cog:float, max_angle:float) -> bool:
    """Returns True if the cog is less than the max_angle towards the bearing else False"""
    if max_angle > 180:
        raise ValueError("Maximum deviation must be lower than 180 degrees")
        return
    if bearing > 360:
        bearing = np.mod(bearing, 360)
    if cog > 360:
        cog = np.mod(cog, 360)
    if bearing - max_angle > 0 and bearing + max_angle < 360:
        if bearing - max_angle < cog < bearing + max_angle:
            return True
        else:
            return False
    else:
        if np.mod(bearing - max_angle, 360) < cog or cog < np.mod(
                bearing + max_angle, 360):
            return True
        else:
            return False

class AIS:
    def __init__(self, omrat:"OMRAT"):
        self.settings: QSettings = QSettings()
        self.db: DB | None = None
        self.omrat = omrat
        self.acw = AISConnectionWidget()
        # Optional LEFT JOIN against an external vessel-metadata table.
        # The default-constructed config has ``enabled=False`` so
        # ``is_valid()`` is False and ``run_sql`` skips the JOIN.
        self.vessel_lookup: VesselLookupConfig = VesselLookupConfig.from_qsettings()
        # When True, ``update_legs`` divides the year-of-seconds by the
        # actual coverage seen in the data and uses that as a multiplier
        # on the per-ping frequency increment, so a partial-year ingest
        # is reported as an annualised rate.
        self.recalc_to_full_year: bool = self._read_recalc_setting()
        self.set_start_ais_settings()
        self.dist_data: dict[str, dict[str, np.ndarray]] = {}

    def _read_recalc_setting(self) -> bool:
        """Read ``omrat/recalc_to_full_year`` and coerce to bool.

        QSettings on Windows stores booleans as the strings ``"true"`` /
        ``"false"`` rather than Python bools, so we accept either.
        """
        raw = self.settings.value("omrat/recalc_to_full_year", False)
        if isinstance(raw, str):
            return raw.strip().lower() == "true"
        return bool(raw)
        
    def unload(self):
        """If signals from omrat was used they are disconnected"""
        pass
        
    def run(self):
        self.acw.show()
        self.acw.accepted.connect(self.update_ais_settings)
        self.acw.exec()
    
    def set_start_ais_settings(self):
        db_host: str = self.settings.value("omrat/db_host", "")
        db_name: str = self.settings.value("omrat/db_name", "")
        db_user: str = self.settings.value("omrat/db_user", "")
        db_pass: str = self.settings.value("omrat/db_pass", "")
        # ``omrat/db_port`` is new (older installs won't have it).  Fall back
        # to 5432, the Postgres default.
        try:
            db_port = int(self.settings.value("omrat/db_port", 5432) or 5432)
        except (TypeError, ValueError):
            db_port = 5432
        self.acw.leDBHost.setText(db_host)
        self.acw.SBPort.setValue(db_port)
        self.acw.leDBName.setText(db_name)
        self.acw.leUserName.setText(db_user)
        self.acw.lePassword.setText(db_pass)
        # External vessel-data lookup: prefill from QSettings.
        self.acw.gbExtVessel.setChecked(bool(self.vessel_lookup.enabled))
        self.acw.leExtSchema.setText(self.vessel_lookup.schema)
        self.acw.leExtTable.setText(self.vessel_lookup.table)
        self.acw.leExtMmsiCol.setText(self.vessel_lookup.mmsi_col or "mmsi")
        self.acw.leExtLoaCol.setText(self.vessel_lookup.loa_col)
        self.acw.leExtBeamCol.setText(self.vessel_lookup.beam_col)
        self.acw.leExtShipTypeCol.setText(self.vessel_lookup.ship_type_col)
        self.acw.leExtAirDraughtCol.setText(self.vessel_lookup.air_draught_col)
        self.acw.cbRecalcFullYear.setChecked(bool(self.recalc_to_full_year))
        self.schema = self.acw.leProvider.text()
        self.year = self.acw.SBYear.value()
        self.months: list[int] = []
        for i in range(1, 13):
            cb = getattr(self.acw, f'CB_{i}')
            if cb.isChecked():
                self.months.append(i)
        try:
            self.db = DB(db_host=db_host,
                        db_name=db_name,
                        db_user=db_user,
                        db_pass=db_pass,
                        db_port=db_port)
        except Exception:
            self.db = None
        self.max_deviation = float(self.acw.leMaxDev.text())
    
    def update_ais_settings(self):
        db_host = self.acw.leDBHost.text()
        db_port = int(self.acw.SBPort.value())
        db_name = self.acw.leDBName.text()
        db_user = self.acw.leUserName.text()
        db_pass = self.acw.lePassword.text()
        self.schema = self.acw.leProvider.text()
        self.year = self.acw.SBYear.value()
        self.months = []
        for i in range(1, 13):
            cb = getattr(self.acw, f'CB_{i}')
            if cb.isChecked():
                self.months.append(i)
        # Catch connection failures (wrong password, missing role, server
        # down, ...) and surface them as a popup instead of letting the
        # exception bubble up into a Python-error dialog.  ``DB._connect``
        # already produces a friendly, cp1252-decoded message so we just
        # show the exception text.
        try:
            self.db = DB(
                db_host=db_host, db_name=db_name,
                db_user=db_user, db_pass=db_pass,
                db_port=db_port,
            )
        except Exception as e:
            self.db = None
            QMessageBox.warning(
                self.omrat.main_widget,
                self.omrat.tr("Could not connect to AIS database"),
                str(e),
            )
            return
        self.max_deviation = float(self.acw.leMaxDev.text())
        for key, value in zip(
            ["db_host", "db_port", "db_user", "db_pass", "db_name"],
            [db_host, db_port, db_user, db_pass, db_name],
        ):
            self.settings.setValue(f"omrat/{key}", value)
        # Capture + persist the external-vessel-lookup config so the
        # next ``run_sql`` call sees the user's edits and the next
        # session restores them.
        self.vessel_lookup = VesselLookupConfig(
            enabled=bool(self.acw.gbExtVessel.isChecked()),
            schema=self.acw.leExtSchema.text().strip(),
            table=self.acw.leExtTable.text().strip(),
            mmsi_col=self.acw.leExtMmsiCol.text().strip() or "mmsi",
            loa_col=self.acw.leExtLoaCol.text().strip(),
            beam_col=self.acw.leExtBeamCol.text().strip(),
            ship_type_col=self.acw.leExtShipTypeCol.text().strip(),
            air_draught_col=self.acw.leExtAirDraughtCol.text().strip(),
        )
        self.vessel_lookup.to_qsettings()
        # Recalculate-to-full-year flag.
        self.recalc_to_full_year = bool(self.acw.cbRecalcFullYear.isChecked())
        self.settings.setValue(
            "omrat/recalc_to_full_year", self.recalc_to_full_year,
        )

    # ----------------------------------------------------------- coverage

    # Year-of-seconds (non-leap) — the reference window the multiplier
    # normalises to.  The 6 h discrepancy from a leap year is well below
    # the resolution of "ships per year" estimates and not worth a
    # branch.
    _YEAR_SECONDS = 365 * 24 * 3600

    # Gaps in the data longer than this are treated as receiver outages
    # rather than valid empty periods, and excluded from the coverage
    # window.  12 h matches the user's spec.
    _GAP_THRESHOLD_HOURS = 12

    def compute_year_multiplier(self) -> tuple[float, float, float] | None:
        """Inspect the segments table and return ``(multiplier, coverage_s, gap_s)``.

        ``coverage_s`` = ``max(date1) - min(date1)`` minus the time spent
        inside any inter-ping gap longer than
        ``_GAP_THRESHOLD_HOURS``.  ``multiplier`` =
        ``_YEAR_SECONDS / coverage_s`` so callers can scale a per-ping
        count to an annualised rate.

        Returns ``None`` when the DB is unset, the query fails, or the
        table is empty (no rows → no coverage to extrapolate from).
        """
        if self.db is None:
            return None
        from psycopg2 import sql as psql

        schema = _validate_sql_identifier(str(self.schema))
        year = int(self.year)
        gap_h = int(self._GAP_THRESHOLD_HOURS)
        sql = psql.SQL(
            "WITH ordered AS ("
            "SELECT date1, date1 - LAG(date1) OVER (ORDER BY date1) AS dt "
            "FROM {schema}.{segments}"
            ") "
            "SELECT EXTRACT(EPOCH FROM (MAX(date1) - MIN(date1))), "
            "COALESCE(SUM(EXTRACT(EPOCH FROM dt)) "
            "FILTER (WHERE dt > make_interval(hours => %s)), 0) "
            "FROM ordered"
        ).format(
            schema=psql.Identifier(schema),
            segments=psql.Identifier(f"segments_{year}"),
        )
        ok, rows = cast(
            tuple[bool, list[list[Any]]],
            self.db.execute_and_return(sql, return_error=True, params=(gap_h,)),
        )
        if not ok or not rows or rows[0][0] is None:
            return None
        try:
            span_s = float(rows[0][0])
            gap_s = float(rows[0][1])
        except (TypeError, ValueError):
            return None
        coverage_s = span_s - gap_s
        if coverage_s <= 0:
            return None
        multiplier = self._YEAR_SECONDS / coverage_s
        return multiplier, coverage_s, gap_s

    def update_legs(self, key:str|None=None):
        """Update AIS data for all legs or a specific leg."""
        if self.db is None:
            QMessageBox.information(
                self.omrat.main_widget,
                self.omrat.tr("AIS database not connected"),
                self.omrat.tr(
                    "No AIS database connection is configured.\n\n"
                    "Open Settings -> AIS Connection and provide a host, "
                    "database, user and password before fetching AIS traffic."
                ),
            )
            return
        segment_data: dict[str, dict[str, str]] = self.get_segment_data_from_table()
        if key is None:
            legs: dict_items[str, dict[str, str]] = segment_data.items()
        else:
            legs = [[key, segment_data[key]]] # type: ignore
        # Compute the annualisation multiplier once per refresh — it's
        # a property of the database's overall coverage, not per-leg.
        # When the user hasn't enabled the option, OR the query fails /
        # the table is empty, we silently fall back to 1.0 (no scaling).
        multiplier = 1.0
        if self.recalc_to_full_year:
            info = self.compute_year_multiplier()
            if info is not None:
                multiplier, coverage_s, gap_s = info
                QMessageBox.information(
                    self.omrat.main_widget,
                    self.omrat.tr("Annualised AIS frequency"),
                    self.omrat.tr(
                        f"Coverage: {coverage_s / 3600:.1f} h "
                        f"(skipped {gap_s / 3600:.1f} h of gaps > 12 h).\n"
                        f"Multiplier: {multiplier:.2f}x "
                        f"({self._YEAR_SECONDS / 3600:.0f} h / "
                        f"{coverage_s / 3600:.1f} h)."
                    ),
                )
        for leg_key, leg_d in legs:
            if leg_key not in self.omrat.qgis_geoms.leg_dirs.keys():
                # Use the current leg_key to look up segment direction labels
                self.omrat.qgis_geoms.leg_dirs[leg_key] = self.omrat.segment_data[leg_key]["Dirs"]
            dirs = self.omrat.qgis_geoms.leg_dirs[leg_key]
            self.omrat.traffic.create_empty_dict(leg_key, dirs)
            start_p = wkt.loads(f"Point ({leg_d['Start_Point']})")
            assert isinstance(start_p, Point)
            end_p = wkt.loads(f"Point ({leg_d['End_Point']})")
            assert isinstance(end_p, Point)
            pl = get_pl(self.db, lat1=start_p.y,
                        lat2=end_p.y,
                        lon1=start_p.x,
                        lon2=end_p.x, 
                        l_width=float(leg_d['Width']))
            try:
                ais_data = self.run_sql(pl)
            except Exception as e:
                self.omrat.show_error_popup(str(e), "AIS.update_legs")
                return
            sql = f"""select degrees(ST_Azimuth(ST_Point({start_p.x}, {start_p.y})::geography, 
            ST_Point({end_p.x}, {end_p.y})::geography))
            """
            _, leg_bearing = cast(tuple[bool, list[list[Any]]], self.db.execute_and_return(sql, return_error=True))
            line1, line2 = self.update_ais_data(
                leg_key, ais_data, leg_bearing[0][0], dirs,
                multiplier=multiplier,
            )
            self.convert_list2avg()
            self.update_dist_data(line1, line2, leg_key)
            key = leg_key
        if key is not None:
            self.omrat.main_widget.leNormMean1_1.setText('')
            self.omrat.distributions.run_update_plot(key)
            self.omrat.main_widget.cbTrafficSelectSeg.setCurrentIndex(self.omrat.main_widget.cbTrafficSelectSeg.count()-1)
        
    def convert_list2avg(self):
        for key1 in self.omrat.traffic.traffic_data.keys():
            for key2 in self.omrat.traffic.traffic_data[key1].keys():
                for key3 in self.omrat.traffic.traffic_data[key1][key2].keys():
                    for idx1, _ in enumerate(self.omrat.traffic.traffic_data[key1][key2][key3]):
                        for idx2, val in enumerate(self.omrat.traffic.traffic_data[key1][key2][key3][idx1]):
                            if isinstance(val, list):
                                if len(val) > 0:
                                    self.omrat.traffic.traffic_data[key1][key2][key3][idx1][idx2] = np.array(val).mean()
                                else:
                                    self.omrat.traffic.traffic_data[key1][key2][key3][idx1][idx2] = np.inf

    def update_dist_data(self, line1:np.ndarray, line2:np.ndarray, key:str) -> None:
        # Keep a dedicated distribution cache used by Distributions.run_update_plot.
        self.dist_data[key] = {
            'line1': line1,
            'line2': line2
        }
        self.omrat.segment_data[key]['mean1_1'] = line1.mean()
        self.omrat.segment_data[key]['std1_1'] = line1.std()
        self.omrat.segment_data[key]['mean2_1'] = line2.mean()
        self.omrat.segment_data[key]['std2_1'] = line2.std()
        self.omrat.segment_data[key]['weight1_1'] = 100
        self.omrat.segment_data[key]['weight2_1'] = 100
        self.omrat.segment_data[key]['dist1'] = line1
        self.omrat.segment_data[key]['dist2'] = line2
        if float(self.omrat.main_widget.leNormMean1_1.text()) == 0.0:
            self.omrat.main_widget.leNormMean1_1.setText(str(line1.mean()))
            self.omrat.main_widget.leNormMean2_1.setText(str(line2.mean()))
            self.omrat.main_widget.leNormStd1_1.setText(str(line1.std()))
            self.omrat.main_widget.leNormStd2_1.setText(str(line2.std()))
        for j in range(1, 3):
            for i in range(2, 4):
                self.omrat.segment_data[key][f'mean{j}_{i}'] = 0
                self.omrat.segment_data[key][f'std{j}_{i}'] = 0
                self.omrat.segment_data[key][f'weight{j}_{i}'] = 0
            self.omrat.segment_data[key][f'u_min{j}'] = 0
            self.omrat.segment_data[key][f'u_max{j}'] = 0
            self.omrat.segment_data[key][f'u_p{j}'] = 0
        self.omrat.segment_data[key]['ai1'] = 180
        self.omrat.segment_data[key]['ai2'] = 180
    
    def run_sql(self, pl: str) -> list[list[Any]]:
        """Runs the SQL query to get the passages.

        The schema name (a SQL identifier — cannot be bound as a parameter)
        is checked against a strict identifier regex; year/month are coerced
        to int; the polygon WKT is bound via the ``%(pl)s`` placeholder so it
        is never inlined into the query string.
        """
        if self.db is None:
            # Should not occur
            raise RuntimeError(self.omrat.tr("No database connection was found"))

        schema = _validate_sql_identifier(str(self.schema))
        year = int(self.year)
        months: list[int] = []
        for m in self.months:
            mi = int(m)
            if not 1 <= mi <= 12:
                raise ValueError(f"Invalid AIS month value: {m!r}")
            months.append(mi)

        # Build the per-table fragments that will be UNION'd together.
        # ``schema`` is a validated SQL identifier and ``year`` / ``month``
        # are ints, so f-string interpolation is safe.  ``pl`` is bound
        # via the ``%(pl)s`` parameter placeholder, never inlined.
        select_template = (
            "select ss.mmsi, segment, cog, sog, draught, type_and_cargo, "
            "date1, dim_a, dim_b, dim_c, dim_d "
            "FROM {tbl} ss "
            "JOIN {schema}.states_{year} st on st.rowid=ss.state_id "
            "JOIN {schema}.statics_{year} si on si.rowid=st.static_id "
            "WHERE ST_intersects(segment, ST_geomfromtext(%(pl)s, 4326))"
        )
        if months:
            tables = [f"{schema}.segments_{year}_{month}" for month in months]
        else:
            tables = [f"{schema}.segments_{year}"]
        union_block = " UNION ".join(
            select_template.format(tbl=tbl, schema=schema, year=year)
            for tbl in tables
        )

        # nosec B608 - schema/year/month identifiers are validated above
        # (regex + int coerce); ``pl`` is bound through the ``%(pl)s``
        # placeholder, never inlined into the query string.
        #
        # Vessel dimensions come straight from the AIS Type-5 statics
        # (dim_a/b/c/d):
        #   loa  = dim_a + dim_b   (length: bow offset + stern offset)
        #   beam = dim_c + dim_d   (beam:   port offset + starboard offset)
        # When the AIS dimensions look bogus (sum < 2 m, or any single
        # offset > the AIS spec maximum) we fall back to the user's
        # external vessel table if one is configured (see
        # ``vessel_lookup`` / the AIS Settings dialog), otherwise NULL.
        # ``update_ais_data`` then falls back to its own default
        # (loa=100 m, skip beam).
        #
        # ``ship_type`` and ``air_draught`` come from the external table
        # only — neither is in the AIS Type-5 statics, so without a
        # configured vessel-lookup table they're NULL.  Ship
        # classification then falls back to ``type_and_cargo`` via
        # ``get_type``; air-draught distributions stay empty.
        ext_active = self.vessel_lookup.is_valid()
        if ext_active:
            # Defence-in-depth: ``is_valid()`` already enforces the
            # SQL-identifier shape, but re-validate before interpolation
            # so any future plumbing change still trips an early raise
            # rather than handing a poisoned identifier to libpq.
            for ident in (
                self.vessel_lookup.schema, self.vessel_lookup.table,
                self.vessel_lookup.mmsi_col, self.vessel_lookup.loa_col,
                self.vessel_lookup.beam_col, self.vessel_lookup.ship_type_col,
                self.vessel_lookup.air_draught_col,
            ):
                if ident:
                    _validate_sql_identifier(ident)
            cte_block = ", " + self.vessel_lookup.build_cte()
            ext_join = (
                " LEFT OUTER JOIN external_vessels ext ON ss.mmsi = ext.mmsi"
            )
            loa_fallback = "ext.ext_loa"
            beam_fallback = "ext.ext_beam"
            ship_type_expr = "ext.ext_ship_type as ship_type"
            air_draught_expr = "ext.ext_air_draught as air_draught"
        else:
            cte_block = ""
            ext_join = ""
            loa_fallback = "NULL"
            beam_fallback = "NULL"
            ship_type_expr = "NULL::int as ship_type"
            air_draught_expr = "NULL::double precision as air_draught"

        sql = (
            "with segments as (" + union_block + ")" + cte_block + " "  # nosec B608
            "SELECT case when dim_a + dim_b < 2 or dim_a > 510 or dim_b > 510 "
            f"then {loa_fallback} else dim_a + dim_b end as loa, "
            "case when dim_c + dim_d < 2 or dim_c > 62 or dim_d > 62 "
            f"then {beam_fallback} else dim_c + dim_d end as beam, "
            "type_and_cargo, draught, "
            f"{ship_type_expr}, "
            "date1, sog, "
            f"{air_draught_expr}, "
            "st_distance("
            "st_intersection(segment, st_geomfromtext(%(pl)s,4326))::geography, "
            "st_startpoint(st_geomfromtext(%(pl)s, 4326))::geography"
            ") - st_length(st_geomfromtext(%(pl)s, 4326)::geography) / 2 "
            "as dist_from_start, cog "
            "FROM segments ss"
            + ext_join
        )
        ok, ais_data = cast(
            tuple[bool, list[list[Any]]],
            self.db.execute_and_return(sql, return_error=True, params={"pl": pl}),
        )
        if ok:
            return ais_data
        else:
            raise TypeError(ais_data[0][0])

    def update_ais_data(
        self, leg_key: str, ais_data: list[list[Any]],
        leg_bearing: float, dirs: list[str],
        multiplier: float = 1.0,
    ) -> tuple[np.ndarray, np.ndarray]:
        """Bin every passing ship into the per-(type, loa) frequency matrix.

        ``multiplier`` (default 1.0) scales the per-ping increment so a
        partial-year ingest can be reported as an annualised rate.  See
        :meth:`compute_year_multiplier`.  Speed/Beam/Draught/Heights
        distributions are *not* scaled — they are observations, not
        counts, and ``convert_list2avg`` averages them downstream.
        """
        line1:list[float] = []
        line2:list[float] = []
        for loa, beam, toc, draugt, sh_type, _, sog, air_draught, dist, cog in ais_data:
            if close_to_line(leg_bearing + 180, cog, self.max_deviation):
                line1.append(dist)
                l1 = True
            elif close_to_line(leg_bearing, cog, self.max_deviation):
                line2.append(dist)
                l1 = False
            else:
                # print(leg_bearing, cog, self.max_deviation)
                continue
            dir_ = dirs[0] if l1 else dirs[1]
            if loa is None:
                loa = 100
            loa = int(loa)
            loa_cat = -1
            freq_data = self.omrat.traffic.traffic_data[leg_key][dir_]['Frequency (ships/year)']
            n_loa_cats = len(freq_data[0]) if freq_data else 0
            for loa_i in range(n_loa_cats):
                if loa_i * 25 < loa <= (loa_i * 25 + 25):
                    loa_cat = loa_i
                    continue
            if loa_cat < 0:
                loa_cat = n_loa_cats - 1 if n_loa_cats > 0 else 0
            if sh_type is None:
                sh_type = ''
            type_cat = get_type(toc)
            self.omrat.traffic.traffic_data[leg_key][dir_]['Frequency (ships/year)'][type_cat][loa_cat] += multiplier
            if sog is not None:
                self.omrat.traffic.traffic_data[leg_key][dir_]['Speed (knots)'][type_cat][loa_cat].append(float(sog))
            if air_draught is not None:
                self.omrat.traffic.traffic_data[leg_key][dir_]['Ship heights (meters)'][type_cat][loa_cat].append(
                    float(air_draught))
            if beam is not None:
                self.omrat.traffic.traffic_data[leg_key][dir_]['Ship Beam (meters)'][type_cat][loa_cat].append(float(beam))
            if draugt is not None:
                self.omrat.traffic.traffic_data[leg_key][dir_]['Draught (meters)'][type_cat][loa_cat].append(
                    float(draugt))
        np_line1 = np.array(line1)
        np_line2 = np.array(line2)
        return np_line1, np_line2

    # ----------------------------------------------------------- junction transitions

    def fetch_passages_for_leg(
        self, leg_d: dict[str, str], near_radius_m: float | None = None,
    ) -> dict[str, list[float]]:
        """Return ``{mmsi: [unix_timestamp, ...]}`` for AIS pings inside the leg.

        ``near_radius_m`` is currently unused — the existing ``run_sql``
        builds a passage line spanning the whole width of the leg, which
        is the same data the per-leg traffic update consumes.  When AIS-
        derived junction transitions become production-grade we will
        switch to a near-junction sub-polygon to better reflect "ships
        actually transiting the junction" instead of "ships anywhere
        on the leg".  For now the wider polygon is good enough because
        the ``transition_counts_from_passages`` algorithm self-filters
        by the per-MMSI temporal window.
        """
        if self.db is None:
            return {}
        start_p = wkt.loads(f"Point ({leg_d['Start_Point']})")
        end_p = wkt.loads(f"Point ({leg_d['End_Point']})")
        if not isinstance(start_p, Point) or not isinstance(end_p, Point):
            return {}
        pl = get_pl(
            self.db,
            lat1=start_p.y, lat2=end_p.y,
            lon1=start_p.x, lon2=end_p.x,
            l_width=float(leg_d.get('Width', 5000)),
        )
        try:
            rows = self.run_sql(pl)
        except Exception:
            return {}
        # ``run_sql`` returns the per-ping row used by ``update_ais_data``;
        # we only need the (mmsi, date1) pair to build the transition
        # counts.  ``date1`` is at column index 5 in the SELECT order
        # (loa, beam, type_and_cargo, draught, ship_type, date1, sog,
        # air_draught, dist_from_start, cog).  ``mmsi`` is not in the
        # public SELECT list — we replay the same query joined back to
        # ``ss.mmsi`` here for the targeted use case.
        # Keep the implementation pragmatic: re-issue a lighter query
        # focused on (mmsi, date1) so we don't have to refactor run_sql.
        return self._fetch_mmsi_passages(pl)

    def _fetch_mmsi_passages(self, pl: str) -> dict[str, list[float]]:
        """Return ``{mmsi: [unix_timestamp, ...]}`` for the given passage line."""
        if self.db is None:
            return {}
        schema = _validate_sql_identifier(str(self.schema))
        year = int(self.year)
        months: list[int] = []
        for m in (self.months or []):
            mi = int(m)
            if 1 <= mi <= 12:
                months.append(mi)
        if months:
            tables = [f"{schema}.segments_{year}_{month}" for month in months]
        else:
            tables = [f"{schema}.segments_{year}"]
        # ``schema`` is regex-validated, ``year``/``month`` are coerced to
        # int, and ``pl`` is bound through the ``%(pl)s`` placeholder, so
        # f-string interpolation of the table name is safe here.
        select_template = (  # nosec B608
            "select ss.mmsi, "
            "extract(epoch from date1) as t "
            "FROM {tbl} ss "
            "JOIN {schema}.states_{year} st on st.rowid=ss.state_id "
            "WHERE ST_intersects(segment, ST_geomfromtext(%(pl)s, 4326))"
        )
        sql = " UNION ".join(  # nosec B608
            select_template.format(tbl=tbl, schema=schema, year=year)
            for tbl in tables
        )
        try:
            ok, rows = cast(
                tuple[bool, list[list[Any]]],
                self.db.execute_and_return(
                    sql, return_error=True, params={"pl": pl},
                ),
            )
        except Exception:
            return {}
        if not ok or not rows:
            return {}
        out: dict[str, list[float]] = {}
        for mmsi, t in rows:
            if mmsi is None or t is None:
                continue
            out.setdefault(str(mmsi), []).append(float(t))
        return out

    def compute_junction_transitions(
        self,
        time_window_s: float | None = None,
    ) -> dict[str, dict[str, dict[str, int]]]:
        """Build a ``{junction_id: {in_leg: {out_leg: count}}}`` table from AIS.

        Returns an empty dict if the database is offline or the project
        has no junctions.  Junctions are read from the live
        :class:`Junctions` handler on the parent OMRAT instance.
        """
        from compute.junction_transitions import (
            DEFAULT_TIME_WINDOW_S,
            transition_counts_from_passages,
        )
        handler = getattr(self.omrat, 'junctions', None)
        if handler is None or not handler.registry:
            return {}
        if self.db is None:
            return {}
        seg_table = self.get_segment_data_from_table()
        # First fetch passages once per leg so junctions sharing legs
        # don't query the database twice.
        leg_ids: set[str] = set()
        for j in handler.registry.values():
            for leg_id in j.legs:
                leg_ids.add(leg_id)
        passages_by_leg: dict[str, dict[str, list[float]]] = {}
        for leg_id in leg_ids:
            leg_d = seg_table.get(str(leg_id))
            if leg_d is None:
                passages_by_leg[leg_id] = {}
                continue
            passages_by_leg[leg_id] = self.fetch_passages_for_leg(leg_d)
        # Build per-junction count tables by restricting the global
        # passage map to the legs that touch each junction.
        out: dict[str, dict[str, dict[str, int]]] = {}
        window = (
            DEFAULT_TIME_WINDOW_S if time_window_s is None
            else float(time_window_s)
        )
        for jid, j in handler.registry.items():
            local = {
                leg_id: passages_by_leg.get(leg_id, {})
                for leg_id in j.legs
            }
            out[jid] = transition_counts_from_passages(
                local, time_window_s=window,
            )
        return out

    def get_segment_data_from_table(self) -> dict[str, dict[str, str]]:
        """Extract segment data from the QTableWidget (twRouteList)."""
        segment_data: dict[str, dict[str, str]] = {}
        table:QTableWidget = self.omrat.main_widget.twRouteList
        row_count: int = table.rowCount()

        for row in range(row_count):
            segment_id: str | None = table.item(row, 0).text() if table.item(row, 0) else None
            route_id: str | None = table.item(row, 1).text() if table.item(row, 1) else None
            leg_name: str | None = table.item(row, 2).text() if table.item(row, 2) else None
            start_point: str | None = table.item(row, 3).text() if table.item(row, 3) else None
            end_point: str | None = table.item(row, 4).text() if table.item(row, 4) else None
            width: str|None = table.item(row, 5).text() if table.item(row, 5) else None

            if segment_id and route_id and start_point and end_point and width:
                segment_data[segment_id] = {
                    'Route_Id': route_id,
                    'Leg_name': leg_name,
                    'Start_Point': start_point,
                    'End_Point': end_point,
                    'Width': width
                }

        return segment_data
