From 478abdda06d266f601aaad9d48cd1288b90bff04 Mon Sep 17 00:00:00 2001 From: FlxPo Date: Thu, 30 Jul 2026 12:22:36 +0200 Subject: [PATCH] Add zero-duration survey correction --- docs/source/api_reference.md | 14 + docs/source/data.md | 22 ++ docs/source/research_method.md | 15 + mobility/surveys/france/emp.py | 7 + mobility/surveys/france/entd.py | 13 + mobility/surveys/mobility_survey.py | 39 ++ mobility/surveys/zero_duration_activities.py | 361 ++++++++++++++++++ ...est_014_survey_zero_duration_activities.py | 227 +++++++++++ 8 files changed, 698 insertions(+) create mode 100644 mobility/surveys/zero_duration_activities.py create mode 100644 tests/back/unit/test_014_survey_zero_duration_activities.py diff --git a/docs/source/api_reference.md b/docs/source/api_reference.md index 473de49d..f0ebd113 100644 --- a/docs/source/api_reference.md +++ b/docs/source/api_reference.md @@ -155,6 +155,20 @@ Result metrics use represented-person weights, so a `sample_size` of 1000 does n Use this for French survey-based behaviour patterns from EMP 2018-2019. +```python +survey = mobility.EMPMobilitySurvey( + correct_zero_durations=True, +) +``` + +Main argument: + +- `correct_zero_durations`: estimate short durations for activities reported + as zero between consecutive trips. The default is `False`. + +Use the constructor argument shown above. Mobility applies the correction +while preparing and caching the survey. + Project-specific survey objects can also be passed to `PopulationGroupDayTrips`, but the parser and documentation should live in the project repository. ## Activities diff --git a/docs/source/data.md b/docs/source/data.md index 46b933ed..8f3f430a 100644 --- a/docs/source/data.md +++ b/docs/source/data.md @@ -15,6 +15,28 @@ Survey transfer is a modelling assumption. National surveys provide detailed beh Some survey codes still appear in model inputs or outputs. The [survey codes page](survey_codes.md) lists the main French codes for socio-professional categories, motives, and trip modes. +### Zero-Duration Activities + +Some survey diaries give the same time for one trip's arrival and the next +trip's departure. This produces an activity with a reported duration of zero, +even though the person probably stopped briefly. For example, an arrival at +08:30 followed by a departure at 08:30 records a zero-minute activity. + +You can ask Mobility to correct these records when it prepares the survey: + +```python +survey = mobility.EMPMobilitySurvey(correct_zero_durations=True) +``` + +The correction is disabled by default. When enabled, Mobility estimates short +activity durations from the same survey and adjusts the two surrounding trip +times. It does not add, remove, or reorder trips. The activity after the final +trip is unchanged because the diary does not record when it ends. + +Enabling this option changes the survey-derived activity-duration +distribution. Record the setting in the study assumptions and check the +resulting durations against suitable evidence. + ### ENTD 2007-2008 The national transport and travel survey, `Enquete Nationale Transports Deplacements`, was run in 2007 and 2008. INSEE describes it as a survey about household trips, use of individual and collective transport modes, and the vehicle fleet owned by households. diff --git a/docs/source/research_method.md b/docs/source/research_method.md index 82969980..0842b710 100644 --- a/docs/source/research_method.md +++ b/docs/source/research_method.md @@ -60,6 +60,21 @@ Mobility first builds daily activity-motive sequences for population groups. These sequences depend on survey behaviour and population characteristics such as socio-professional category, household car ownership, household size, and urban setting. Each sequence contains activity steps and expected activity-time needs. +If a survey enables `correct_zero_durations`, Mobility estimates plausible +short durations for activities whose preceding arrival equals their following +departure. This optional preprocessing changes only the two surrounding +clocks; it does not add, remove, or reorder trips. + +Timing resolution is inferred for each diary. A reported zero from a +one-minute diary is corrected to one minute. For diaries whose clocks all lie +on five-minute boundaries, Mobility fits an interval-censored Gamma duration +distribution. In other words, a duration reported as 10 minutes is treated as +an observation within the range that could round to 10 minutes, rather than +as an exact measurement. Motives with enough observations are fitted +separately; sparse motives use the survey-wide fit. The correction is limited +by the travel time available on the two surrounding trips so neither trip +becomes negative. See [Data Sources](data.md) for the user-facing assumption. + The model then searches for destinations and mode sequences that can make the daily plan feasible under the current costs and opportunity constraints. ## Opportunity Capacity diff --git a/mobility/surveys/france/emp.py b/mobility/surveys/france/emp.py index 46c9100a..6aa6f0e3 100644 --- a/mobility/surveys/france/emp.py +++ b/mobility/surveys/france/emp.py @@ -31,11 +31,15 @@ class EMPMobilitySurvey(MobilitySurvey): def __init__( self, parameters: MobilitySurveyParameters | None = None, + *, + correct_zero_durations: bool | None = None, ): """Initialize EMP mobility survey with optional parameter overrides. Args: parameters: Optional pre-built survey parameters model. + correct_zero_durations: Correct activities recorded with zero + duration. Disabled by default. """ parameters = self.prepare_parameters( parameters=parameters, @@ -43,6 +47,7 @@ def __init__( explicit_args={ "survey_name": "fr-EMP-2019", "country": "fr", + "correct_zero_durations": correct_zero_durations, }, owner_name="EMPMobilitySurvey", ) @@ -66,6 +71,8 @@ def create_and_get_asset(self) -> dict[str, pd.DataFrame]: self.download_survey_data(dataset_path) self.parse_survey_data(dataset_path) + if self.inputs["parameters"].correct_zero_durations: + self.correct_zero_durations() return self.get_cached_asset() diff --git a/mobility/surveys/france/entd.py b/mobility/surveys/france/entd.py index 6a74d6bc..fe32eb4a 100644 --- a/mobility/surveys/france/entd.py +++ b/mobility/surveys/france/entd.py @@ -31,11 +31,16 @@ class ENTDMobilitySurvey(MobilitySurvey): def __init__( self, parameters: MobilitySurveyParameters | None = None, + *, + correct_zero_durations: bool | None = None, ): """Initialize ENTD mobility survey with optional parameter overrides. Args: parameters: Optional pre-built survey parameters model. + correct_zero_durations: Unsupported until the ENTD parser exposes + the survey's trip clock times. Passing ``True`` raises + ``ValueError``. """ parameters = self.prepare_parameters( parameters=parameters, @@ -43,9 +48,17 @@ def __init__( explicit_args={ "survey_name": "fr-ENTD-2008", "country": "fr", + "correct_zero_durations": correct_zero_durations, }, owner_name="ENTDMobilitySurvey", ) + if parameters.correct_zero_durations: + # ENTD contains trip times, but this parser does not load and + # standardize them yet. + raise ValueError( + "ENTDMobilitySurvey cannot correct zero-duration activities " + "until its parser exposes the survey's trip clock times." + ) super().__init__(parameters=parameters) diff --git a/mobility/surveys/mobility_survey.py b/mobility/surveys/mobility_survey.py index 94b63d8b..f44c53f9 100644 --- a/mobility/surveys/mobility_survey.py +++ b/mobility/surveys/mobility_survey.py @@ -5,6 +5,9 @@ from typing import Annotated, Any from pydantic import BaseModel, ConfigDict, Field from mobility.runtime.assets.file_asset import FileAsset +from mobility.surveys.zero_duration_activities import ( + _correct_zero_duration_activities, +) class MobilitySurvey(FileAsset): @@ -81,6 +84,31 @@ def get_cached_asset(self) -> dict[str, pd.DataFrame]: """ return {k: pd.read_parquet(path) for k, path in self.cache_path.items()} + def correct_zero_durations(self) -> None: + """Correct and persist zero-duration activities in this survey. + + This runs once at survey creation, after the national parser has + written its standardized tables. The fit uses the survey's own trip + diaries and day weights; downstream consumers therefore receive the + same corrected ``short_trips`` table without owning correction logic. + """ + short_trips = pd.read_parquet(self.cache_path["short_trips"]) + days_trip = pd.read_parquet(self.cache_path["days_trip"]) + day_weights = ( + days_trip.reset_index()[["day_id", "pondki"]] + .drop_duplicates("day_id") + ) + weighted_trips = short_trips.reset_index().merge( + day_weights, + on="day_id", + how="left", + validate="many_to_one", + ) + corrected = _correct_zero_duration_activities(weighted_trips) + corrected.drop(columns="pondki", inplace=True) + corrected.set_index("day_id", inplace=True) + corrected.to_parquet(self.cache_path["short_trips"]) + class MobilitySurveyParameters(BaseModel): """Parameters used to configure a mobility survey asset.""" @@ -101,3 +129,14 @@ class MobilitySurveyParameters(BaseModel): description="ISO-like country code used to map surveys to population inputs.", ), ] + + correct_zero_durations: Annotated[ + bool, + Field( + title="Correct zero-duration activities", + description=( + "Fit plausible short activity durations from this survey and " + "correct equal consecutive arrival and departure times." + ), + ), + ] = False diff --git a/mobility/surveys/zero_duration_activities.py b/mobility/surveys/zero_duration_activities.py new file mode 100644 index 00000000..37fff39c --- /dev/null +++ b/mobility/surveys/zero_duration_activities.py @@ -0,0 +1,361 @@ +"""Correct activities recorded with zero duration in mobility surveys. + +An activity duration is the time between one trip's arrival and the next +trip's departure: + + incoming trip arrives -> activity -> outgoing trip departs + +When both clocks show 09:00, the survey reports a zero-duration activity even +though the person probably stopped briefly. This module fits plausible short +durations from the same survey and moves the two clocks apart. The activity +after the last trip is not corrected because its end is outside the diary. + +The timing resolution is inferred from each diary. If every clock lies on a +five-minute boundary, reported durations use a conservative five-minute +interval half-width. Other diaries use a one-minute half-width. +""" + +from __future__ import annotations + +import logging +import math + +import numpy as np +import pandas as pd +from scipy.optimize import minimize +from scipy.special import gammainc + + +_MINIMUM_MOTIVE_OBSERVATIONS = 30 +_MINIMUM_PROBABILITY = np.finfo(np.float64).tiny +_MISSING_MOTIVE = "__missing__" + + +def _gamma_cdf( + upper: np.ndarray | float, + shape: float, + scale: float, +) -> np.ndarray: + """Return the Gamma cumulative probability through ``upper``.""" + return gammainc(shape, np.asarray(upper, dtype=np.float64) / scale) + + +def _interval_probability( + observed: np.ndarray, + interval_half_width: np.ndarray, + shape: float, + scale: float, +) -> np.ndarray: + """Return Gamma mass around each reported duration.""" + lower = np.maximum(0.0, observed - interval_half_width) + upper = observed + interval_half_width + probability = _gamma_cdf(upper, shape, scale) - _gamma_cdf( + lower, + shape, + scale, + ) + return np.maximum(probability, _MINIMUM_PROBABILITY) + + +def _fit_gamma(observations: pd.DataFrame) -> tuple[float, float] | None: + """Fit one interval-censored Gamma distribution.""" + grouped = ( + observations.groupby( + ["observed_minutes", "interval_half_width_minutes"], + as_index=False, + sort=True, + )["weight"] + .sum() + ) + observed = grouped["observed_minutes"].to_numpy() + half_width = grouped["interval_half_width_minutes"].to_numpy() + weights = grouped["weight"].to_numpy() + weights = weights / weights.sum() + + # Use approximate unrounded durations for a stable initial estimate only. + initial_durations = np.maximum(observed, half_width * 0.5) + mean = max(float(np.average(initial_durations, weights=weights)), 0.1) + variance = max( + float( + np.average( + np.square(initial_durations - mean), + weights=weights, + ) + ), + mean, + ) + initial_shape = min(max(mean * mean / variance, 0.05), 20.0) + initial_scale = min(max(variance / mean, 0.05), 24.0 * 60.0) + + def negative_log_likelihood(log_parameters: np.ndarray) -> float: + shape, scale = np.exp(log_parameters) + probability = _interval_probability( + observed, + half_width, + float(shape), + float(scale), + ) + return float(-np.dot(weights, np.log(probability))) + + result = minimize( + negative_log_likelihood, + np.log([initial_shape, initial_scale]), + method="L-BFGS-B", + bounds=[(-7.0, 7.0), (-5.0, math.log(24.0 * 60.0))], + ) + if not result.success or not np.isfinite(result.fun): + return None + shape, scale = np.exp(result.x) + return float(shape), float(scale) + + +def _zero_interval_mean( + shape: float, + scale: float, + interval_half_width: int, +) -> float: + """Return mean duration given a reported value of zero.""" + interval_probability = float( + _gamma_cdf(interval_half_width, shape, scale) + ) + first_moment = ( + shape + * scale + * float( + gammainc( + shape + 1.0, + interval_half_width / scale, + ) + ) + ) + return first_moment / interval_probability + + +def _prepare_activity_durations( + trips: pd.DataFrame, +) -> tuple[pd.DataFrame, pd.Index] | None: + """Sort diaries and derive activity durations and timing resolution.""" + required_columns = { + "individual_id", + "daily_trip_index", + "departure_time", + "arrival_time", + "motive", + } + if not required_columns.issubset(trips.columns): + return None + + prepared = trips.copy() + original_index = prepared.index.copy() + prepared["_original_order"] = np.arange(len(prepared)) + if "day_id" in prepared.columns: + prepared["_diary_day_id"] = prepared["day_id"].to_numpy() + elif "day_id" in prepared.index.names: + prepared["_diary_day_id"] = prepared.index.get_level_values("day_id") + else: + return None + + prepared.reset_index(drop=True, inplace=True) + diary_columns = ["_diary_day_id", "individual_id"] + prepared.sort_values( + diary_columns + ["daily_trip_index"], + kind="stable", + inplace=True, + ) + diary = [prepared[column] for column in diary_columns] + + prepared["_departure"] = pd.to_numeric( + prepared["departure_time"], + errors="coerce", + ) + prepared["_arrival"] = pd.to_numeric( + prepared["arrival_time"], + errors="coerce", + ) + + def lies_on_five_minute_boundary(seconds: pd.Series) -> np.ndarray: + distance = np.mod(seconds, 300.0) + return np.isclose(distance, 0.0, atol=0.5) | np.isclose( + distance, + 300.0, + atol=0.5, + ) + + clocks_on_five_minutes = pd.Series( + lies_on_five_minute_boundary(prepared["_departure"]) + & lies_on_five_minute_boundary(prepared["_arrival"]), + index=prepared.index, + ) + prepared["_interval_half_width_minutes"] = np.where( + clocks_on_five_minutes.groupby(diary).transform("all"), + 5, + 1, + ) + prepared["_next_departure"] = prepared["_departure"].groupby(diary).shift( + -1 + ) + prepared["_observed_minutes"] = ( + (prepared["_next_departure"] - prepared["_arrival"]) + / 60.0 + ) + prepared["_zero_duration"] = prepared["_next_departure"].notna() & ( + np.isclose( + prepared["_next_departure"] - prepared["_arrival"], + 0.0, + atol=0.5, + ) + ) + return prepared, original_index + + +def _fit_five_minute_corrections( + prepared: pd.DataFrame, +) -> dict[str, int]: + """Fit corrections only for motives containing reported zeros. + + A motive with fewer than 30 observations uses the survey-wide fit. + """ + weights = ( + prepared["pondki"] + if "pondki" in prepared.columns + else pd.Series(1.0, index=prepared.index) + ) + observations = pd.DataFrame( + { + "motive": prepared["motive"] + .astype("string") + .fillna(_MISSING_MOTIVE), + "observed_minutes": prepared["_observed_minutes"], + "interval_half_width_minutes": prepared[ + "_interval_half_width_minutes" + ], + "weight": pd.to_numeric(weights, errors="coerce"), + } + ) + observations = observations.loc[ + observations["observed_minutes"].between(0.0, 24.0 * 60.0) + & observations["weight"].gt(0.0) + & observations["weight"].notna() + ] + if len(observations) < _MINIMUM_MOTIVE_OBSERVATIONS: + return {} + + pooled_fit = _fit_gamma(observations) + zero_motives = ( + prepared.loc[prepared["_zero_duration"], "motive"] + .astype("string") + .fillna(_MISSING_MOTIVE) + .unique() + ) + corrections: dict[str, int] = {} + for motive in zero_motives: + motive_observations = observations.loc[ + observations["motive"] == motive + ] + parameters = pooled_fit + if len(motive_observations) >= _MINIMUM_MOTIVE_OBSERVATIONS: + parameters = _fit_gamma(motive_observations) or pooled_fit + if parameters is None: + continue + shape, scale = parameters + corrections[str(motive)] = math.ceil( + _zero_interval_mean(shape, scale, 5) + ) + return corrections + + +def _apply_clock_corrections( + prepared: pd.DataFrame, + five_minute_corrections: dict[str, int], +) -> None: + """Move the two trip clocks around each zero-duration activity. + + For a four-minute correction around 09:00, the preferred split moves the + incoming arrival to 08:58 and the outgoing departure to 09:02. If either + trip is shorter than two minutes, more of the correction is assigned to + the other trip so travel durations cannot become negative. + """ + motives = prepared["motive"].astype("string").fillna(_MISSING_MOTIVE) + correction_minutes = np.where( + prepared["_interval_half_width_minutes"] == 1, + 1, + motives.map(five_minute_corrections).fillna(0), + ) + requested_correction = np.where( + prepared["_zero_duration"], + correction_minutes * 60.0, + 0.0, + ) + + remaining_travel_time = np.nan_to_num( + (prepared["_arrival"] - prepared["_departure"]).to_numpy(), + nan=0.0, + posinf=0.0, + neginf=0.0, + ) + remaining_travel_time = np.maximum(remaining_travel_time, 0.0) + arrival_shift = np.zeros(len(prepared)) + departure_shift = np.zeros(len(prepared)) + was_corrected = np.zeros(len(prepared), dtype=bool) + + for incoming in np.flatnonzero(requested_correction): + # Rows are sorted by diary and trip index. A requested correction + # therefore always belongs to the trip immediately after this one. + outgoing = incoming + 1 + requested = requested_correction[incoming] + feasible = min( + requested, + remaining_travel_time[incoming] + + remaining_travel_time[outgoing], + ) + incoming_share = np.clip( + feasible * 0.5, + feasible - remaining_travel_time[outgoing], + remaining_travel_time[incoming], + ) + outgoing_share = feasible - incoming_share + + arrival_shift[incoming] = incoming_share + departure_shift[outgoing] = outgoing_share + remaining_travel_time[incoming] -= incoming_share + remaining_travel_time[outgoing] -= outgoing_share + was_corrected[incoming] = feasible > 0.0 + + prepared["arrival_time"] = ( + prepared["_arrival"].to_numpy() - arrival_shift + ) + prepared["departure_time"] = ( + prepared["_departure"].to_numpy() + departure_shift + ) + prepared["_was_corrected"] = was_corrected + + +def _correct_zero_duration_activities( + trips: pd.DataFrame, +) -> pd.DataFrame: + """Return survey trips with plausible durations for reported-zero stays.""" + result = _prepare_activity_durations(trips) + if result is None: + return trips + prepared, original_index = result + if not prepared["_zero_duration"].any(): + return trips + + corrections = _fit_five_minute_corrections(prepared) + _apply_clock_corrections( + prepared, + corrections, + ) + correction_count = int(prepared["_was_corrected"].sum()) + if correction_count == 0: + return trips + + prepared.sort_values("_original_order", kind="stable", inplace=True) + output_columns = list(trips.columns) + corrected = prepared[output_columns].copy() + corrected.index = original_index + logging.info( + "Corrected %s zero-duration activities between consecutive trips", + correction_count, + ) + return corrected diff --git a/tests/back/unit/test_014_survey_zero_duration_activities.py b/tests/back/unit/test_014_survey_zero_duration_activities.py new file mode 100644 index 00000000..0662beea --- /dev/null +++ b/tests/back/unit/test_014_survey_zero_duration_activities.py @@ -0,0 +1,227 @@ +import numpy as np +import pandas as pd +import pytest +from scipy.stats import gamma + +from mobility.surveys.france import EMPMobilitySurvey, ENTDMobilitySurvey +from mobility.surveys.mobility_survey import MobilitySurveyParameters +from mobility.surveys.zero_duration_activities import ( + _correct_zero_duration_activities, + _fit_five_minute_corrections, + _interval_probability, + _prepare_activity_durations, + _zero_interval_mean, +) + + +def _survey_trips(*, one_minute_resolution: bool = False) -> pd.DataFrame: + rows = [] + gaps = [0] * 8 + [5] * 8 + [10] * 8 + [20] * 8 + [40] * 8 + for day_id, gap_minutes in enumerate(gaps, start=1): + first_departure = 8 * 3600 + ( + 60 if one_minute_resolution else 0 + ) + rows.extend( + [ + { + "day_id": day_id, + "individual_id": day_id, + "daily_trip_index": 1, + "departure_time": first_departure, + "arrival_time": 9 * 3600, + "motive": "2.20", + "pondki": 1.0, + }, + { + "day_id": day_id, + "individual_id": day_id, + "daily_trip_index": 2, + "departure_time": 9 * 3600 + gap_minutes * 60, + "arrival_time": 11 * 3600, + "motive": "1.1", + "pondki": 1.0, + }, + ] + ) + return pd.DataFrame(rows).set_index("day_id") + + +def test_interval_probability_matches_gamma_cdf_difference() -> None: + shape = 0.7 + scale = 12.0 + observed = np.asarray([0.0, 5.0, 20.0]) + half_width = np.asarray([5.0, 5.0, 5.0]) + + actual = _interval_probability( + observed, + half_width, + shape, + scale, + ) + lower = np.maximum(0.0, observed - half_width) + upper = observed + half_width + expected = gamma.cdf(upper, shape, scale=scale) - gamma.cdf( + lower, + shape, + scale=scale, + ) + + assert actual == pytest.approx(expected, rel=1e-9, abs=1e-12) + + +def test_zero_interval_mean_stays_inside_observation_interval() -> None: + assert 0.0 < _zero_interval_mean(0.6, 20.0, 1) < 1.0 + assert 0.0 < _zero_interval_mean(0.6, 20.0, 5) < 5.0 + + +def test_public_emp_flag_is_disabled_by_default_and_can_be_enabled() -> None: + assert not EMPMobilitySurvey().inputs[ + "parameters" + ].correct_zero_durations + assert EMPMobilitySurvey(correct_zero_durations=True).inputs[ + "parameters" + ].correct_zero_durations + parameters = MobilitySurveyParameters( + survey_name="fr-EMP-2019", + country="fr", + correct_zero_durations=True, + ) + assert EMPMobilitySurvey(parameters=parameters).inputs[ + "parameters" + ].correct_zero_durations + + +def test_entd_rejects_zero_duration_correction() -> None: + with pytest.raises( + ValueError, + match="until its parser exposes the survey's trip clock times", + ): + ENTDMobilitySurvey(correct_zero_durations=True) + + +def test_five_minute_correction_is_fitted_from_the_supplied_survey() -> None: + prepared, _ = _prepare_activity_durations( + _survey_trips(), + ) + + corrections = _fit_five_minute_corrections(prepared) + + assert set(corrections) == {"2.20"} + assert 1 <= corrections["2.20"] <= 5 + + +def test_public_survey_method_persists_a_known_correction( + monkeypatch, + tmp_path, +) -> None: + monkeypatch.setattr( + "mobility.surveys.zero_duration_activities." + "_fit_five_minute_corrections", + lambda _prepared: {"2.20": 4}, + ) + trips = _survey_trips().drop(columns="pondki") + days = pd.DataFrame( + { + "day_id": range(1, 41), + "pondki": np.ones(40), + } + ).set_index("day_id") + survey = EMPMobilitySurvey() + survey.cache_path["short_trips"] = tmp_path / "short_trips.parquet" + survey.cache_path["days_trip"] = tmp_path / "days_trip.parquet" + trips.to_parquet(survey.cache_path["short_trips"]) + days.to_parquet(survey.cache_path["days_trip"]) + + survey.correct_zero_durations() + + corrected = pd.read_parquet(survey.cache_path["short_trips"]) + first_trip = corrected.loc[1].iloc[0] + second_trip = corrected.loc[1].iloc[1] + + assert first_trip["arrival_time"] == 9 * 3600 - 2 * 60 + assert second_trip["departure_time"] == 9 * 3600 + 2 * 60 + + +def test_one_minute_diaries_need_no_fitted_distribution() -> None: + trips = _survey_trips(one_minute_resolution=True).iloc[:4] + + corrected = _correct_zero_duration_activities(trips) + second_trip = corrected.loc[1].iloc[1] + + assert second_trip["departure_time"] == 9 * 3600 + 30 + + +def test_clock_shifts_cannot_make_adjacent_travel_times_negative( + monkeypatch, +) -> None: + monkeypatch.setattr( + "mobility.surveys.zero_duration_activities." + "_fit_five_minute_corrections", + lambda _prepared: {"2.20": 4}, + ) + trips = pd.DataFrame( + { + "day_id": [1, 1], + "individual_id": [1, 1], + "daily_trip_index": [1, 2], + "departure_time": [9 * 3600, 9 * 3600], + "arrival_time": [9 * 3600, 9 * 3600 + 5 * 60], + "motive": ["2.20", "1.1"], + } + ) + corrected = _correct_zero_duration_activities(trips) + + assert corrected.loc[0, "arrival_time"] == 9 * 3600 + assert corrected.loc[1, "departure_time"] == 9 * 3600 + 4 * 60 + assert ( + corrected["arrival_time"] >= corrected["departure_time"] + ).all() + + +def test_consecutive_corrections_share_the_middle_trip_capacity() -> None: + trips = pd.DataFrame( + { + "day_id": [1, 1, 1], + "individual_id": [1, 1, 1], + "daily_trip_index": [1, 2, 3], + "departure_time": [ + 8 * 3600 + 50 * 60, + 9 * 3600, + 9 * 3600 + 20, + ], + "arrival_time": [ + 9 * 3600, + 9 * 3600 + 20, + 9 * 3600 + 10 * 60, + ], + "motive": ["2.20", "4.1", "1.1"], + } + ) + + corrected = _correct_zero_duration_activities(trips) + + assert ( + corrected["arrival_time"] >= corrected["departure_time"] + ).all() + assert corrected.loc[1, "departure_time"] > corrected.loc[0, "arrival_time"] + assert corrected.loc[2, "departure_time"] > corrected.loc[1, "arrival_time"] + + +def test_emp_creation_calls_enabled_correction( + monkeypatch, +) -> None: + survey = EMPMobilitySurvey(correct_zero_durations=True) + correction_calls = [] + monkeypatch.setattr(survey, "download_survey_data", lambda _path: None) + monkeypatch.setattr(survey, "parse_survey_data", lambda _path: None) + monkeypatch.setattr( + survey, + "correct_zero_durations", + lambda: correction_calls.append(True), + ) + monkeypatch.setattr(survey, "get_cached_asset", lambda: {"ready": True}) + + result = survey.create_and_get_asset() + + assert correction_calls == [True] + assert result == {"ready": True}