App/default inputs

.pre-commit-config.yaml

@@ -10,6 +10,7 @@ repos:
       - id: check-toml
       - id: check-docstring-first
       - id: check-added-large-files
+        args: ["--maxkb=1500"]
       - id: detect-private-key
   - repo: https://github.com/charliermarsh/ruff-pre-commit
     rev: 'v0.0.254'

app/main.py

@@ -1,24 +1,26 @@
 """Main script to run for streamlit app."""
 
-import datetime as dt
-
 import numpy as np
 import streamlit as st
 from water_tracker import connectors
-from water_tracker.display import chronicles
+from water_tracker.display import chronicles, defaults, inputs
 from water_tracker.transformers import trends
 
 default_start_date = "2022-01-01"
 default_end_date = "2022-12-31"
 st.set_page_config(page_title="Water-Tracker", layout="wide")
 st.title("Water-Tracker")
 
-departements = [*list(range(1, 20)), "2A", "2B", *list(range(21, 96))]
-
-code_departement = st.selectbox(
-    label="Sélection du département",
-    options=[str(dpt).zfill(2) for dpt in departements],
+dept_input = inputs.DepartmentInput(
+    "Sélection du Département",
+    defaults.DefaultDepartement(
+        st.experimental_get_query_params(),
+        "longitude",
+        "latitude",
+    ),
 )
+code_departement = dept_input.build(st.container())
+
 stations_connector = connectors.PiezoStationsConnector()
 stations_params = {
     "code_departement": code_departement,
@@ -34,52 +36,30 @@
 unknown_name = valid_stations["nom_commune"].isna()
 valid_stations.loc[unknown_name, "nom_commune"] = "Commune Inconnue"
 
-
-def format_func(row_id: int) -> str:
-    """Format the bss code display name.
-
-    Parameters
-    ----------
-    row_id : int
-        Index of the row to display the code of.
-
-    Returns
-    -------
-    str
-        'bss_code (city name)'
-    """
-    bss_code = valid_stations.loc[row_id, "code_bss"]
-    city_name = valid_stations.loc[row_id, "nom_commune"]
-    return f"{bss_code} ({city_name})"
-
-
-bss_code_id = st.selectbox(
-    label="Sélection du code bss de la station",
-    options=valid_stations.index,
-    format_func=format_func,
+station_input = inputs.StationInput(
+    label="Sélection du code BSS de la station",
+    stations_df=valid_stations,
+    default_input=defaults.DefaultStation(
+        stations_df=valid_stations,
+    ),
+    bss_field_name="code_bss",
+    city_field_name="nom_commune",
 )
+bss_code_id = station_input.build(st.container())
+
 bss_code = valid_stations.loc[bss_code_id, "code_bss"]
 min_date = valid_stations.loc[bss_code_id, "date_debut_mesure"]
 max_date = valid_stations.loc[bss_code_id, "date_fin_mesure"]
-col1, col2 = st.columns(2)
-default = max((max_date - dt.timedelta(days=365)), min_date)
-mesure_date_start = col1.date_input(
-    label="Date de début de mesure",
-    value=default,
-    max_value=max_date,
-    min_value=min_date,
-)
-if type(mesure_date_start) == dt.date:
-    min_date_end = mesure_date_start
-else:
-    min_date_end = min_date
 
-mesure_date_end = col2.date_input(
-    label="Date de fin de mesure",
-    value=max_date,
-    max_value=max_date,
-    min_value=min_date_end,
+period_input = inputs.PeriodInput(
+    "Date de début de mesure",
+    "Date de fin de mesure",
+    min_date,
+    max_date,
+    defaults.DefaultMinDate(min_date, max_date),
+    defaults.DefaultMaxDate(min_date, max_date),
 )
+date_start, date_end = period_input.build(st.container())
 
 # Chronicles
 
@@ -110,8 +90,8 @@ def format_func(row_id: int) -> str:
 chronicle_connector = connectors.PiezoChroniclesConnector()
 chronicles_params = {
     "code_bss": bss_code,
-    "date_debut_mesure": mesure_date_start,
-    "date_fin_mesure": mesure_date_end,
+    "date_debut_mesure": date_start,
+    "date_fin_mesure": date_end,
 }
 chronicles_df = chronicle_connector.retrieve(chronicles_params)
 display_trend = False

pyproject.toml

@@ -19,6 +19,8 @@ python-dotenv = "^1.0.0"
 netcdf4 = "^1.6.3"
 scipy = "^1.10.1"
 plotly = "^5.14.1"
+geopandas = "^0.13.0"
+shapely = "^2.0.1"
 
 
 [tool.poetry.group.dev.dependencies]

src/water_tracker/__init__.py

@@ -1 +1,5 @@
 """Provide collection and visualisation tools for water-related variables."""
+
+from pathlib import Path
+
+BASE_DIR = Path(__file__).parent.resolve()

src/water_tracker/display/defaults.py

@@ -0,0 +1,207 @@
+"""Tools to compute defaults values for user inputs."""
+
+import datetime as dt
+from abc import ABC, abstractproperty
+from functools import cached_property
+from pathlib import Path
+from typing import Generic, TypeVar
+
+import geopandas as gpd
+import pandas as pd
+from shapely import Point
+
+from water_tracker import BASE_DIR
+
+depts_geojson_path_relative = Path("resources/departments_geojson.json")
+depts_geojson_path = BASE_DIR.joinpath(depts_geojson_path_relative)
+
+DefaultValueT = TypeVar("DefaultValueT")
+
+
+class DefaultInput(ABC, Generic[DefaultValueT]):
+    """Base class for default user inputs."""
+
+    @abstractproperty
+    def value(self) -> DefaultValueT:
+        """Value to use as default."""
+
+
+class DefaultDepartement(DefaultInput[str]):
+    """Default input for department selection from query_params.
+
+    Parameters
+    ----------
+    query_params : dict[str, list[str]]
+        Query parameters shown in the browser's url.
+    longitude_query_param : str
+        Name of the latitude field.
+    latitude_query_param : str
+        Name of the longitude field.
+    depts_geojson_path : Path, optional
+        Path to the geojson with departments boundaries.
+        , by default depts_geojson_path
+    geojson_code_field : str, optional
+        Geojson field with departments code., by default "code"
+    """
+
+    _default_dept_nb: str = "01"
+
+    def __init__(
+        self,
+        query_params: dict[str, list[str]],
+        longitude_query_param: str,
+        latitude_query_param: str,
+        depts_geojson_path: Path = depts_geojson_path,
+        geojson_code_field: str = "code",
+    ) -> None:
+        self.lat_param = latitude_query_param
+        self.lon_param = longitude_query_param
+        self._depts_path = depts_geojson_path
+        self.geojson_code_field = geojson_code_field
+        self.query_params = query_params
+
+    @property
+    def default_value(self) -> str:
+        """Default value."""
+        return self._default_dept_nb
+
+    @cached_property
+    def departments_geojson(self) -> gpd.GeoDataFrame:
+        """Geodataframe with departments polygons."""
+        return gpd.read_file(self._depts_path)
+
+    @property
+    def query_params(self) -> dict:
+        """Query parameters."""
+        return self._query
+
+    @query_params.setter
+    def query_params(self, query_params: dict[str, list[str]]) -> None:
+        self._query = query_params
+        self._valid_query_params = self.check_params()
+
+    @property
+    def value(self) -> str:
+        """Value to use a default for department."""
+        if self._valid_query_params:
+            point = self.read_query_params()
+            return self.get_point_department(point)
+        return self.default_value
+
+    def check_params(self) -> bool:
+        """Check if the query parameters have latitude and longitude fields.
+
+        Returns
+        -------
+        bool
+            True if the query_parameters has fields for longitude and latitude.
+        """
+        is_lat_in_key = self.lat_param in self.query_params
+        is_lon_in_key = self.lon_param in self.query_params
+        return is_lat_in_key and is_lon_in_key
+
+    def read_query_params(self) -> Point:
+        """Read the query parameters and returns the corresponding Point.
+
+        Returns
+        -------
+        Point
+            Point corresponding to the query parameters.
+        """
+        lat = float(self.query_params[self.lat_param][0])
+        lon = float(self.query_params[self.lon_param][0])
+        return Point(lon, lat)
+
+    def get_point_department(self, point: Point) -> str:
+        """Find department to which the point belongs to.
+
+        Parameters
+        ----------
+        point : Point
+            Point to find the departement of.
+
+        Returns
+        -------
+        str
+            Departement's code.
+        """
+        contains_point = self.departments_geojson.contains(point)
+        if not contains_point.any():
+            return self.default_value
+        first_containing = self.departments_geojson[contains_point].iloc[0]
+        return first_containing[self.geojson_code_field]
+
+
+class DefaultStation(DefaultInput[int]):
+    """Default input for Station Selection.
+
+    Parameters
+    ----------
+    stations_df : pd.DataFrame
+        DataFrame with all stations.
+    """
+
+    def __init__(self, stations_df: pd.DataFrame) -> None:
+        self._stations_index = stations_df.index
+
+    @property
+    def stations_index(self) -> pd.Index:
+        """Indexes of the station DataFrame."""
+        return self._stations_index
+
+    @property
+    def value(self) -> int:
+        """Value to use as default input."""
+        return self.stations_index[0]
+
+
+class DefaultDate(DefaultInput[dt.date]):
+    """Default Input for date selection.
+
+    Parameters
+    ----------
+    min_date : dt.date
+        Data minimum date.
+    max_date : dt.date
+        Data maximum date.
+    """
+
+    def __init__(self, min_date: dt.date, max_date: dt.date) -> None:
+        self._min = min_date
+        self._max = max_date
+
+
+class DefaultMinDate(DefaultDate):
+    """Default Input for minimum date selection.
+
+    Parameters
+    ----------
+    min_date : dt.date
+        Data minimum date.
+    max_date : dt.date
+        Data maximum date.
+    """
+
+    @property
+    def value(self) -> dt.date:
+        """Value to use as default input."""
+        max_minus_year = self._max - dt.timedelta(days=365)
+
+        return max(self._min, max_minus_year)
+
+
+class DefaultMaxDate(DefaultDate):
+    """Default Input for maximum date selection.
+
+    Parameters
+    ----------
+    min_date : dt.date
+        Data minimum date.
+    max_date : dt.date
+        Data maximum date.
+    """
+
+    @property
+    def value(self) -> dt.date:
+        """Value to use as default input."""
+        return self._max