Use Pydantic to manage parameter metadata and validation

docs/source/model_steps.md

@@ -8,15 +8,15 @@ Source : https://github.com/mobility-team/mobility/issues/145#issuecomment-32280
 Le fonctionnement actuel est le suivant :
 
 Initialisation :
-- Génération des séquences de motifs de déplacement dans chaque zone de transport, selon le profil de la population résidente (CSP, nombre de voitures du ménage, type de catégorie urbaine de la commune), et des besoins en heures d'activité pour chaque étape des séquences.
-- Calcul des opportunités disponibles (=heures d'activités disponibles) par motif, pour chaque zone de transport.
+- Génération des séquences de motifs de déplacement dans chaque zone de transport, selon le profil de la population résidente (CSP, nombre de voitures du ménage, type de catégorie urbaine de la commune), et des besoins en heures d'activité pour chaque étape des séquences.
+- Calcul des opportunités disponibles (=heures d'activités disponibles) par motif, pour chaque zone de transport.
 
 Boucle :
-- Calcul des coûts généralisés de transport pour chaque couple motif - origine - destination (sans congestion pour la première itération).
-- Calcul des probabilités de choisir une destination en fonction du motif et de l'origine du déplacement ainsi que du lieu de résidence des personnes.
-- Echantillonnage d'une séquence de destinations pour chaque séquence de motifs, zone de transport de résidence et CSP.
-- Recherche des top k séquences de modes disponibles pour réaliser ces séquences de déplacements (k<=10)
-- Calcul des flux résultants par OD et par mode, puis recalcul des coûts généralisés.
-- Calcul d'une part de personnes qui vont changer d'assignation séquence de motifs + modes (en fonction de la saturation des opportunités à destination, de possibilités d'optimisation comparatives, et d'une part de changements aléatoires).
-- Calcul des opportunités restantes à destination.
-- Recommencement de la procédure avec cette part de personnes non assignées.
+- Calcul des coûts généralisés de transport pour chaque couple motif - origine - destination (sans congestion pour la première itération).
+- Calcul des probabilités de choisir une destination en fonction du motif et de l'origine du déplacement ainsi que du lieu de résidence des personnes.
+- Echantillonnage d'une séquence de destinations pour chaque séquence de motifs, zone de transport de résidence et CSP.
+- Recherche des top k séquences de modes disponibles pour réaliser ces séquences de déplacements (k<=10)
+- Calcul des flux résultants par OD et par mode, puis recalcul des coûts généralisés.
+- Calcul d'une part de personnes qui vont changer d'assignation séquence de motifs + modes (en fonction de la saturation des opportunités à destination, de possibilités d'optimisation comparatives, et d'une part de changements aléatoires).
+- Calcul des opportunités restantes à destination.
+- Recommencement de la procédure avec cette part de personnes non assignées.

mobility/asset.py

@@ -7,6 +7,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import is_dataclass, fields
 from pandas.util import hash_pandas_object
+from pydantic import BaseModel
 
 class Asset(ABC):
     """
@@ -78,6 +79,9 @@ def serialize(value):
                 attr_hash = hash_pandas_object(value.drop(columns="geometry")).sum()
                 return hashlib.sha256((geom_hash + str(attr_hash)).encode()).hexdigest()
 
+            elif isinstance(value, BaseModel):
+                return value.model_dump(mode="json")
+
             else:
                 return value
 

mobility/choice_models/population_trips.py

@@ -244,8 +244,6 @@ def resolve_parameters(
 
             parameters = PopulationTripsParameters(**old_args)
 
-        parameters.validate()
-
         return parameters
 
 

mobility/choice_models/population_trips_parameters.py

@@ -1,24 +1,148 @@
-from dataclasses import dataclass
-
-@dataclass(frozen=True)
-class PopulationTripsParameters:
-    n_iterations: int = 1
-    alpha: float = 0.01
-    k_mode_sequences: int = 6
-    dest_prob_cutoff: float = 0.99
-    n_iter_per_cost_update: int = 3
-    cost_uncertainty_sd: float = 1.0
-    seed: int = 0
-    mode_sequence_search_parallel: bool = True
-    min_activity_time_constant: float = 1.0
-    simulate_weekend: bool = False
-
-    def validate(self) -> None:
-        assert self.n_iterations >= 1
-        assert 0.0 < self.dest_prob_cutoff <= 1.0
-        assert self.alpha >= 0.0
-        assert self.k_mode_sequences >= 1
-        assert self.n_iter_per_cost_update >= 0
-        assert self.cost_uncertainty_sd > 0.0
-        assert self.seed >= 0
-        assert self.min_activity_time_constant >= 0
+from pydantic import BaseModel, Field, ConfigDict
+from typing import Annotated
+
+
+class PopulationTripsParameters(BaseModel):
+
+    model_config = ConfigDict(extra="forbid")
+
+    n_iterations: Annotated[
+        int,
+        Field(
+            default=1,
+            ge=1,
+            title="Number of iterations",
+            description=(
+                "Number of simulation iterations used to compute the population "
+                "trips. Increase this to get more diverse programmes and to allow "
+                "congestion feedbacks to propagate."
+            ),
+        ),
+    ]
+
+    alpha: Annotated[
+        float,
+        Field(
+            default=0.01,
+            ge=0.0,
+            title="Next anchor cost weighting",
+            description=(
+                "Weight of the cost to get to the next anchor destination in the "
+                "chain when considering destination options (shopping place when "
+                "the next anchor is work, for example) and computing probabilities."
+            ),
+        ),
+    ]
+
+    k_mode_sequences: Annotated[
+        int,
+        Field(
+            default=6,
+            ge=1,
+            title="Number of mode combinations",
+            description=(
+                "Number of mode combinations considered in the simulation, for "
+                "a given destination sequence. Only the top k combinations are "
+                "considered."
+            ),
+        ),
+    ]
+
+    dest_prob_cutoff: Annotated[
+        float,
+        Field(
+            default=0.99,
+            gt=0.0,
+            le=1.0,
+            title="Destination probability distribution cutoff",
+            description=(
+                "Cutoff used to prune the less probable destinations for a given "
+                "origin. Only the first dest_prob_cutoff % of the cumulative "
+                "distribution is considered."
+            ),
+        ),
+    ]
+
+    n_iter_per_cost_update: Annotated[
+        int,
+        Field(
+            default=3,
+            ge=0,
+            title="Travel costs update period",
+            description=(
+                "The simulation will update the travel costs every n_iter_per_cost_update. ",
+                "Set n_iter_per_cost_update to zero to ignore congestion in the "
+                "simulation, set to 1 to update congestion at each iteration, " 
+                "set to a higher level to speed up the simulation."
+            ),
+        ),
+    ]
+
+    cost_uncertainty_sd: Annotated[
+        float,
+        Field(
+            default=1.0,
+            gt=0.0,
+            title="Standard deviation of travel costs estimates",
+            description=( 
+                "Travel costs are estimated between specific representative points " 
+                "located in transport zones, but the actual travel costs between "
+                "all origins and all destinations in the transport zones will be "
+                "slightly different than these point estimates. "
+                "cost_uncertainty_sd controls how uncertain are these estimates, "
+                "and spreads the opportunities in the destination transport zone "
+                "based on a normal distribution centered on the point estimates "
+                "and a standard deviation of cost_uncertainty_sd."
+            ),
+        ),
+    ]
+
+    seed: Annotated[
+        int,
+        Field(
+            default=0,
+            ge=0,
+            title="Simulation seed",
+            description=(
+                "Seed used to get reproducible results for stochastic simulation "
+                "steps (like destination sampling when building destination chains). "
+                "Change this value to get new programmes and results for a given " 
+                "set of inputs."
+            ),
+        ),
+    ]
+
+    mode_sequence_search_parallel: Annotated[
+        bool,
+        Field(
+            default=True,
+            title="Parallel mode for the top k mode sequence search",
+            description=(
+                "Set to False to debug or for small simulations, otherwise set " 
+                "to True to speed up the simulation."
+            ),
+        ),
+    ]
+
+    min_activity_time_constant: Annotated[
+        float,
+        Field(
+            default=1.0,
+            ge=0.0,
+            title="Minimum activity time coefficient",
+            description=(
+                "Coefficient controlling the minimum activity time necessary to "
+                "get a positive utility from the activity. This minimum time is "
+                "equal to average_activity_time x exp(-min_activity_time_constant)."
+            ),
+        ),
+    ]
+
+    simulate_weekend: Annotated[
+        bool,
+        Field(
+            default=False,
+            title="Week day only or week day + weekend day mode",
+            description="Wether to simulate a weekend day or only a week day.",
+        ),
+    ]

mobility/parsers/mobility_survey/mobility_survey.py

@@ -50,13 +50,11 @@ def get_cached_asset(self) -> dict[str, pd.DataFrame]:
 
     def get_chains_probability(self, motives, modes):
 
-        motive_mapping = [{"group": m.name, "motive": m.survey_ids} for m in motives]
+        motive_mapping = [{"group": m.name, "motive": m.survey_ids} for m in motives if m.name != "other"]
         motive_mapping = pd.DataFrame(motive_mapping)
         motive_mapping = motive_mapping.explode("motive")
         motive_mapping = motive_mapping.set_index("motive").to_dict()["group"]
 
-        motive_names = [m.name for m in motives]
-
         mode_mapping = [
             {"group": m.name, "mode": m.survey_ids} 
             for m in modes if len(m.survey_ids) > 0
@@ -173,24 +171,12 @@ def get_chains_probability(self, motives, modes):
 
             # Map detailed motives to grouped motives
             .with_columns(
-                pl.col("motive").replace(motive_mapping)
-            )
-            .with_columns(
-                motive=( 
-                    pl.when(pl.col("motive").is_in(motive_names))
-                    .then(pl.col("motive"))
-                    .otherwise(pl.lit("other"))
-                )
+                pl.col("motive").replace_strict(motive_mapping, default="other")
             )
 
             # Map detailed modes to grouped modes
             .with_columns(
-                mode=pl.col("mode_id").replace(mode_mapping)
-            )
-            .with_columns(
-                mode=pl.when(pl.col("mode").is_in(mode_names))
-                .then(pl.col("mode"))
-                .otherwise(pl.lit("other"))
+                mode=pl.col("mode_id").replace_strict(mode_mapping, default="other")
             )
 
             # Remove motive sequences that are longer than 10 motives to speed 

mobility/population.py

@@ -258,7 +258,7 @@ def get_sample_sizes(self, lau_to_tz_coeff: pd.DataFrame, sample_size: int):
                     transport zones will be set to zero.
                 """
             )
-            population["legal_population"].fillna(0.0, inplace=True)
+            population["legal_population"] = population["legal_population"].fillna(0.0)
 
         population["n_persons"] = sample_size*population["legal_population"].pow(0.5)/population["legal_population"].pow(0.5).sum()
         population["n_persons"] = np.ceil(population["n_persons"])

pyproject.toml

@@ -27,7 +27,8 @@ dependencies = [
   "plotly",
   "scikit-learn",
   "gtfs_kit",
-  "kaleido"
+  "kaleido",
+  "pydantic",
 ]
 
 requires-python = ">=3.11"