[US] Établir un rapport de tous les paramètres d'entrée #198

examples/quickstart-fr.py.py

@@ -38,3 +38,11 @@
 # You can also plot the flows, with labels for the cities that are bigger than their neighbours
 labels = pop_trips.get_prominent_cities()
 pop_trips.plot_od_flows(labels=labels)
+
+# You can print a report of all parameters used in the model
+#report = pop_trips.parameters_dict()
+#print(report.T)
+
+tz_params = transport_zones.get_parameters()
+for p in tz_params:
+    print(p.to_dict())

mobility/asset.py

@@ -82,7 +82,7 @@ def serialize(value):
                 return value
 
         hashable_inputs = {k: serialize(v) for k, v in self.inputs.items()}
-        serialized_inputs = json.dumps(hashable_inputs, sort_keys=True).encode('utf-8')
+        serialized_inputs = json.dumps(hashable_inputs, sort_keys=True,default=lambda o: o.to_dict() if hasattr(o, 'to_dict') else str(o)).encode('utf-8')
 
         return hashlib.md5(serialized_inputs).hexdigest()
 

mobility/choice_models/population_trips.py

@@ -4,6 +4,7 @@
 import shutil
 import random
 import warnings
+import pandas as pd
 
 import geopandas as gpd
 import matplotlib.pyplot as plt
@@ -685,8 +686,9 @@ def plot_od_flows(self, mode="all", motive="all", period="weekdays", level_of_de
         # Put a legend for width on bottom right, title on the top
         x_min = float(biggest_flows[["x"]].min().iloc[0])
         y_min = float(biggest_flows[["y"]].min().iloc[0])
-        plt.plot([x_min, x_min+4000], [y_min, y_min], linewidth=2, color=color)
-        plt.text(x_min+6000, y_min-1000, "1 000", color=color)
+        plt.plot([x_min-6000, x_min-4000], [y_min, y_min], linewidth=2, color=color)
+        plt.text(x_min-2000, y_min-200, "1 000", color=color)
+        plt.text(x_min-6000, y_min-2000, f"hash: {self.inputs_hash}", fontsize=7, color=color)
         plt.title(f"{mode_name} flows between transport zones on {period}")
 
         # Draw all origin-destinations
@@ -717,7 +719,7 @@ def get_prominent_cities(self, n_cities=20, n_levels=3, distance_km=2):
         """
         Get the most prominent cities, ie the biggest cities that are not close to a bigger city.
 
-        Useful to label a map and reducing the number of overlaps without mising an important city.
+        Useful to label a map and reducing the number of overlaps without missing an important city.
 
         Parameters
         ----------
@@ -771,3 +773,38 @@ def get_prominent_cities(self, n_cities=20, n_levels=3, distance_km=2):
         geoflows = geoflows.merge(xy_coords, left_index=True, right_index=True)
 
         return geoflows
+
+    def parameters_dict(self) :
+        params_general = {
+            "inner_radius": self.population.transport_zones.inner_radius,
+            "local_admin_unit_id": self.population.transport_zones.study_area.local_admin_unit_id,
+            "level_of_detail" : self.population.transport_zones.level_of_detail,
+            "nb_local_admin_units": len(self.population.transport_zones.study_area.get()),
+            "osm_geofabrik_extract_date": self.population.transport_zones.osm_buildings.geofabrik_extract_date,
+            "population_sample_size": self.population.sample_size,
+            "survey_used": [s.survey_name for s in self.surveys],
+            "inputs_hash" : self.inputs_hash
+        }
+        params_modes = {
+            key: value
+            for i, m in enumerate(self.modes, start=1)
+            for key, value in [
+                (f"mode_{i}", m.name),
+                (f"mode_{i}_filter_max_time",m.travel_costs.routing_parameters.filter_max_time),
+                (f"mode_{i}_filter_max_speed",m.travel_costs.routing_parameters.filter_max_speed),
+                (f"mode_{i}_cost_constant",m.generalized_cost.parameters.cost_constant),
+                (f"mode_{i}_cost_of_distance",m.generalized_cost.parameters.cost_of_distance),
+                (f"mode_{i}_cost_of_time_intercept",m.generalized_cost.parameters.cost_of_time.intercept) #à voir ce qu'on veut connaitre
+            ]
+        }
+        params_motives = {
+            key: value
+            for i, m in enumerate(self.motives, start=1)
+            for key, value in [
+                (f"motive_{i}", m.name),
+                (f"motive_{i}_value_of_time",m.value_of_time),
+                (f"motive_{i}_value_of_time_v2",m.value_of_time_v2)
+            ]
+        }
+        params = params_general | params_modes | params_motives
+        return pd.DataFrame([params])

mobility/file_asset.py

@@ -3,7 +3,8 @@
 import networkx as nx
 
 from mobility.asset import Asset
-from typing import Any
+from mobility.model_parameters import Parameter
+from typing import Any, List
 from abc import abstractmethod
 
 class FileAsset(Asset):
@@ -206,4 +207,25 @@ def remove(self):
             path = pathlib.Path(self.cache_path)
             if path.exists():
                 path.unlink()
+
+    def get_parameters(self, recursive: bool = True) -> List[Parameter]:
+
+        params = []
+
+        for inp in self.inputs.values():
+            if isinstance(inp, Parameter) :
+                params.append(inp)
+
+        if recursive:
+            for inp in self.inputs.values():
+                if isinstance(inp, FileAsset):
+                    params.extend(inp.get_parameters(recursive=True))
+
+        unique_params = {}
+        for p in params:
+            if p.name not in unique_params:
+                unique_params[p.name] = p
+
+        return list(unique_params.values())
+
 

mobility/model_parameters.py

@@ -0,0 +1,114 @@
+from dataclasses import dataclass, field, fields, asdict
+from typing import List, Union
+
+Number = Union[int, float]    
+
+@dataclass(frozen=True)
+class Parameter:
+    name: str
+    name_fr: str
+    value: Number | bool
+    description: str
+    parameter_type: type
+    default_value: Number | bool
+    possible_values: List[float] | List[int] | tuple = None
+    min_value: Number = None
+    max_value: Number = None
+    unit: str = None
+    interval: Number = None
+    source_default: str = ""
+    parameter_role: str = ""
+
+    def to_dict(self):
+        # Convert the parameter to a dictionary with serializable values
+        return {
+            "name": self.name,
+            "name_fr": self.name_fr,
+            "value": self.value,
+            "description": self.description,
+            "parameter_type": str(self.parameter_type),  # Convert type to string
+            "possible_values": self.possible_values,
+            "min_value": self.min_value,
+            "max_value": self.max_value,
+            "unit": self.unit,
+            "interval": self.interval,
+            "source_default": self.source_default,
+            "parameter_role": self.parameter_role,
+        }
+
+    # def get(self):
+    #     """Return parameter value."""
+    #     val = self.default_value
+    #     self._validate(val)
+    #     return val
+
+    def set(self, new_value):
+        self.value = new_value
+
+    def validate(self):
+        if self.value is None: #todo: improve this!
+            return None
+        if self.parameter_type is not None:
+            if not isinstance(self.value, self.parameter_type):
+                t = type(self.value)
+                raise TypeError(f"Parameter '{self.name}' must be {self.parameter_type} (currently {t}).")
+
+        if self.min_value is not None and self.value < self.min_value:
+            raise ValueError(
+                f"Parameter '{self.name}' below minimum {self.min_value}"
+            )
+
+        if self.max_value is not None and self.value > self.max_value:
+            raise ValueError(
+                f"Parameter '{self.name}' above maximum {self.max_value}"
+            )
+
+    def __repr__(self):
+        unit_str = f" [{self.unit}]" if self.unit else ""
+        return f"<Parameter {self.name}={self.value}{unit_str}>"
+
+    def get_values_for_sensitivity_analyses(self, i_max=10):
+        value = self.value
+        values = [value]
+        if self.interval is None:
+            raise ValueError("To run a sensitivity analysis, interval must be specified in the parameter configuration")
+        i = 0
+        if self.max_value is not None:
+            while i < i_max and value < self.max_value:
+                value += self.interval
+                values.append(round(value,3))
+                i += 1
+        else:
+            while i < i_max:
+                value += self.interval
+                values.append(round(value,3))                
+                i += 1
+        value = self.value
+        i = 0
+
+        if self.min_value is not None:
+            while i < i_max and value > self.min_value:
+                value -= self.interval
+                values.append(round(value,3))
+                i += 1
+        else:
+            while i < i_max:
+                value -= self.interval
+                values.append(round(value,3))                
+                i += 1
+
+        values.sort()
+        return values
+
+@dataclass    
+class ParameterSet:
+    parameters : dict = field(init=False, compare=False)    
+
+    def validate(self):
+        for param in fields(self)[1:]:
+            param_name = "param_" + param.name
+            self.parameters[param_name].validate()
+        self._validate_param_interdependency()
+
+    def _validate_param_interdependency(self):
+        pass

mobility/study_area.py

@@ -7,6 +7,7 @@
 from typing import Union, List
 
 from mobility.file_asset import FileAsset
+from mobility.model_parameters import Parameter
 from mobility.parsers.local_admin_units import LocalAdminUnits
 
 
@@ -38,11 +39,23 @@ def __init__(
             cutout_geometries: gpd.GeoDataFrame = None
         ):
 
+        radius_param = Parameter(
+            name="radius",
+            name_fr="rayon",
+            value=radius,
+            description="radius",
+            default_value=20.0,
+            parameter_type=float | int,
+            min_value=5.0,
+            max_value=100.0,
+            unit="km"
+        )
+
         inputs = {
             "version": "1",
             "local_admin_units": LocalAdminUnits(),
             "local_admin_unit_id": local_admin_unit_id,
-            "radius": radius,
+            "radius": radius_param,
             "cutout_geometries": cutout_geometries
         }
 
@@ -91,7 +104,7 @@ def create_and_get_asset(self) -> gpd.GeoDataFrame:
             local_admin_units = self.filter_within_radius(
                 local_admin_units,
                 local_admin_unit_id,
-                self.inputs["radius"]
+                self.inputs["radius"].value
             )
 
         else:

mobility/transport_zones.py

@@ -10,6 +10,7 @@
 from shapely.geometry import Point
 
 from mobility.file_asset import FileAsset
+from mobility.model_parameters import Parameter
 from mobility.study_area import StudyArea
 from mobility.parsers.osm import OSMData
 from mobility.r_utils.r_script import RScript
@@ -61,6 +62,16 @@ def __init__(
             cutout_geometries: gpd.GeoDataFrame = None
         ):
 
+        level_of_detail_param = Parameter(
+            name="level of detail",
+            name_fr="niveau de détail",
+            value=level_of_detail,
+            description="radius",
+            default_value=0,
+            possible_values = [0, 1],
+            parameter_type=bool
+        )
+
         # If the user does not choose an inner radius or a list of inner 
         # transport zones, we suppose that there is no inner / outer zones
         # (= all zones are inner zones)
@@ -83,7 +94,7 @@ def __init__(
         inputs = {
             "version": "1",
             "study_area": study_area,
-            "level_of_detail": level_of_detail,
+            "level_of_detail": level_of_detail_param,
             "osm_buildings": osm_buildings,
             "inner_radius": inner_radius,
             "inner_local_admin_unit_id": inner_local_admin_unit_id,
@@ -138,7 +149,7 @@ def create_and_get_asset(self) -> gpd.GeoDataFrame:
             args=[
                 study_area_fp,
                 osm_buildings_fp,
-                str(self.level_of_detail),
+                str(self.inputs["level_of_detail"].value),
                 self.cache_path
             ]
         )