Files
traccar_animation/main.py
2025-06-05 16:29:51 +03:00

671 lines
26 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
kivy.require("2.0.0")
from kivy.core.window import Window
Window.size = (360, 780)
RESOURCES_FOLDER = "resources"
CREDENTIALS_FILE = os.path.join(RESOURCES_FOLDER, "credentials.enc")
class LoginScreen(Screen):
def login(self):
username = self.ids.username_input.text.strip()
password = self.ids.username_input.text.strip()
if not username or not password:
self.manager.get_screen("home").ids.result_label.text = "Please fill in all fields."
return
credentials = {"username": username, "password": password}
encrypted_data = encrypt_data(json.dumps(credentials))
with open(CREDENTIALS_FILE, "wb") as file:
file.write(encrypted_data)
self.manager.current = "home"
class SettingsScreen(Screen):
server_response = "Waiting to test connection..."
def on_pre_enter(self):
settings = check_server_settings()
if settings:
self.ids.server_url_input.text = settings.get("server_url", "")
self.ids.username_input.text = settings.get("username", "")
self.ids.password_input.text = settings.get("password", "")
self.ids.token_input.text = settings.get("token", "")
else:
self.ids.server_url_input.text = ""
self.ids.username_input.text = ""
self.ids.password_input.text = ""
self.ids.token_input.text = ""
def test_connection(self):
server_url = self.ids.server_url_input.text.strip()
username = self.ids.username_input.text.strip()
password = self.ids.password_input.text.strip()
token = self.ids.token_input.text.strip()
result = test_connection(server_url, username, password, token)
self.server_response = result["message"]
self.ids.result_label.text = self.server_response
def save_settings(self):
server_url = self.ids.server_url_input.text.strip()
username = self.ids.username_input.text.strip()
password = self.ids.password_input.text.strip()
token = self.ids.token_input.text.strip()
if not server_url or not username or not password or not token:
self.ids.result_label.text = "Please fill in all fields."
return
settings_data = {
"server_url": server_url,
"username": username,
"password": password,
"token": token,
}
try:
save_server_settings(settings_data)
self.ids.result_label.text = "Settings saved successfully!"
self.manager.current = "home"
except Exception as e:
self.ids.result_label.text = f"Failed to save settings: {str(e)}"
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.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.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)
# trhis screen is used to create a new user
# register screen
class RegisterScreen(Screen):
def create_user(self):
"""Handle user creation."""
username = self.ids.set_username_input.text.strip()
password = self.ids.set_password_input.text.strip()
confirm_password = self.ids.confirm_password_input.text.strip()
email = self.ids.set_email_input.text.strip()
if not username or not password or not confirm_password or not email:
self.ids.result_label.text = "Please fill in all fields."
return
if password != confirm_password:
self.ids.result_label.text = "Passwords do not match."
return
# Check if the username or email already exists
if self.user_exists(username, email):
self.ids.result_label.text = "User or email already exists."
return
# Save user data (encrypted)
user_data = {
"username": username,
"password": password,
"email": email,
}
encrypted_data = encrypt_data(json.dumps(user_data))
try:
with open(CREDENTIALS_FILE, "ab") as file: # Append encrypted data
file.write(encrypted_data + b"\n") # Add a newline for separation
except Exception as e:
self.ids.result_label.text = f"Failed to save user: {str(e)}"
return
self.ids.result_label.text = "User created successfully!"
# Navigate back to the login screen
self.manager.current = "login"
def user_exists(self, username, email):
"""Check if a username or email already exists in the credentials.enc file."""
try:
with open(CREDENTIALS_FILE, "rb") as file:
for line in file:
decrypted_data = decrypt_data(line.strip())
user = json.loads(decrypted_data)
if user["username"] == username or user["email"] == email:
return True
except FileNotFoundError:
pass
except Exception as e:
self.ids.result_label.text = f"Error checking user: {str(e)}"
return False
class CreateAnimationScreen(Screen):
project_name = StringProperty("")
def on_pre_enter(self):
# This will be called when the screen is shown
pass
def open_rename_popup(self):
from kivy.uix.popup import Popup
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.button import Button
from kivy.uix.textinput import TextInput
from kivy.uix.label import Label
layout = BoxLayout(orientation='vertical', spacing=10, padding=10)
label = Label(text="Enter new project name:")
input_field = TextInput(text=self.project_name, multiline=False)
btn_save = Button(text="Save", background_color=(0.008, 0.525, 0.290, 1))
btn_cancel = Button(text="Cancel")
layout.add_widget(label)
layout.add_widget(input_field)
layout.add_widget(btn_save)
layout.add_widget(btn_cancel)
popup = Popup(title="Rename Project", content=layout, size_hint=(None, None), size=(350, 260), auto_dismiss=False)
def do_rename(instance):
new_name = input_field.text.strip()
if new_name and new_name != self.project_name:
if self.rename_project_folder(self.project_name, new_name):
self.project_name = new_name
popup.dismiss()
btn_save.bind(on_press=do_rename)
btn_cancel.bind(on_press=lambda x: popup.dismiss())
popup.open()
def rename_project_folder(self, old_name, new_name):
import os
old_path = os.path.join(RESOURCES_FOLDER, "projects", old_name)
new_path = os.path.join(RESOURCES_FOLDER, "projects", new_name)
if os.path.exists(old_path) and not os.path.exists(new_path):
os.rename(old_path, new_path)
return True
return False
class HomeScreen(Screen):
def on_pre_enter(self):
"""Load existing projects/trips when the screen is entered."""
self.load_existing_projects()
def load_existing_projects(self):
projects_folder = os.path.join(RESOURCES_FOLDER, "projects")
archive_folder = os.path.join(RESOURCES_FOLDER, "trip_archive")
if not os.path.exists(projects_folder):
os.makedirs(projects_folder)
if not os.path.exists(archive_folder):
os.makedirs(archive_folder)
self.ids.projects_list.clear_widgets()
for project in os.listdir(projects_folder):
row = BoxLayout(
orientation="horizontal",
size_hint_y=None,
height=44,
spacing=6,
padding=(8, 6)
)
from kivy.graphics import Color, Line, RoundedRectangle
with row.canvas.before:
Color(0.11, 0.10, 0.15, 1)
row.bg_rect = RoundedRectangle(pos=row.pos, size=row.size, radius=[6])
Color(0.341, 0.235, 0.980, 1)
row.border_line = Line(rounded_rectangle=[row.x, row.y, row.width, row.height, 6], width=1)
# Use a closure to bind the correct row instance
def make_update_bg_rect(r):
def update_bg_rect(instance, value):
r.bg_rect.pos = r.pos
r.bg_rect.size = r.size
r.border_line.rounded_rectangle = [r.x, r.y, r.width, r.height, 6]
return update_bg_rect
row.bind(pos=make_update_bg_rect(row), size=make_update_bg_rect(row))
project_label = Label(
text=project,
size_hint_x=0.64,
color=(1, 1, 1, 1),
font_size=15,
shorten=True,
shorten_from='right'
)
edit_button = IconButton(
source="resources/images/edit.png",
size_hint_x=0.18,
allow_stretch=True,
keep_ratio=True
)
edit_button.bind(on_press=lambda instance, p=project: self.edit_project(p))
delete_button = IconButton(
source="resources/images/delete.png",
size_hint_x=0.18,
allow_stretch=True,
keep_ratio=True
)
delete_button.bind(on_press=lambda instance, p=project: self.confirm_delete_project(p))
row.add_widget(project_label)
row.add_widget(edit_button)
row.add_widget(delete_button)
self.ids.projects_list.add_widget(row)
def open_project(self, project_name):
"""Handle opening an existing project/trip."""
print(f"Opening project: {project_name}")
self.ids.result_label.text = f"Opened project: {project_name}"
def create_new_project(self):
"""Navigate to the GetTripFromServer screen to create a new project/trip."""
self.manager.current = "get_trip_from_server"
def edit_project(self, project_name):
# Set the project name on the CreateAnimationScreen before switching
create_anim_screen = self.manager.get_screen("create_animation")
create_anim_screen.project_name = project_name
self.manager.current = "create_animation"
# delete or archive project
def confirm_delete_project(self, project_name):
from kivy.uix.popup import Popup
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.button import Button
from kivy.uix.label import Label
layout = BoxLayout(orientation='vertical', spacing=10, padding=10)
label = Label(text=f"Delete project '{project_name}'?\nChoose an option:")
btn_delete = Button(text="Delete Completely", background_color=(1, 0, 0, 1))
btn_archive = Button(text="Archive Trip", background_color=(0.341, 0.235, 0.980, 1))
btn_cancel = Button(text="Cancel")
popup = Popup(title="Delete Project", content=layout, size_hint=(None, None), size=(400, 250), auto_dismiss=False)
layout.add_widget(label)
layout.add_widget(btn_delete)
layout.add_widget(btn_archive)
layout.add_widget(btn_cancel)
def do_delete(instance):
self.delete_project(project_name)
popup.dismiss()
self.load_existing_projects()
def do_archive(instance):
self.archive_project(project_name)
popup.dismiss()
self.load_existing_projects()
btn_delete.bind(on_press=do_delete)
btn_archive.bind(on_press=do_archive)
btn_cancel.bind(on_press=lambda x: popup.dismiss())
popup.open()
def delete_project(self, project_name):
import shutil
folder_path = os.path.join(RESOURCES_FOLDER, "projects", project_name)
if os.path.exists(folder_path):
shutil.rmtree(folder_path)
def archive_project(self, project_name):
import shutil
src_file = os.path.join(RESOURCES_FOLDER, "projects", project_name, "positions.json")
archive_folder = os.path.join(RESOURCES_FOLDER, "trip_archive")
if not os.path.exists(archive_folder):
os.makedirs(archive_folder)
dst_file = os.path.join(archive_folder, f"{project_name}.json")
if os.path.exists(src_file):
shutil.copy2(src_file, dst_file)
# Optionally, delete the project folder after archiving
self.delete_project(project_name)
class IconButton(ButtonBehavior, Image):
pass
class TraccarApp(App):
def build(self):
if not os.path.exists(RESOURCES_FOLDER):
os.makedirs(RESOURCES_FOLDER)
generate_key()
sm = ScreenManager()
sm.add_widget(LoginScreen(name="login"))
sm.add_widget(HomeScreen(name="home"))
sm.add_widget(GetTripFromServer(name="get_trip_from_server"))
sm.add_widget(SettingsScreen(name="settings"))
sm.add_widget(RegisterScreen(name="register"))
sm.add_widget(CreateAnimationScreen(name="create_animation"))
print("Screens added to ScreenManager:", [screen.name for screen in sm.screens])
return sm
if __name__ == "__main__":
TraccarApp().run()