Repository: mie-lab/trackintel (language: python)

Issue:
Data quality assessments
The data quality assessments by time (tracking quality per hour/day/week per user) are already implemented and need to be integrated into trackintel. The current idea is to put it into the `analysis` module and output a quality df. The user can then decide whether to plot or save it.

Question: how do we measure data quality spatially? 


Discussion / hints:
Hm... good question. I think spatial data quality is very hard to evaluate maybe we should focus on temporal completeness first.
In the past we sometimes analyzed the "accuracy" attribute of GPS trackpoints. For trackpoints we could maybe calculate the  average distance between two consecutive trackpoints?

Fix patch:
diff --git a/docs/modules/analysis.rst b/docs/modules/analysis.rst
index 5b4afb33..87d1a378 100644
--- a/docs/modules/analysis.rst
+++ b/docs/modules/analysis.rst
@@ -4,6 +4,11 @@ Analysis
 The analysis module provides a variety of functions to analyze the mobility
 datasets computed by trackintel. 
 
+Tracking Quality
+================
+
+.. autofunction:: trackintel.analysis.tracking_quality.temporal_tracking_quality
+
 Triplegs
 ========
 
diff --git a/trackintel/analysis/__init__.py b/trackintel/analysis/__init__.py
index 849b37fb..71d18416 100644
--- a/trackintel/analysis/__init__.py
+++ b/trackintel/analysis/__init__.py
@@ -1,1 +1,2 @@
 from .transport_mode_identification import predict_transport_mode
+from .tracking_quality import temporal_tracking_quality
\ No newline at end of file
diff --git a/trackintel/analysis/tracking_quality.py b/trackintel/analysis/tracking_quality.py
new file mode 100644
index 00000000..6b7ec9b7
--- /dev/null
+++ b/trackintel/analysis/tracking_quality.py
@@ -0,0 +1,169 @@
+import datetime
+
+import pandas as pd
+
+
+def temporal_tracking_quality(source, granularity="all"):
+    """
+    Calculate per-user temporal tracking quality.
+
+    Parameters
+    ----------
+    df : GeoDataFrame (as trackintel datamodels)
+        The source dataframe to calculate temporal tracking quality.
+
+    granularity : {"all", "day", "hour"}, default="all"
+        The level of which the tracking quality is claculated. The default "all" returns
+        the overall tracking quality, and "day" and "hour" returns the tracking quality
+        by days and hours, respectively.
+
+    Returns
+    -------
+    quality: DataFrame
+        A per-user temporal tracking quality dataframe.
+
+    Note
+    ----
+    The temporal tracking quality is the time proportion of tracked period with the possible
+    time extent. The possible time extents of the different granularities are different: "all"
+    considers the time between the latest "finished_at" and the earliest "started_at", whereas
+    "day" considers the whole day (86400 sec) and "hour" considers the whole hour (3600 sec).
+
+    Examples
+    --------
+    >>> # calculate overall tracking quality of stps
+    >>> temporal_tracking_quality(stps, granularity="all")
+    >>> # calculate per-day tracking quality of stps and tpls sequence
+    >>> temporal_tracking_quality(spts_tpls, granularity="day")
+    """
+    df = source.copy()
+    df.reset_index(inplace=True)
+    if granularity == "all":
+        quality = df.groupby("user_id", as_index=False).apply(_get_tracking_quality_user, granularity)
+    elif granularity == "day":
+        # split records that span several days
+        df = _split_overlaps(df, granularity=granularity)
+        # get the tracked day relative to the first day
+        start_date = df["started_at"].min().date()
+        df["day"] = df["started_at"].apply(lambda x: (x.date() - start_date).days)
+        # calculate per-user per-day tracking quality
+        quality = df.groupby(["user_id", "day"], as_index=False).apply(_get_tracking_quality_user, granularity)
+    elif granularity == "hour":
+        # split records that span several hours
+        df = _split_overlaps(df, granularity="hour")
+        # get the tracked day relative to the first day
+        start_date = df["started_at"].min().date()
+        df["day"] = df["started_at"].apply(lambda x: (x.date() - start_date).days)
+        # get the hour
+        df["hour"] = df["started_at"].dt.hour
+        # calculate per-user per-hour tracking quality
+        quality = df.groupby(["user_id", "hour"], as_index=False).apply(_get_tracking_quality_user, granularity)
+    else:
+        raise AttributeError(f"granularity unknown. We only support ['all', 'day', 'hour']. You passed {granularity}")
+    return quality
+
+
+def _get_tracking_quality_user(df, granularity="all"):
+    """
+    Tracking quality per-user per-granularity.
+
+    Parameters
+    ----------
+    df : GeoDataFrame (as trackintel datamodels)
+        The source dataframe
+
+    granularity : {"all", "day", "hour"}, default "all"
+        Determines the extent of the tracking. "all" the entire tracking period,
+        "day" a whole day and "hour" a whole hour.
+
+    Returns
+    -------
+    pandas.Series
+        A pandas.Series object containing the tracking quality
+    """
+    tracked_duration = (df["finished_at"] - df["started_at"]).dt.total_seconds().sum()
+    if granularity == "all":
+        # the whole tracking period
+        extent = (df["finished_at"].max() - df["started_at"].min()).total_seconds()
+    elif granularity == "day":
+        # total seconds in a day
+        extent = 60 * 60 * 24
+    elif granularity == "hour":
+        # total seconds in an hour * number of different days
+        # (entries from multiple days may be grouped together)
+        extent = (60 * 60) * len(df["day"].unique())
+    return pd.Series([tracked_duration / extent], index=["quality"])
+
+
+def _split_overlaps(source, granularity="day"):
+    """
+    Split input df that have a duration of several days or hours.
+
+    Parameters
+    ----------
+    source : GeoDataFrame (as trackintel datamodels)
+        The source to perform the split
+
+    granularity : {'day', 'hour'}, default 'day'
+        The criteria of spliting. "day" splits records that have duration of several
+        days and "hour" splits records that have duration of several hours.
+
+    Returns
+    -------
+    GeoDataFrame (as trackintel datamodels)
+        The GeoDataFrame object after the spliting
+    """
+    df = source.copy()
+    change_flag = __get_split_index(df, granularity=granularity)
+
+    # Iteratively split one day/hour from multi day/hour entries until no entry spans over multiple days/hours
+    while change_flag.sum() > 0:
+
+        # calculate new finished_at timestamp (1sec before midnight)
+        finished_at_temp = df.loc[change_flag, "finished_at"].copy()
+        if granularity == "day":
+            df.loc[change_flag, "finished_at"] = df.loc[change_flag, "started_at"].apply(
+                lambda x: x.replace(hour=23, minute=59, second=59)
+            )
+        elif granularity == "hour":
+            df.loc[change_flag, "finished_at"] = df.loc[change_flag, "started_at"].apply(
+                lambda x: x.replace(minute=59, second=59)
+            )
+
+        # create new entries with remaining timestamp
+        new_df = df.loc[change_flag].copy()
+        new_df.loc[change_flag, "started_at"] = df.loc[change_flag, "finished_at"] + datetime.timedelta(seconds=1)
+        new_df.loc[change_flag, "finished_at"] = finished_at_temp
+
+        df = df.append(new_df, ignore_index=True, sort=True)
+
+        change_flag = __get_split_index(df, granularity=granularity)
+
+    return df
+
+
+def __get_split_index(df, granularity="day"):
+    """
+    Get the index that needs to be splitted.
+
+    Parameters
+    ----------
+    df : GeoDataFrame (as trackintel datamodels)
+        The source to perform the split.
+
+    granularity : {'day', 'hour'}, default 'day'
+        The criteria of spliting. "day" splits records that have duration of several
+        days and "hour" splits records that have duration of several hours.
+
+    Returns
+    -------
+    change_flag: pd.Series
+        Boolean index indicating which records needs to be splitted
+    """
+    change_flag = df["started_at"].dt.date != df["finished_at"].dt.date
+    if granularity == "hour":
+        hour_flag = df["started_at"].dt.hour != df["finished_at"].dt.hour
+        # union of day and hour change flag
+        change_flag = change_flag | hour_flag
+
+    return change_flag
\ No newline at end of file


Test patch:
diff --git a/tests/analysis/test_tracking_quality.py b/tests/analysis/test_tracking_quality.py
new file mode 100644
index 00000000..aa384675
--- /dev/null
+++ b/tests/analysis/test_tracking_quality.py
@@ -0,0 +1,142 @@
+import os
+import pytest
+
+import trackintel as ti
+
+
+@pytest.fixture
+def testdata_stps_tpls_geolife_long():
+    """Generate stps and tpls sequences of the original pfs for subsequent testing."""
+    pfs = ti.io.dataset_reader.read_geolife(os.path.join("tests", "data", "geolife_long"))
+    pfs, stps = pfs.as_positionfixes.generate_staypoints(method="sliding", dist_threshold=25, time_threshold=5 * 60)
+    pfs, tpls = pfs.as_positionfixes.generate_triplegs(stps, method="between_staypoints")
+
+    tpls["type"] = "tripleg"
+    stps["type"] = "staypoint"
+    stps_tpls = stps.append(tpls, ignore_index=True).sort_values(by="started_at")
+    return stps_tpls
+
+
+class TestTemporal_tracking_quality:
+    """Tests for the temporal_tracking_quality() function and its subfunctions."""
+
+    def test_tracking_quality_all(self, testdata_stps_tpls_geolife_long):
+        """Test if the calculated total tracking quality is correct."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+
+        # calculate tracking quality for a sample user
+        user_0 = stps_tpls.loc[stps_tpls["user_id"] == 0]
+        extent = (user_0["finished_at"].max() - user_0["started_at"].min()).total_seconds()
+        tracked = (user_0["finished_at"] - user_0["started_at"]).dt.total_seconds().sum()
+        quality_manual = tracked / extent
+
+        # test if the result of the user agrees
+        quality = ti.analysis.tracking_quality.temporal_tracking_quality(stps_tpls, granularity="all")
+
+        assert quality_manual == quality.loc[quality["user_id"] == 0, "quality"].values[0]
+
+    def test_tracking_quality_day(self, testdata_stps_tpls_geolife_long):
+        """Test if the calculated tracking quality per day is correct."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+        splitted_records = ti.analysis.tracking_quality._split_overlaps(stps_tpls, granularity="day")
+
+        # get the day relative to the start day
+        start_date = splitted_records["started_at"].min().date()
+        splitted_records["day"] = splitted_records["started_at"].apply(lambda x: (x.date() - start_date).days)
+        # calculate tracking quality of the first day for the first user
+        user_0 = splitted_records.loc[(splitted_records["user_id"] == 0) & (splitted_records["day"] == 0)]
+        extent = 60 * 60 * 24
+        tracked = (user_0["finished_at"] - user_0["started_at"]).dt.total_seconds().sum()
+        quality_manual = tracked / extent
+        # test if the result of the user agrees
+        quality = ti.analysis.tracking_quality.temporal_tracking_quality(stps_tpls, granularity="day")
+
+        assert quality_manual == quality.loc[(quality["user_id"] == 0) & (quality["day"] == 0), "quality"].values[0]
+
+    def test_tracking_quality_hour(self, testdata_stps_tpls_geolife_long):
+        """Test if the calculated tracking quality per hour is correct."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+
+        splitted = ti.analysis.tracking_quality._split_overlaps(stps_tpls, granularity="hour")
+
+        # get the day relative to the start day
+        start_date = splitted["started_at"].min().date()
+        splitted["day"] = splitted["started_at"].apply(lambda x: (x.date() - start_date).days)
+        # get the hour of the record
+        splitted["hour"] = splitted["started_at"].dt.hour
+
+        # calculate tracking quality of an hour for the first user
+        user_0 = splitted.loc[(splitted["user_id"] == 0) & (splitted["hour"] == 2)]
+        extent = (60 * 60) * len(user_0["day"].unique())
+        tracked = (user_0["finished_at"] - user_0["started_at"]).dt.total_seconds().sum()
+        quality_manual = tracked / extent
+
+        # test if the result of the user agrees
+        quality = ti.analysis.tracking_quality.temporal_tracking_quality(stps_tpls, granularity="hour")
+
+        assert quality_manual == quality.loc[(quality["user_id"] == 0) & (quality["hour"] == 2), "quality"].values[0]
+
+    def test_tracking_quality_error(self, testdata_stps_tpls_geolife_long):
+        """Test if the an error is raised when passing unknown 'granularity'."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+
+        with pytest.raises(AttributeError):
+            ti.analysis.tracking_quality.temporal_tracking_quality(stps_tpls, granularity=12345)
+        with pytest.raises(AttributeError):
+            ti.analysis.tracking_quality.temporal_tracking_quality(stps_tpls, granularity="random")
+
+    def test_split_overlaps_days(self, testdata_stps_tpls_geolife_long):
+        """Test if _split_overlaps() function can split records that span several days."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+
+        # some of the records span several day
+        multi_day_records = stps_tpls["finished_at"].dt.day - stps_tpls["started_at"].dt.day
+        assert (multi_day_records > 0).any()
+
+        # split the records according to day
+        stps_tpls.reset_index(inplace=True)
+        splitted = ti.analysis.tracking_quality._split_overlaps(stps_tpls, granularity="day")
+
+        # no record spans several days after the split
+        multi_day_records = splitted["finished_at"].dt.day - splitted["started_at"].dt.day
+        assert (multi_day_records == 0).all()
+
+    def test_split_overlaps_hours(self, testdata_stps_tpls_geolife_long):
+        """Test if _split_overlaps() function can split records that span several hours."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+
+        # some of the records span several hours
+        hour_diff = stps_tpls["finished_at"].dt.hour - stps_tpls["started_at"].dt.hour
+        assert (hour_diff > 0).any()
+
+        # split the records according to hour
+        stps_tpls.reset_index(inplace=True)
+        splitted = ti.analysis.tracking_quality._split_overlaps(stps_tpls, granularity="hour")
+
+        # no record spans several hours after the split
+        hour_diff = splitted["finished_at"].dt.hour - splitted["started_at"].dt.hour
+        assert (hour_diff == 0).all()
+
+    def test_split_overlaps_hours_case2(self, testdata_stps_tpls_geolife_long):
+        """Test if _split_overlaps() function can split record that have the same hour but different days."""
+        stps_tpls = testdata_stps_tpls_geolife_long
+
+        # get the first two records
+        head2 = stps_tpls.head(2).copy()
+        # construct the finished_at exactly one day after started_at
+        head2["finished_at"] = head2.apply(lambda x: x["started_at"].replace(day=x["started_at"].day + 1), axis=1)
+
+        # the records have the same hour
+        hour_diff = head2["finished_at"].dt.hour - head2["started_at"].dt.hour
+        assert (hour_diff == 0).all()
+        # but have different days
+        day_diff = head2["finished_at"].dt.day - head2["started_at"].dt.day
+        assert (day_diff > 0).all()
+
+        # split the records according to hour
+        head2.reset_index(inplace=True)
+        splitted = ti.analysis.tracking_quality._split_overlaps(head2, granularity="hour")
+
+        # no record has different days after the split
+        day_diff = splitted["finished_at"].dt.day - splitted["started_at"].dt.day
+        assert (day_diff == 0).all()
\ No newline at end of file