Source code for trackintel.model.triplegs

import pandas as pd
import trackintel as ti

from trackintel.analysis.labelling import predict_transport_mode
from trackintel.analysis.modal_split import calculate_modal_split
from trackintel.analysis.tracking_quality import temporal_tracking_quality
from trackintel.geogr.distances import calculate_distance_matrix
from import write_triplegs_csv
from import write_triplegs_postgis
from trackintel.model.util import _copy_docstring, get_speed_triplegs
from trackintel.preprocessing.filter import spatial_filter
from trackintel.preprocessing.triplegs import generate_trips
from trackintel.visualization.triplegs import plot_triplegs

[docs]@pd.api.extensions.register_dataframe_accessor("as_triplegs") class TriplegsAccessor(object): """A pandas accessor to treat (Geo)DataFrames as collections of `Tripleg`. This will define certain methods and accessors, as well as make sure that the DataFrame adheres to some requirements. Requires at least the following columns: ['user_id', 'started_at', 'finished_at'] Requires valid line geometries; the 'index' of the GeoDataFrame will be treated as unique identifier of the `triplegs` For several usecases, the following additional columns are required: ['mode', 'trip_id'] Notes ----- A `Tripleg` (also called `stage`) is defined as continuous movement without changing the mode of transport. 'started_at' and 'finished_at' are timezone aware pandas datetime objects. Examples -------- >>> df.as_triplegs.plot() """ required_columns = ["user_id", "started_at", "finished_at"] def __init__(self, pandas_obj): self._validate(pandas_obj) self._obj = pandas_obj @staticmethod def _validate(obj): assert obj.shape[0] > 0, "Geodataframe is empty with shape: {}".format(obj.shape) # check columns if any([c not in obj.columns for c in TriplegsAccessor.required_columns]): raise AttributeError( "To process a DataFrame as a collection of triplegs, " + "it must have the properties [%s], but it has [%s]." % (", ".join(TriplegsAccessor.required_columns), ", ".join(obj.columns)) ) # check geometry assert obj.geometry.is_valid.all(), ( "Not all geometries are valid. Try x[~ x.geometry.is_valid] " "where x is you GeoDataFrame" ) if obj.geometry.iloc[0].geom_type != "LineString": raise AttributeError("The geometry must be a LineString (only first checked).") # check timestamp dtypes assert pd.api.types.is_datetime64tz_dtype( obj["started_at"] ), "dtype of started_at is {} but has to be datetime64 and timezone aware".format(obj["started_at"].dtype) assert pd.api.types.is_datetime64tz_dtype( obj["finished_at"] ), "dtype of finished_at is {} but has to be datetime64 and timezone aware".format(obj["finished_at"].dtype)
[docs] @_copy_docstring(plot_triplegs) def plot(self, *args, **kwargs): """ Plot this collection of triplegs. See :func:`trackintel.visualization.triplegs.plot_triplegs`. """ ti.visualization.triplegs.plot_triplegs(self._obj, *args, **kwargs)
[docs] @_copy_docstring(write_triplegs_csv) def to_csv(self, filename, *args, **kwargs): """ Store this collection of triplegs as a CSV file. See :func:``. """, filename, *args, **kwargs)
[docs] @_copy_docstring(write_triplegs_postgis) def to_postgis( self, name, con, schema=None, if_exists="fail", index=True, index_label=None, chunksize=None, dtype=None ): """ Store this collection of triplegs to PostGIS. See :func:``. """ self._obj, name, con, schema, if_exists, index, index_label, chunksize, dtype )
[docs] @_copy_docstring(calculate_distance_matrix) def calculate_distance_matrix(self, *args, **kwargs): """ Calculate pair-wise distance among triplegs or to other triplegs. See :func:`trackintel.geogr.distances.calculate_distance_matrix`. """ return ti.geogr.distances.calculate_distance_matrix(self._obj, *args, **kwargs)
[docs] @_copy_docstring(spatial_filter) def spatial_filter(self, *args, **kwargs): """ Filter triplegs with a geo extent. See :func:`trackintel.preprocessing.filter.spatial_filter`. """ return ti.preprocessing.filter.spatial_filter(self._obj, *args, **kwargs)
[docs] @_copy_docstring(generate_trips) def generate_trips(self, *args, **kwargs): """ Generate trips based on staypoints and triplegs. See :func:`trackintel.preprocessing.triplegs.generate_trips`. """ # if staypoints in kwargs: 'staypoints' can not be in args as it would be the first argument if "staypoints" in kwargs: return ti.preprocessing.triplegs.generate_trips(triplegs=self._obj, **kwargs) # if 'staypoints' no in kwargs it has to be the first argument in 'args' else: assert len(args) <= 1, ( "All arguments except 'staypoints' have to be given as keyword arguments. You gave" f" {args[1:]} as positional arguments." ) return ti.preprocessing.triplegs.generate_trips(staypoints=args[0], triplegs=self._obj, **kwargs)
[docs] @_copy_docstring(predict_transport_mode) def predict_transport_mode(self, *args, **kwargs): """ Predict/impute the transport mode with which each tripleg was likely covered. See :func:`trackintel.analysis.labelling.predict_transport_mode`. """ return ti.analysis.labelling.predict_transport_mode(self._obj, *args, **kwargs)
[docs] @_copy_docstring(calculate_modal_split) def calculate_modal_split(self, *args, **kwargs): """ Calculate the modal split of the triplegs. See :func:`trackintel.analysis.modal_split.calculate_modal_split`. """ return ti.analysis.modal_split.calculate_modal_split(self._obj, *args, **kwargs)
[docs] @_copy_docstring(temporal_tracking_quality) def temporal_tracking_quality(self, *args, **kwargs): """ Calculate per-user temporal tracking quality (temporal coverage). See :func:`trackintel.analysis.tracking_quality.temporal_tracking_quality`. """ return ti.analysis.tracking_quality.temporal_tracking_quality(self._obj, *args, **kwargs)
[docs] @_copy_docstring(get_speed_triplegs) def get_speed(self, *args, **kwargs): """ Compute the average speed for each tripleg, given by overall distance and duration (in m/s) See :func:`trackintel.model.util.get_speed_triplegs`. """ return ti.model.util.get_speed_triplegs(self._obj, *args, **kwargs)