fusionpbx_connect/fleet_management/traccar_api.py

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from frappe.utils.password import get_decrypted_password
import frappe
import requests


def traccar_auth(ts):
    """authenticates on traccar server and returns cookies"""
    response = requests.get(
        f"{ts.traccar_server}/api/session",
        params = {
            'token': get_decrypted_password('Traccar Settings', ts.name, 'traccar_token')
        }
    )

    if response.status_code == 200:
        return response.cookies
    else:
        frappe.throw(f"Authentication failed: {response.status_code}: {response.reason}<br />Dict ts: {ts.as_dict()}")


@frappe.whitelist()
def get_devices(doc=None):
    """Fetches all devices the user has access to and stores or updates them in Tracker Doctype."""

    # get informations for authentication on traccar server
    ts = frappe.get_doc('Traccar Settings')

    # get all devices from traccar server
    try:
        devices = requests.get(
            f"{ts.traccar_server}/api/devices",
            cookies = traccar_auth(ts)  # Assuming traccar_auth is defined elsewhere
        )
    except Exception as e:
        frappe.throw(f"Could not fetch devices from Traccar server: {str(e)}")

    # Process each device
    for dev in devices.json():
        tracker_id = str(dev['id'])
        # Check if the Tracker exists
        existing_tracker = frappe.db.exists('Tracker', {'portal_id': tracker_id})

        if existing_tracker:
            # If exists, get the document and update
            tracker_doc = frappe.get_doc('Tracker', existing_tracker)
        else:
            # If not, create a new Tracker document
            tracker_doc = frappe.get_doc({'doctype': 'Tracker', 'portal_id': tracker_id})

        # Update or set fields
        tracker_doc.portal_name = dev['name']
        tracker_doc.unique_id = dev['uniqueId']
        tracker_doc.status = dev['status']
        tracker_doc.last_update = datetime.fromisoformat(dev['lastUpdate']).strftime('%Y-%m-%d %H:%M:%S')
        tracker_doc.position_id = dev['positionId']
        tracker_doc.group_id = dev['groupId']
        tracker_doc.phone = dev.get('phone', '')  # Using .get() in case some fields might be missing
        tracker_doc.model = dev.get('model', '')
        tracker_doc.contact = dev.get('contact', '')
        tracker_doc.categorie = dev.get('category', '')
        tracker_doc.disabled = dev['disabled']
        # Save changes to the database
        tracker_doc.save()
    # Commit the transaction
    frappe.db.commit()


@frappe.whitelist()
def get_trips_for_device(device_id, start=None, end=None):
    """
    Fetches all trips for a device, stores or updates them in the Trip Doctype,
    and links them to the vehicle assigned to the device.
    """
    ts = frappe.get_doc('Traccar Settings')

    if not start:
        start = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%dT%H:%M:%SZ')
    if not end:
        end = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ')

    try:
        trips_response = requests.get(
            f"{ts.traccar_server}/api/reports/trips",
            cookies=traccar_auth(ts),
            params={'deviceId': device_id, 'from': start, 'to': end}
        )
        trips_response.raise_for_status()
    except requests.HTTPError as e:
        frappe.throw(f"HTTP error occurred: {e}")
    except Exception as e:
        frappe.throw(f"Error fetching trips: {e}")

    # Remove all data from device (to renew)
    existing_trips = frappe.get_all('Trip', filters={'device_id': device_id}, fields=['portal_id'])
    for trip in existing_trips:
        frappe.delete_doc('Trip', trip['portal_id'], force=1)

    for trip in trips_response.json():
        trip_doc = frappe.get_doc({'doctype': 'Trip'})

        trip_doc.device_id = trip['deviceId']
        trip_doc.start = datetime.fromisoformat(trip['startTime']).strftime('%Y-%m-%d %H:%M:%S')
        trip_doc.end = datetime.fromisoformat(trip['endTime']).strftime('%Y-%m-%d %H:%M:%S')
        trip_doc.distance = trip['distance']
        trip_doc.average_speed = trip['averageSpeed']
        trip_doc.max_speed = trip['maxSpeed']
        trip_doc.spent_fuel = trip['spentFuel']
        trip_doc.start_address = trip['startAddress']
        trip_doc.end_address = trip['endAddress']
        trip_doc.duration = trip['duration']
        trip_doc.driver_unique_id = trip['driverUniqueId']
        trip_doc.tracker = frappe.get_value('Tracker', {'portal_id': trip['deviceId']}, 'portal_name')
        trip_doc.fleet_vehicle = frappe.get_value('Fleet Vehicle', {'tracker': trip_doc.tracker}, 'name')

        trip_doc.save(ignore_permissions=True)

    frappe.db.commit()


@frappe.whitelist()
def getroutes(name, start=None, end=None):
    ts = frappe.get_doc('Traccar Settings')
    device_id = frappe.get_value('Trip', {'name': name}, 'device_id')
    startdate = datetime.strptime(start, "%Y-%m-%d %H:%M:%S")\
        .replace(tzinfo=ZoneInfo("Europe/Berlin"))\
        .astimezone(ZoneInfo("UTC"))\
        .strftime("%Y-%m-%dT%H:%M:%S.000+00:00")
    enddate = datetime.strptime(end, "%Y-%m-%d %H:%M:%S")\
        .replace(tzinfo=ZoneInfo("Europe/Berlin"))\
        .astimezone(ZoneInfo("UTC"))\
        .strftime("%Y-%m-%dT%H:%M:%S.000+00:00")

    # get route by filter from traccar server
    try:
        route_response = requests.get(
            f"{ts.traccar_server}/api/reports/route",
            cookies=traccar_auth(ts),
            headers={'Accept': 'application/json'},
            params={'deviceId': device_id, 'from': startdate, 'to': enddate}
        )
        route_response.raise_for_status()  # This will raise an error for HTTP error codes
    except requests.HTTPError as e:
        frappe.throw(f"HTTP error occurred: {e}")
    except Exception as e:
        frappe.throw(f"Error fetching trips: {e}")

    print(route_response)

    return route_response.json()

@frappe.whitelist()
def get_traccar_url():
    ts = frappe.get_doc('Traccar Settings')

    return(f"{ts.traccar_server}/?token=" + get_decrypted_password('Traccar Settings', ts.name, 'traccar_token'))