52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from uuid import UUID
|
|
|
|
import sqlalchemy as sa
|
|
from geoalchemy2.elements import WKTElement
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.db.models import Track
|
|
from backend.app.models import TrackCreate
|
|
from backend.app.repositories.base import BaseRepository
|
|
|
|
|
|
class TrackRepository(BaseRepository[Track]):
|
|
model = Track
|
|
|
|
def __init__(self, session: Session) -> None:
|
|
super().__init__(session)
|
|
|
|
def list_all(self) -> list[Track]:
|
|
statement = sa.select(self.model)
|
|
return list(self.session.scalars(statement))
|
|
|
|
@staticmethod
|
|
def _ensure_uuid(value: UUID | str) -> UUID:
|
|
if isinstance(value, UUID):
|
|
return value
|
|
return UUID(str(value))
|
|
|
|
@staticmethod
|
|
def _line_string(coordinates: list[tuple[float, float]]) -> WKTElement:
|
|
if len(coordinates) < 2:
|
|
raise ValueError("Track geometry requires at least two coordinate pairs")
|
|
parts = [f"{lon} {lat}" for lat, lon in coordinates]
|
|
return WKTElement(f"LINESTRING({', '.join(parts)})", srid=4326)
|
|
|
|
def create(self, data: TrackCreate) -> Track:
|
|
coordinates = list(data.coordinates)
|
|
geometry = self._line_string(coordinates)
|
|
track = Track(
|
|
name=data.name,
|
|
start_station_id=self._ensure_uuid(data.start_station_id),
|
|
end_station_id=self._ensure_uuid(data.end_station_id),
|
|
length_meters=data.length_meters,
|
|
max_speed_kph=data.max_speed_kph,
|
|
is_bidirectional=data.is_bidirectional,
|
|
status=data.status,
|
|
track_geometry=geometry,
|
|
)
|
|
self.session.add(track)
|
|
return track
|