Introducing observation refreshes things

arpav_ppcv/cliapp/app.py

@@ -1,11 +1,7 @@
 import datetime as dt
 import uuid
-from typing import (
-    Annotated,
-    Optional,
-)
+from typing import Optional
 
-import geojson_pydantic
 import pydantic_core
 import sqlmodel
 import typer
@@ -17,60 +13,20 @@
     observations,
 )
 from . import schemas
+from .climaticindicators import app as climatic_indicators_app
+from .observations import (
+    series_configurations_app,
+    stations_app,
+)
 
 app = typer.Typer()
+app.add_typer(climatic_indicators_app, name="climatic-indicators")
+app.add_typer(stations_app, name="observation-stations")
+app.add_typer(series_configurations_app, name="observation-series-configurations")
 
 _JSON_INDENTATION = 2
 
 
-@app.command(name="list-stations")
-def list_stations(ctx: typer.Context) -> None:
-    """List stations."""
-    with sqlmodel.Session(ctx.obj["engine"]) as session:
-        result = [
-            schemas.StationRead(**s.model_dump())
-            for s in database.collect_all_stations(session)
-        ]
-        print(pydantic_core.to_json(result, indent=_JSON_INDENTATION).decode("utf-8"))
-
-
-@app.command(name="create-station", context_settings={"ignore_unknown_options": True})
-def create_station(
-    ctx: typer.Context,
-    code: str,
-    longitude: Annotated[float, typer.Argument(min=-180, max=180)],
-    latitude: Annotated[float, typer.Argument(min=-90, max=90)],
-    altitude: Annotated[float, typer.Option(min=-50, max=10_000)] = None,
-    name: Annotated[str, typer.Option(help="Station name")] = "",
-    type: Annotated[str, typer.Option(help="Station type")] = "",
-) -> None:
-    station_create = schemas.StationCreate(
-        geom=geojson_pydantic.Point(type="Point", coordinates=(longitude, latitude)),
-        code=code,
-        altitude_m=altitude,
-        name=name,
-        type_=type.lower().replace(" ", "_"),
-    )
-    """Create a new station."""
-    with sqlmodel.Session(ctx.obj["engine"]) as session:
-        db_station = database.create_station(session, station_create)
-        print(
-            schemas.StationRead(**db_station.model_dump()).model_dump_json(
-                indent=_JSON_INDENTATION
-            )
-        )
-
-
-@app.command(name="delete-station")
-def delete_station(
-    ctx: typer.Context,
-    station_id: uuid.UUID,
-) -> None:
-    """Delete a station."""
-    with sqlmodel.Session(ctx.obj["engine"]) as session:
-        database.delete_station(session, station_id)
-
-
 @app.command(name="list-monthly-measurements")
 def list_monthly_measurements(ctx: typer.Context) -> None:
     """List monthly measurements."""

arpav_ppcv/cliapp/climaticindicators.py

@@ -0,0 +1,32 @@
+import sqlmodel
+import typer
+from operator import attrgetter
+from rich import print
+
+from . import schemas
+from .. import database
+
+app = typer.Typer()
+
+
+@app.command(name="list")
+def list_climatic_indicators(ctx: typer.Context) -> None:
+    """List climatic indicators."""
+    with sqlmodel.Session(ctx.obj["engine"]) as session:
+        all_indicators = list(database.collect_all_climatic_indicators(session))
+        all_indicators.sort(key=attrgetter("sort_order"))
+        for db_climatic_indicator in all_indicators:
+            item = schemas.ClimaticIndicatorItem(**db_climatic_indicator.model_dump())
+            print(item.identifier)
+
+
+@app.command(name="get")
+def get_climatic_indicator(ctx: typer.Context, identifier: str) -> None:
+    with sqlmodel.Session(ctx.obj["engine"]) as session:
+        if (
+            db_climatic_indicator := database.get_climatic_indicator_by_identifier(
+                session, identifier
+            )
+        ) is not None:
+            item = schemas.ClimaticIndicatorDetail(**db_climatic_indicator.model_dump())
+            print(item)

arpav_ppcv/cliapp/observations.py

@@ -0,0 +1,64 @@
+import sqlmodel
+import typer
+from operator import attrgetter
+from rich import print
+
+from . import schemas
+from .. import database
+
+stations_app = typer.Typer()
+series_configurations_app = typer.Typer()
+
+
+@stations_app.command(name="list")
+def list_observation_stations(ctx: typer.Context) -> None:
+    """List observation stations."""
+    with sqlmodel.Session(ctx.obj["engine"]) as session:
+        all_stations = list(database.collect_all_observation_stations(session))
+        all_stations.sort(key=attrgetter("code"))
+        for db_station in all_stations:
+            item = schemas.ObservationStationItem(**db_station.model_dump())
+            print(f"{item.code} - {item.name}")
+
+
+@stations_app.command(name="get")
+def get_observation_station(ctx: typer.Context, code: str) -> None:
+    with sqlmodel.Session(ctx.obj["engine"]) as session:
+        if (
+            db_station := database.get_observation_station_by_code(session, code)
+        ) is not None:
+            item = schemas.ObservationStationDetail(**db_station.model_dump())
+            print(item)
+
+
+@series_configurations_app.command(name="list")
+def list_observation_series_configurations(ctx: typer.Context) -> None:
+    """List observation series configurations."""
+    with sqlmodel.Session(ctx.obj["engine"]) as session:
+        all_series_configurations = list(
+            database.collect_all_observation_series_configurations(session)
+        )
+        for db_series_configuration in all_series_configurations:
+            item = schemas.ObservationSeriesConfigurationItem(
+                **db_series_configuration.model_dump()
+            )
+            print(item.identifier)
+
+
+@series_configurations_app.command(name="get")
+def get_observation_series_configuration(ctx: typer.Context, identifier: str) -> None:
+    with sqlmodel.Session(ctx.obj["engine"]) as session:
+        if (
+            db_series_configuration := (
+                database.get_observation_series_configuration_by_identifier(
+                    session, identifier
+                )
+            )
+        ) is not None:
+            item = schemas.ObservationSeriesConfigurationDetail(
+                **db_series_configuration.model_dump(exclude={"climatic_indicator_id"}),
+                climatic_indicator=schemas.ClimaticIndicatorItem(
+                    **db_series_configuration.climatic_indicator.model_dump()
+                ),
+            )
+            print(item)

arpav_ppcv/cliapp/schemas.py

@@ -1,15 +1,56 @@
-from ..schemas import observations
+import pydantic
 
+from ..schemas import (
+    observations,
+    static,
+)
 
-class StationRead(observations.StationBase):
-    ...
+
+class ClimaticIndicatorItem(pydantic.BaseModel):
+    identifier: str
+
+
+class ClimaticIndicatorDetail(pydantic.BaseModel):
+    identifier: str
+    name: str
+    measure_type: static.MeasureType
+    aggregation_period: static.AggregationPeriod
+    display_name_english: str
+    description_english: str
+    unit_english: str
+    palette: str
+    color_scale_min: float
+    color_scale_max: float
+    data_precision: float
+
+
+class ObservationStationItem(pydantic.BaseModel):
+    code: str
+    name: str
+
+
+class ObservationStationDetail(observations.StationBase):
+    code: str
+    name: str
+
+
+class ObservationSeriesConfigurationItem(pydantic.BaseModel):
+    identifier: str
+
+
+class ObservationSeriesConfigurationDetail(pydantic.BaseModel):
+    identifier: str
+    climatic_indicator: ClimaticIndicatorItem
+    indicator_internal_name: str
+    measurement_aggregation_type: static.MeasurementAggregationType
+    station_owners: list[static.ObservationStationOwner]
 
 
-class StationCreate(observations.StationCreate):
+class ObservationStationCreate(observations.StationCreate):
     ...
 
 
-class StationUpdate(observations.StationUpdate):
+class ObservationStationUpdate(observations.StationUpdate):
     ...