fix: extract travel_mode from activity segments preceding each visit in semanticSegments
Some checks failed
pre-commit / pre-commit (push) Has been cancelled
tests / Detect unreleased dependencies (push) Has been cancelled
tests / test with OCB (push) Has been cancelled
tests / test with Odoo (push) Has been cancelled

This commit is contained in:
2026-03-14 02:59:53 +00:00
parent aa714a8fd6
commit 7c65a34fce

View File

@@ -8,8 +8,12 @@ from datetime import datetime, timedelta
from math import radians, sin, cos, sqrt, atan2
from collections import Counter
VEHICLE_ACTIVITIES = {'IN_VEHICLE', 'IN_ROAD_VEHICLE', 'IN_RAIL_VEHICLE', 'IN_TWO_WHEELER_VEHICLE'}
WALKING_ACTIVITIES = {'WALKING', 'ON_FOOT', 'RUNNING', 'ON_BICYCLE'}
VEHICLE_ACTIVITIES = {
'IN_VEHICLE', 'IN_ROAD_VEHICLE', 'IN_RAIL_VEHICLE',
'IN_TWO_WHEELER_VEHICLE', 'IN_PASSENGER_VEHICLE',
}
WALKING_ACTIVITIES = {'WALKING', 'ON_FOOT', 'RUNNING'}
CYCLING_ACTIVITIES = {'ON_BICYCLE', 'CYCLING'}
SEMANTIC_TYPE_CATEGORY = {
'HOME': 'Home',
@@ -33,11 +37,19 @@ def _distance_meters(lat1, lon1, lat2, lon2):
return _haversine_miles(lat1, lon1, lat2, lon2) * 1609.34
def _get_travel_mode(activity_type):
if activity_type in VEHICLE_ACTIVITIES:
def _activity_type_to_travel_mode(activity_type):
"""Map Google activity type string to Odoo travel_mode selection value."""
if not activity_type:
return 'unknown'
t = activity_type.upper()
if t in VEHICLE_ACTIVITIES:
return 'driving'
if activity_type in WALKING_ACTIVITIES:
if t == 'IN_RAIL_VEHICLE':
return 'transit'
if t in WALKING_ACTIVITIES:
return 'walking'
if t in CYCLING_ACTIVITIES:
return 'cycling'
return 'unknown'
@@ -147,7 +159,7 @@ class WtImportTimelineWizard(models.TransientModel):
stop['distance_from_previous'] = 0.0
stop['travel_time_from_previous'] = 0.0
# Skip existing records
# Skip existing records (deduplicate by arrived_at)
LocationLog = self.env['wt.location.log']
existing = set(
r.arrived_at.strftime('%Y-%m-%d %H:%M:%S')
@@ -195,22 +207,47 @@ class WtImportTimelineWizard(models.TransientModel):
}
def _parse_semantic_timeline(self, data):
"""Parse Timeline.json (semanticSegments format from Android export).
Iterates segments in order. Activity segments (travel between stops)
are captured to provide travel_mode for the following visit segment.
"""
stops = []
for seg in data.get('semanticSegments', []):
segments = data.get('semanticSegments', [])
pending_travel_mode = 'unknown'
for seg in segments:
# --- Activity / travel segment ---
activity = seg.get('activity')
if activity:
top = activity.get('topCandidate', {})
activity_type = top.get('type', '')
mode = _activity_type_to_travel_mode(activity_type)
if mode != 'unknown':
pending_travel_mode = mode
continue
# --- Visit / stop segment ---
visit = seg.get('visit')
if not visit:
# timelinePath or unknown — ignore but keep pending travel mode
continue
start_ts = _parse_ts(seg.get('startTime'))
end_ts = _parse_ts(seg.get('endTime'))
if not start_ts or not end_ts:
pending_travel_mode = 'unknown'
continue
candidate = visit.get('topCandidate', {})
semantic_type = candidate.get('semanticType', '')
latlng_str = candidate.get('placeLocation', {}).get('latLng', '')
lat, lng = _parse_latlng(latlng_str) if latlng_str else (None, None)
category = SEMANTIC_TYPE_CATEGORY.get(semantic_type, '')
if not category and semantic_type:
category = semantic_type.replace('_', ' ').title()
stops.append({
'arrived_at': start_ts,
'departed_at': end_ts,
@@ -218,11 +255,16 @@ class WtImportTimelineWizard(models.TransientModel):
'lng': lng,
'place_name': '',
'category': category,
'travel_mode': 'unknown',
'travel_mode': pending_travel_mode,
})
# Reset after consuming for the next visit
pending_travel_mode = 'unknown'
return stops
def _parse_raw_timeline(self, data, proximity_meters=200):
"""Parse Timeline Edits.json (raw GPS signal format from Google Takeout)."""
positions = []
activities = []
for entry in data.get('timelineEdits', []):
@@ -256,7 +298,7 @@ class WtImportTimelineWizard(models.TransientModel):
if not window:
return 'unknown'
counts = Counter(a['type'] for a in window)
return _get_travel_mode(counts.most_common(1)[0][0])
return _activity_type_to_travel_mode(counts.most_common(1)[0][0])
stops = []
current_cluster = [positions[0]]