"""Coordinate management for seeding, retrieval, and central-point validation.This module provides a focused set of utilities for initializing and validatingthe coordinate grid used by the weather pipeline. It inserts a single centralcoordinate and a deterministic ring of surrounding coordinates, and exposesread helpers that return Pydantic schemas. The goal is to keep persistence andvalidation responsibilities decoupled from input validation and orchestrationlogic 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: +SKIPTrue"""importloggingfromtypingimportList,Optionalfromsqlalchemy.ormimportSessionfrom.coordinates_utilsimportcalculate_destination_pointasdestination_pointfrom.modelsimportCoordinatefrom.schemasimportCoordinateSchemalogger=logging.getLogger(__name__)# Constants for coordinate generationBEARINGS:list[int]=[0,45,90,135,180,225,270,315]NEAR_DISTANCES:list[float]=[1.0,5.0]# kmFAR_DISTANCES:list[float]=[50.0,150.0]# kmCOORDINATE_MATCH_TOLERANCE:float=1e-6# Threshold for coordinate match in degreesdefseed_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()ifnotexisting_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 pointsession.add(Coordinate(latitude=central_latitude,longitude=central_longitude,label="center",is_central=True,))# Surrounding pointsfordistance_kminNEAR_DISTANCES+FAR_DISTANCES:forbearing_deginBEARINGS: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()ifcentral_coordinateisNone:raiseRuntimeError("No central coordinate found in DB; DB may be corrupted.")assertcentral_coordinateisnotNone,("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_TOLERANCEorlongitude_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}).")raiseRuntimeError("Central coordinate mismatch. Backup your data/models and clear ""the coordinates table to re-seed.")defget_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)forcoordinateincoordinates]defset_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}")