"""
Coordinate management for seeding, retrieval, and central-point validation.
This module provides a focused set of utilities for initializing and validating
the coordinate grid used by the weather pipeline. It inserts a single central
coordinate and a deterministic ring of surrounding coordinates, and exposes
read helpers that return Pydantic schemas. The goal is to keep persistence and
validation responsibilities decoupled from input validation and orchestration
logic in higher-level modules.
See Also
--------
app.coordinates_setup : Input validation and high-level orchestration.
app.coordinates_utils.calculate_destination_point : Great-circle
destination computation used to derive surrounding points.
app.models.Coordinate : SQLAlchemy ORM entity persisted by this module.
app.schemas.CoordinateSchema : Pydantic schema returned by query helpers.
Notes
-----
- Primary role: persist a central coordinate and a deterministic set of
surrounding points and validate that the stored central coordinate matches
the configured central values.
- Key dependencies: a live SQLAlchemy `Session`, the ORM model
`app.models.Coordinate`, and the Pydantic schema `app.schemas.CoordinateSchema`.
Geodesic calculations are delegated to
`app.coordinates_utils.calculate_destination_point`.
- Invariants: callers own the `Session` lifecycle. Exactly one row is marked
as the central coordinate (``is_central=True``). Generated points are
deterministic for the given bearings and distances.
Examples
--------
>>> # Typical usage within a request/worker context (requires a real DB)
>>> # from app.database import SessionLocal # doctest: +SKIP
>>> # session = SessionLocal() # doctest: +SKIP
>>> # seed_coordinates_if_needed(session, 59.3293, 18.0686) # doctest: +SKIP
>>> # coords = get_coordinates(session) # doctest: +SKIP
>>> # len(coords) > 0 # doctest: +SKIP
True
"""
import logging
from typing import List, Optional
from sqlalchemy.orm import Session
from .coordinates_utils import calculate_destination_point as destination_point
from .models import Coordinate
from .schemas import CoordinateSchema
logger = logging.getLogger(__name__)
# Constants for coordinate generation
BEARINGS: list[int] = [0, 45, 90, 135, 180, 225, 270, 315]
NEAR_DISTANCES: list[float] = [1.0, 5.0] # km
FAR_DISTANCES: list[float] = [50.0, 150.0] # km
COORDINATE_MATCH_TOLERANCE: float = 1e-6 # Threshold for coordinate match in degrees
[docs]
def seed_coordinates_if_needed(
session: Session,
central_latitude: float,
central_longitude: float,
) -> None:
"""Seed the table or validate the persisted central coordinate.
If the table is empty, inserts the central point and a deterministic set of
surrounding points. Otherwise validates that the stored central point is
equal (within tolerance) to the configured central coordinate.
Parameters
----------
session : sqlalchemy.orm.Session
Database session controlling the transaction.
central_latitude : float
Central latitude, typically provided by configuration.
central_longitude : float
Central longitude, typically provided by configuration.
Raises
------
RuntimeError
If the central coordinate is missing or mismatches the provided values.
See Also
--------
app.coordinates_manager.get_coordinates
app.coordinates_manager.set_coordinate
app.coordinates_utils.calculate_destination_point
Examples
--------
>>> # Using a real DB session in application code # doctest: +SKIP
>>> # seed_coordinates_if_needed(session, 59.3293, 18.0686) # doctest: +SKIP
>>> # (no return value) # doctest: +SKIP
"""
existing_coordinates = session.query(Coordinate).all()
if not existing_coordinates:
logger.info("Seeding coordinates table with central and surrounding points.")
_seed_all_coordinates(session, central_latitude, central_longitude)
logger.info("Coordinates table seeded.")
else:
_validate_central_coordinate(session, central_latitude, central_longitude)
def _seed_all_coordinates(
session: Session,
central_latitude: float,
central_longitude: float,
) -> None:
"""Seed central and surrounding coordinates into the database.
Parameters
----------
session : sqlalchemy.orm.Session
Database session controlling the transaction.
central_latitude : float
Central latitude in decimal degrees.
central_longitude : float
Central longitude in decimal degrees.
Notes
-----
- Surrounding points are generated for bearings in ``BEARINGS`` at distances
in ``NEAR_DISTANCES`` and ``FAR_DISTANCES`` using great-circle formulas.
- Commits at the end to persist the batch.
See Also
--------
app.coordinates_utils.calculate_destination_point
"""
# Central point
session.add(
Coordinate(
latitude=central_latitude,
longitude=central_longitude,
label="center",
is_central=True,
)
)
# Surrounding points
for distance_km in NEAR_DISTANCES + FAR_DISTANCES:
for bearing_deg in BEARINGS:
destination_latitude, destination_longitude = destination_point(
central_latitude, central_longitude, distance_km, bearing_deg
)
session.add(
Coordinate(
latitude=destination_latitude,
longitude=destination_longitude,
label=f"{distance_km}km_{bearing_deg}deg",
is_central=False,
)
)
session.commit()
def _validate_central_coordinate(
session: Session,
central_latitude: float,
central_longitude: float,
) -> None:
"""Validate that the stored central coordinate matches the expected values.
Parameters
----------
session : sqlalchemy.orm.Session
Database session used for the query.
central_latitude : float
Expected latitude from configuration.
central_longitude : float
Expected longitude from configuration.
Raises
------
RuntimeError
If the central coordinate is missing or the values differ beyond the
configured tolerance ``COORDINATE_MATCH_TOLERANCE``.
"""
central_coordinate = session.query(Coordinate).filter_by(is_central=True).first()
if central_coordinate is None:
raise RuntimeError("No central coordinate found in DB; DB may be corrupted.")
assert central_coordinate is not None, (
"Invariant violated: central_coordinate must not be None "
"after existence check."
)
latitude_diff = abs(central_coordinate.latitude - central_latitude)
longitude_diff = abs(central_coordinate.longitude - central_longitude)
if (
latitude_diff > COORDINATE_MATCH_TOLERANCE
or longitude_diff > COORDINATE_MATCH_TOLERANCE
):
logger.error(
f"Central coordinate in DB ({central_coordinate.latitude:.6f}, "
f"{central_coordinate.longitude:.6f}) does not match .env "
f"({central_latitude:.6f}, {central_longitude:.6f})."
)
raise RuntimeError(
"Central coordinate mismatch. Backup your data/models and clear "
"the coordinates table to re-seed."
)
[docs]
def get_coordinates(session: Session) -> List[CoordinateSchema]:
"""Retrieve all coordinates as Pydantic schemas.
Parameters
----------
session : sqlalchemy.orm.Session
Open database session to run the query.
Returns
-------
list[app.schemas.CoordinateSchema]
List of coordinates converted from ORM objects using ``model_validate``.
Raises
------
sqlalchemy.exc.SQLAlchemyError
If an underlying database error occurs during the query.
Examples
--------
>>> # coords = get_coordinates(session) # doctest: +SKIP
>>> # isinstance(coords, list) # doctest: +SKIP
True
"""
coordinates = session.query(Coordinate).all()
return [CoordinateSchema.model_validate(coordinate) for coordinate in coordinates]
[docs]
def set_coordinate(
session: Session,
latitude: float,
longitude: float,
label: Optional[str] = None,
is_central: bool = False,
) -> None:
"""Add a new coordinate to the table.
Parameters
----------
session : sqlalchemy.orm.Session
Session used to persist the new coordinate.
latitude : float
Latitude in decimal degrees.
longitude : float
Longitude in decimal degrees.
label : str | None, optional
Optional label describing the coordinate.
is_central : bool, optional
Whether the coordinate represents the central point (default ``False``).
Notes
-----
- Commits immediately after inserting the object.
Raises
------
sqlalchemy.exc.SQLAlchemyError
If a database error occurs while persisting the new coordinate.
"""
coordinate = Coordinate(
latitude=latitude,
longitude=longitude,
label=label,
is_central=is_central,
)
session.add(coordinate)
session.commit()
logger.info(
f"Added coordinate: ({latitude:.6f}, {longitude:.6f}) "
f"label={label} is_central={is_central}"
)