186 lines
7.2 KiB
Python
186 lines
7.2 KiB
Python
from odoo import models, fields, api, _
|
|
from math import radians, sin, cos, sqrt, atan2
|
|
|
|
|
|
CATEGORY_MAP = {
|
|
'IN_PASSENGER_VEHICLE': 'In Vehicle',
|
|
'IN_VEHICLE': 'In Vehicle',
|
|
'IN_ROAD_VEHICLE': 'In Vehicle',
|
|
'IN_RAIL_VEHICLE': 'Rail / Transit',
|
|
'IN_TWO_WHEELER_VEHICLE': 'Motorcycle / Scooter',
|
|
'WALKING': 'Walking',
|
|
'ON_FOOT': 'Walking',
|
|
'RUNNING': 'Running',
|
|
'ON_BICYCLE': 'Cycling',
|
|
'STILL': 'Stationary',
|
|
'UNKNOWN': 'Unknown',
|
|
}
|
|
|
|
GEOCODE_BATCH_SIZE = 20
|
|
|
|
|
|
class WtLocationLog(models.Model):
|
|
_name = 'wt.location.log'
|
|
_description = 'WorkTrace Location Log'
|
|
_order = 'arrived_at asc'
|
|
|
|
name = fields.Char(string='Name', compute='_compute_name', store=True)
|
|
date = fields.Date(string='Date', required=True, index=True)
|
|
arrived_at = fields.Datetime(string='Begin Time', required=True)
|
|
departed_at = fields.Datetime(string='End Time')
|
|
|
|
latitude = fields.Float(string='Latitude', digits=(10, 7))
|
|
longitude = fields.Float(string='Longitude', digits=(10, 7))
|
|
address = fields.Char(string='Address')
|
|
place_name = fields.Char(string='Place / Business Name')
|
|
category = fields.Char(string='Category')
|
|
|
|
time_at_location = fields.Float(
|
|
string='Duration (hrs)',
|
|
compute='_compute_time_at_location',
|
|
store=True
|
|
)
|
|
distance_from_previous = fields.Float(string='Distance (mi)', digits=(10, 2))
|
|
travel_time_from_previous = fields.Float(string='Travel Time (hrs)', digits=(10, 2))
|
|
travel_mode = fields.Selection([
|
|
('driving', 'Driving'),
|
|
('walking', 'Walking'),
|
|
('cycling', 'Cycling'),
|
|
('transit', 'Transit'),
|
|
('flying', 'Flying'),
|
|
('unknown', 'Unknown'),
|
|
], string='Travel Mode', default='unknown')
|
|
|
|
source = fields.Selection([
|
|
('google_timeline', 'Google Timeline'),
|
|
('manual', 'Manual'),
|
|
], string='Source', default='google_timeline')
|
|
|
|
calendar_event_id = fields.Many2one('calendar.event', string='Calendar Event', ondelete='set null')
|
|
add_to_calendar = fields.Boolean(string='Add to Calendar', default=True)
|
|
notes = fields.Text(string='Notes')
|
|
|
|
@api.depends('place_name', 'address', 'latitude', 'longitude')
|
|
def _compute_name(self):
|
|
for rec in self:
|
|
rec.name = (
|
|
rec.place_name
|
|
or rec.address
|
|
or (f'{rec.latitude:.5f}, {rec.longitude:.5f}' if rec.latitude else 'Unknown Location')
|
|
)
|
|
|
|
@api.depends('arrived_at', 'departed_at')
|
|
def _compute_time_at_location(self):
|
|
for rec in self:
|
|
if rec.arrived_at and rec.departed_at:
|
|
rec.time_at_location = (rec.departed_at - rec.arrived_at).total_seconds() / 3600
|
|
else:
|
|
rec.time_at_location = 0.0
|
|
|
|
@staticmethod
|
|
def _haversine_distance(lat1, lon1, lat2, lon2):
|
|
R = 3958.8
|
|
lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
|
|
dlat, dlon = lat2 - lat1, lon2 - lon1
|
|
a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
|
|
return R * 2 * atan2(sqrt(a), sqrt(1 - a))
|
|
|
|
def action_geocode(self):
|
|
"""Resolve coordinates to address using OpenStreetMap Nominatim.
|
|
Processes up to GEOCODE_BATCH_SIZE records per call to avoid HTTP worker timeouts.
|
|
"""
|
|
import requests
|
|
import time
|
|
|
|
to_geocode = self.filtered(lambda r: r.latitude and r.longitude and not r.address)
|
|
if not to_geocode:
|
|
to_geocode = self.filtered(lambda r: r.latitude and r.longitude)
|
|
|
|
batch = to_geocode[:GEOCODE_BATCH_SIZE]
|
|
processed = 0
|
|
|
|
for rec in batch:
|
|
url = 'https://nominatim.openstreetmap.org/reverse'
|
|
params = {'lat': rec.latitude, 'lon': rec.longitude, 'format': 'json', 'addressdetails': 1}
|
|
headers = {'User-Agent': 'WorkTrace/1.0 (Odoo Module)'}
|
|
try:
|
|
resp = requests.get(url, params=params, headers=headers, timeout=10)
|
|
resp.raise_for_status()
|
|
result = resp.json()
|
|
addr = result.get('address', {})
|
|
|
|
rec.address = result.get('display_name', '')
|
|
rec.place_name = rec.place_name or (
|
|
result.get('name')
|
|
or addr.get('amenity')
|
|
or addr.get('shop')
|
|
or addr.get('building')
|
|
or addr.get('office')
|
|
or addr.get('tourism')
|
|
or ''
|
|
)
|
|
osm_type = (
|
|
addr.get('amenity') or addr.get('shop') or addr.get('tourism')
|
|
or addr.get('leisure') or addr.get('office') or addr.get('building')
|
|
or result.get('type', '') or result.get('class', '')
|
|
)
|
|
if osm_type and not rec.category:
|
|
rec.category = osm_type.replace('_', ' ').title()
|
|
|
|
processed += 1
|
|
time.sleep(1)
|
|
except Exception:
|
|
pass
|
|
|
|
remaining = self.search_count([('latitude', '!=', 0), ('address', '=', False)])
|
|
|
|
if remaining:
|
|
msg = _('%d address(es) resolved. %d still need geocoding — click "Geocode Next %d" again to continue.') % (
|
|
processed, remaining, GEOCODE_BATCH_SIZE)
|
|
notif_type = 'warning'
|
|
else:
|
|
msg = _('All done! %d address(es) resolved.') % processed
|
|
notif_type = 'success'
|
|
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('Geocoding Batch Complete'),
|
|
'message': msg,
|
|
'type': notif_type,
|
|
'sticky': True,
|
|
},
|
|
}
|
|
|
|
def action_geocode_all_pending(self):
|
|
"""Geocode the next batch of ALL unresolved records (ignores selection)."""
|
|
pending = self.search([('latitude', '!=', 0), ('address', '=', False)], limit=GEOCODE_BATCH_SIZE)
|
|
if not pending:
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _('Nothing to Geocode'),
|
|
'message': _('All records with coordinates already have addresses.'),
|
|
'type': 'info',
|
|
'sticky': False,
|
|
},
|
|
}
|
|
return pending.action_geocode()
|
|
|
|
def action_compute_distances(self):
|
|
"""Recompute distances and travel times for all logs grouped by date."""
|
|
dates = self.mapped('date')
|
|
for date in set(dates):
|
|
logs = self.search([('date', '=', date)], order='arrived_at asc')
|
|
prev = None
|
|
for log in logs:
|
|
if prev and prev.latitude and log.latitude:
|
|
log.distance_from_previous = self._haversine_distance(
|
|
prev.latitude, prev.longitude, log.latitude, log.longitude
|
|
)
|
|
if prev.departed_at and log.arrived_at:
|
|
delta = log.arrived_at - prev.departed_at
|
|
log.travel_time_from_previous = max(delta.total_seconds() / 3600, 0)
|
|
prev = log |