Files
traccar_animation/screens/get_trip_from_server.py
2025-06-06 16:02:50 +03:00

348 lines
13 KiB
Python

import kivy
from kivy.app import App
from kivy.uix.screenmanager import ScreenManager, Screen
import os
import json
from kivy.clock import Clock
from kivy.properties import StringProperty, ListProperty
from utils import (
generate_key, load_key, encrypt_data, decrypt_data,
check_server_settings, save_server_settings,
test_connection, get_devices_from_server, save_route_to_file, fetch_positions_for_selected_day
)
from datetime import date
from kivy.uix.popup import Popup
from kivy.uix.gridlayout import GridLayout
from kivy.uix.button import Button
from kivy.uix.label import Label
from kivy.uix.boxlayout import BoxLayout
from threading import Thread
from kivy.clock import mainthread
from kivy.uix.image import Image
from kivy.uix.behaviors import ButtonBehavior
from kivy.uix.progressbar import ProgressBar
import webbrowser
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from PIL import Image
import time
from config import RESOURCES_FOLDER, CREDENTIALS_FILE
class GetTripFromServer(Screen):
server_info_text = StringProperty("LOADING DATA...")
server_box_color = ListProperty([0.984, 0.553, 0.078, 1])
device_mapping = {}
def on_pre_enter(self):
self.server_box_color = [0.984, 0.553, 0.078, 1]
self.server_info_text = "LOADING DATA..."
Clock.schedule_once(self.check_server_settings, 1)
def check_server_settings(self, dt):
settings = check_server_settings()
if settings:
server_url = settings["server_url"]
username = settings["username"]
password = settings["password"]
token = settings["token"]
self.server_info_text = f"CHECKING server: {server_url}"
self.server_box_color = [0.984, 0.553, 0.078, 1]
self.ids.devices_spinner.text = "Loading devices..."
Clock.schedule_once(lambda dt: self.test_connection(server_url, username, password, token), 1)
else:
self.server_info_text = "Go to settings and set Traccar Server"
self.server_box_color = [0.909, 0.031, 0.243, 1]
self.ids.devices_spinner.text = "No devices available"
def set_result_message(self, message, color=(1, 1, 1, 1)):
self.ids.result_label.text = message
self.ids.result_label.color = color
def update_devices_spinner(self, devices):
if devices:
self.device_mapping = devices
device_names = list(devices.keys())
self.ids.devices_spinner.values = device_names
self.ids.devices_spinner.text = "Select a device"
else:
self.ids.devices_spinner.text = "No devices found"
self.ids.devices_spinner.values = []
def test_connection(self, server_url, username, password, token):
result = test_connection(server_url, username, password, token)
if result["status"]:
self.server_info_text = f"Connected to {server_url}"
self.server_box_color = [0.008, 0.525, 0.290, 1]
devices = get_devices_from_server()
self.update_devices_spinner(devices)
else:
self.server_info_text = result["message"]
self.server_box_color = [0.909, 0.031, 0.243, 1]
def on_device_selected(self, device_name):
if device_name != "Loading devices..." and device_name != "No devices found":
self.ids.devices_spinner.background_color = (0.008, 0.525, 0.290, 1)
else:
self.ids.devices_spinner.background_color = (1, 1, 1, 1)
def open_date_picker(self, which):
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.gridlayout import GridLayout
from kivy.uix.button import Button
from kivy.uix.label import Label
from kivy.uix.popup import Popup
import calendar
today = date.today()
selected = {"year": today.year, "month": today.month}
def update_grid():
grid.clear_widgets()
days_in_month = calendar.monthrange(selected["year"], selected["month"])[1]
for day in range(1, days_in_month + 1):
btn = Button(
text=str(day),
size_hint=(None, None),
size=(38, 38),
background_color=(0.341, 0.235, 0.980, 1),
color=(1, 1, 1, 1),
font_size=15
)
def on_date_selected(instance, day=day):
date_str = f"{selected['year']}-{selected['month']:02d}-{day:02d}"
if which == 'start':
self.ids.start_date_picker_button.text = date_str
else:
self.ids.end_date_picker_button.text = date_str
popup.dismiss()
btn.bind(on_press=on_date_selected)
grid.add_widget(btn)
def prev_month(instance):
if selected["month"] > 1:
selected["month"] -= 1
else:
selected["year"] -= 1
selected["month"] = 12
# Don't allow future months
if (selected["year"], selected["month"]) > (today.year, today.month):
selected["year"], selected["month"] = today.year, today.month
title_label.text = f"Select a Date ({selected['year']}-{selected['month']:02d})"
update_grid()
def next_month(instance):
# Only allow up to current month
if (selected["year"], selected["month"]) < (today.year, today.month):
if selected["month"] < 12:
selected["month"] += 1
else:
selected["year"] += 1
selected["month"] = 1
title_label.text = f"Select a Date ({selected['year']}-{selected['month']:02d})"
update_grid()
def on_cancel(instance):
popup.dismiss()
# Main vertical layout
main_layout = BoxLayout(orientation='vertical', spacing=10, padding=10)
# Month navigation row
nav_layout = BoxLayout(orientation='horizontal', size_hint_y=None, height=40, spacing=10)
prev_btn = Button(text="<", size_hint_x=None, width=40, font_size=18, background_color=(0.341, 0.235, 0.980, 1), color=(1,1,1,1))
next_btn = Button(text=">", size_hint_x=None, width=40, font_size=18, background_color=(0.341, 0.235, 0.980, 1), color=(1,1,1,1))
title_label = Label(
text=f"Select a Date ({selected['year']}-{selected['month']:02d})",
size_hint_x=1,
font_size=18,
color=(1, 1, 1, 1)
)
prev_btn.bind(on_press=prev_month)
next_btn.bind(on_press=next_month)
nav_layout.add_widget(prev_btn)
nav_layout.add_widget(title_label)
nav_layout.add_widget(next_btn)
main_layout.add_widget(nav_layout)
# Grid of days: 6 columns
grid = GridLayout(cols=6, spacing=6, size_hint_y=None)
grid.bind(minimum_height=grid.setter('height'))
main_layout.add_widget(grid)
# Cancel button
cancel_btn = Button(
text="Cancel",
size_hint_y=None,
height=44,
background_color=(0.909, 0.031, 0.243, 1),
color=(1, 1, 1, 1),
font_size=16
)
cancel_btn.bind(on_press=on_cancel)
main_layout.add_widget(cancel_btn)
popup = Popup(
title="",
content=main_layout,
size_hint=(0.95, 0.8),
background_color=(0.11, 0.10, 0.15, 1),
separator_height=0,
auto_dismiss=False
)
update_grid()
popup.open()
def open_hour_picker(self, which):
from kivy.uix.popup import Popup
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.button import Button
from kivy.uix.scrollview import ScrollView
from kivy.uix.gridlayout import GridLayout
popup_layout = BoxLayout(orientation='vertical', spacing=8, padding=8)
scroll = ScrollView(size_hint=(1, 1))
grid = GridLayout(cols=1, size_hint_y=None, spacing=4)
grid.bind(minimum_height=grid.setter('height'))
# Add hour buttons
for h in range(24):
hour_str = f"{h:02d}"
btn = Button(
text=hour_str,
size_hint_y=None,
height=44,
font_size=18,
background_color=(0.341, 0.235, 0.980, 1),
color=(1, 1, 1, 1)
)
def set_hour(instance, hour=hour_str):
if which == 'start':
self.ids.start_hour_button.text = hour
else:
self.ids.end_hour_button.text = hour
popup.dismiss()
btn.bind(on_press=set_hour)
grid.add_widget(btn)
scroll.add_widget(grid)
popup_layout.add_widget(scroll)
# Cancel button
cancel_btn = Button(
text="Cancel",
size_hint_y=None,
height=44,
background_color=(0.909, 0.031, 0.243, 1),
color=(1, 1, 1, 1),
font_size=16
)
cancel_btn.bind(on_press=lambda x: popup.dismiss())
popup_layout.add_widget(cancel_btn)
popup = Popup(
title="Select Hour",
content=popup_layout,
size_hint=(0.6, 0.7),
auto_dismiss=False
)
popup.open()
def update_points_count(self, count):
"""Update the label showing the number of points."""
self.ids.points_count_label.text = f"Points: {count}"
def save_route(self):
"""Save the current list of positions as a route in resources/projects/<route_name>/positions.json."""
route_name = self.ids.route_name_input.text.strip()
positions = getattr(self, "last_positions", None)
success, message, file_path = save_route_to_file(route_name, positions)
self.ids.result_label.text = message
if success:
# Reset UI fields
self.ids.devices_spinner.text = "Select a device"
self.ids.start_date_picker_button.text = "Select Start Date"
self.ids.end_date_picker_button.text = "Select End Date"
self.ids.start_hour_button.text = "00"
self.ids.end_hour_button.text = "23"
self.ids.points_count_label.text = "Points: 0"
self.ids.route_name_input.text = ""
# Show popup and schedule reroute to home
popup = Popup(title="Success",
content=Label(text=f"Route '{route_name}' saved!"),
size_hint=(None, None), size=(300, 150),
auto_dismiss=False)
popup.open()
def close_and_go_home(dt):
popup.dismiss()
self.manager.current = "home"
Clock.schedule_once(close_and_go_home, 3)
def get_trip_server_data(self):
"""Handle the Get trip server data button press."""
selected_device = self.ids.devices_spinner.text
start_date = self.ids.start_date_picker_button.text
end_date = self.ids.end_date_picker_button.text
if selected_device == "Loading devices..." or selected_device == "No devices found":
print("No valid device selected.")
self.ids.result_label.text = "Please select a valid device."
return
if start_date == "Select Date" or end_date == "Select Date":
print("No valid date selected.")
self.ids.result_label.text = "Please select valid start and end dates."
return
# Fetch trip data from the server
print(f"Fetching trip data for device: {selected_device} from {start_date} to {end_date}")
self.ids.result_label.text = f"Fetching trip data for {selected_device} from {start_date} to {end_date}..."
positions = self.fetch_positions_for_selected_day()
self.last_positions = positions # Store for saving
self.update_points_count(len(positions) if positions else 0)
if positions:
print("Positions received:")
for pos in positions:
print(f"{pos['deviceTime']}: {pos['latitude']}, {pos['longitude']}")
else:
print("No positions found or error occurred.")
def fetch_positions_for_selected_day(self):
settings = check_server_settings()
device_name = self.ids.devices_spinner.text
start_date = self.ids.start_date_picker_button.text
end_date = self.ids.end_date_picker_button.text
start_hour = self.ids.start_hour_button.text
end_hour = self.ids.end_hour_button.text
positions, error = fetch_positions_for_selected_day(
settings,
self.device_mapping,
device_name,
start_date,
end_date,
start_hour,
end_hour
)
if error:
self.ids.result_label.text = error
return []
self.ids.result_label.text = f"Retrieved {len(positions)} positions."
return positions
def fetch_devices_async(self):
Thread(target=self._fetch_devices_worker).start()
def _fetch_devices_worker(self):
devices = get_devices_from_server()
self.update_devices_spinner_mainthread(devices)
@mainthread
def update_devices_spinner_mainthread(self, devices):
self.update_devices_spinner(devices)