mirror of
https://github.com/sstent/AICyclingCoach.git
synced 2026-01-25 16:41:58 +00:00
451 lines
16 KiB
Python
451 lines
16 KiB
Python
"""
|
|
Routes view for AI Cycling Coach TUI.
|
|
Displays GPX routes, route management, and visualization.
|
|
"""
|
|
import math
|
|
from typing import List, Dict, Tuple, Optional
|
|
from textual.app import ComposeResult
|
|
from textual.containers import Container, Horizontal, Vertical, ScrollableContainer
|
|
from textual.widgets import (
|
|
Static, DataTable, Button, Input, TextArea, LoadingIndicator,
|
|
TabbedContent, TabPane, Label, DirectoryTree
|
|
)
|
|
from textual.widget import Widget
|
|
from textual.reactive import reactive
|
|
from textual.message import Message
|
|
|
|
from backend.app.database import AsyncSessionLocal
|
|
from tui.services.route_service import RouteService
|
|
|
|
|
|
class GPXVisualization(Widget):
|
|
"""ASCII-based GPX route visualization."""
|
|
|
|
def __init__(self, route_data: Dict):
|
|
super().__init__()
|
|
self.route_data = route_data
|
|
|
|
def compose(self) -> ComposeResult:
|
|
"""Create GPX visualization."""
|
|
yield Label(f"Route: {self.route_data.get('name', 'Unknown')}")
|
|
|
|
# Route summary
|
|
distance = self.route_data.get('total_distance', 0)
|
|
elevation = self.route_data.get('elevation_gain', 0)
|
|
|
|
summary = f"Distance: {distance/1000:.2f} km | Elevation Gain: {elevation:.0f} m"
|
|
yield Static(summary)
|
|
|
|
# ASCII route visualization
|
|
if self.route_data.get('track_points'):
|
|
yield self.create_route_map()
|
|
yield self.create_elevation_profile()
|
|
else:
|
|
yield Static("No track data available for visualization")
|
|
|
|
def create_route_map(self) -> Static:
|
|
"""Create ASCII map of the route."""
|
|
track_points = self.route_data.get('track_points', [])
|
|
if not track_points:
|
|
return Static("No track points available")
|
|
|
|
# Extract coordinates
|
|
lats = [float(p.get('lat', 0)) for p in track_points if p.get('lat')]
|
|
lons = [float(p.get('lon', 0)) for p in track_points if p.get('lon')]
|
|
|
|
if not lats or not lons:
|
|
return Static("Invalid coordinate data")
|
|
|
|
# Normalize coordinates to terminal space
|
|
width, height = 60, 20
|
|
|
|
min_lat, max_lat = min(lats), max(lats)
|
|
min_lon, max_lon = min(lons), max(lons)
|
|
|
|
# Avoid division by zero
|
|
lat_range = max_lat - min_lat if max_lat != min_lat else 1
|
|
lon_range = max_lon - min_lon if max_lon != min_lon else 1
|
|
|
|
# Create ASCII grid
|
|
grid = [[' ' for _ in range(width)] for _ in range(height)]
|
|
|
|
# Plot route points
|
|
for i, (lat, lon) in enumerate(zip(lats, lons)):
|
|
x = int((lon - min_lon) / lon_range * (width - 1))
|
|
y = int((lat - min_lat) / lat_range * (height - 1))
|
|
|
|
# Use different characters for start, end, and middle
|
|
if i == 0:
|
|
char = 'S' # Start
|
|
elif i == len(lats) - 1:
|
|
char = 'E' # End
|
|
else:
|
|
char = '●' # Route point
|
|
|
|
if 0 <= y < height and 0 <= x < width:
|
|
grid[height - 1 - y][x] = char # Flip Y axis
|
|
|
|
# Convert grid to string
|
|
map_lines = [''.join(row) for row in grid]
|
|
map_text = "Route Map:\n" + '\n'.join(map_lines)
|
|
map_text += f"\nS = Start, E = End, ● = Route"
|
|
|
|
return Static(map_text)
|
|
|
|
def create_elevation_profile(self) -> Static:
|
|
"""Create ASCII elevation profile."""
|
|
track_points = self.route_data.get('track_points', [])
|
|
elevations = [float(p.get('ele', 0)) for p in track_points if p.get('ele')]
|
|
|
|
if not elevations:
|
|
return Static("No elevation data available")
|
|
|
|
# Normalize elevation data
|
|
width = 60
|
|
height = 10
|
|
|
|
min_ele, max_ele = min(elevations), max(elevations)
|
|
ele_range = max_ele - min_ele if max_ele != min_ele else 1
|
|
|
|
# Sample elevations to fit width
|
|
if len(elevations) > width:
|
|
step = len(elevations) // width
|
|
elevations = elevations[::step][:width]
|
|
|
|
# Create elevation profile
|
|
profile_lines = []
|
|
for h in range(height):
|
|
line = []
|
|
threshold = min_ele + (height - h) / height * ele_range
|
|
|
|
for ele in elevations:
|
|
if ele >= threshold:
|
|
line.append('█')
|
|
else:
|
|
line.append(' ')
|
|
profile_lines.append(''.join(line))
|
|
|
|
# Add elevation markers
|
|
profile_text = f"Elevation Profile ({min_ele:.0f}m - {max_ele:.0f}m):\n"
|
|
profile_text += '\n'.join(profile_lines)
|
|
|
|
return Static(profile_text)
|
|
|
|
|
|
class RouteFileUpload(Widget):
|
|
"""File upload widget for GPX files."""
|
|
|
|
def compose(self) -> ComposeResult:
|
|
"""Create file upload interface."""
|
|
yield Label("Upload GPX Files")
|
|
yield Button("Browse Files", id="browse-gpx-btn", variant="primary")
|
|
yield Static("", id="upload-status")
|
|
|
|
# Directory tree for local file browsing
|
|
yield Label("Or browse local files:")
|
|
yield DirectoryTree("./data/gpx", id="gpx-directory")
|
|
|
|
|
|
class RouteView(Widget):
|
|
"""Route management and visualization view."""
|
|
|
|
# Reactive attributes
|
|
routes = reactive([])
|
|
selected_route = reactive(None)
|
|
loading = reactive(True)
|
|
|
|
DEFAULT_CSS = """
|
|
.view-title {
|
|
text-align: center;
|
|
color: $accent;
|
|
text-style: bold;
|
|
margin-bottom: 1;
|
|
}
|
|
|
|
.section-title {
|
|
text-style: bold;
|
|
color: $primary;
|
|
margin: 1 0;
|
|
}
|
|
|
|
.route-column {
|
|
width: 1fr;
|
|
margin: 0 1;
|
|
}
|
|
|
|
.upload-container {
|
|
border: solid $primary;
|
|
padding: 1;
|
|
margin: 1 0;
|
|
}
|
|
|
|
.button-row {
|
|
margin: 1 0;
|
|
}
|
|
|
|
.visualization-container {
|
|
border: solid $secondary;
|
|
padding: 1;
|
|
margin: 1 0;
|
|
min-height: 30;
|
|
}
|
|
"""
|
|
|
|
class RouteSelected(Message):
|
|
"""Message sent when a route is selected."""
|
|
def __init__(self, route_id: int):
|
|
super().__init__()
|
|
self.route_id = route_id
|
|
|
|
class RouteUploaded(Message):
|
|
"""Message sent when a route is uploaded."""
|
|
def __init__(self, route_data: Dict):
|
|
super().__init__()
|
|
self.route_data = route_data
|
|
|
|
def compose(self) -> ComposeResult:
|
|
"""Create route view layout."""
|
|
yield Static("Routes & GPX Files", classes="view-title")
|
|
|
|
if self.loading:
|
|
yield LoadingIndicator(id="routes-loader")
|
|
else:
|
|
with TabbedContent():
|
|
with TabPane("Route List", id="route-list-tab"):
|
|
yield self.compose_route_list()
|
|
|
|
with TabPane("Upload GPX", id="upload-gpx-tab"):
|
|
yield self.compose_file_upload()
|
|
|
|
if self.selected_route:
|
|
with TabPane("Route Visualization", id="route-viz-tab"):
|
|
yield self.compose_route_visualization()
|
|
|
|
def compose_route_list(self) -> ComposeResult:
|
|
"""Create route list view."""
|
|
with Container():
|
|
with Horizontal(classes="button-row"):
|
|
yield Button("Refresh", id="refresh-routes-btn")
|
|
yield Button("Import GPX", id="import-gpx-btn", variant="primary")
|
|
yield Button("Analyze Routes", id="analyze-routes-btn")
|
|
|
|
# Routes table
|
|
routes_table = DataTable(id="routes-table")
|
|
routes_table.add_columns("Name", "Distance", "Elevation", "Sections", "Actions")
|
|
yield routes_table
|
|
|
|
# Route sections (if any)
|
|
yield Static("Route Sections", classes="section-title")
|
|
sections_table = DataTable(id="sections-table")
|
|
sections_table.add_columns("Section", "Distance", "Grade", "Difficulty")
|
|
yield sections_table
|
|
|
|
def compose_file_upload(self) -> ComposeResult:
|
|
"""Create file upload view."""
|
|
with Container(classes="upload-container"):
|
|
yield RouteFileUpload()
|
|
|
|
def compose_route_visualization(self) -> ComposeResult:
|
|
"""Create route visualization view."""
|
|
if not self.selected_route:
|
|
yield Static("No route selected")
|
|
return
|
|
|
|
with Container(classes="visualization-container"):
|
|
yield GPXVisualization(self.selected_route)
|
|
|
|
async def on_mount(self) -> None:
|
|
"""Load route data when mounted."""
|
|
try:
|
|
await self.load_routes_data()
|
|
except Exception as e:
|
|
self.log(f"Routes loading error: {e}", severity="error")
|
|
self.loading = False
|
|
self.refresh()
|
|
|
|
async def load_routes_data(self) -> None:
|
|
"""Load routes data."""
|
|
try:
|
|
async with AsyncSessionLocal() as db:
|
|
route_service = RouteService(db)
|
|
self.routes = await route_service.get_routes()
|
|
|
|
# Update loading state
|
|
self.loading = False
|
|
self.refresh()
|
|
|
|
# Populate UI elements
|
|
await self.populate_routes_table()
|
|
|
|
except Exception as e:
|
|
self.log(f"Error loading routes data: {e}", severity="error")
|
|
self.loading = False
|
|
self.refresh()
|
|
|
|
async def populate_routes_table(self) -> None:
|
|
"""Populate the routes table."""
|
|
try:
|
|
routes_table = self.query_one("#routes-table", DataTable)
|
|
routes_table.clear()
|
|
|
|
for route in self.routes:
|
|
distance_km = route.get("total_distance", 0) / 1000
|
|
elevation_m = route.get("elevation_gain", 0)
|
|
|
|
routes_table.add_row(
|
|
route.get("name", "Unknown"),
|
|
f"{distance_km:.1f} km",
|
|
f"{elevation_m:.0f} m",
|
|
"0", # TODO: Count sections
|
|
"View | Edit"
|
|
)
|
|
|
|
except Exception as e:
|
|
self.log(f"Error populating routes table: {e}", severity="error")
|
|
|
|
async def on_button_pressed(self, event: Button.Pressed) -> None:
|
|
"""Handle button press events."""
|
|
try:
|
|
if event.button.id == "refresh-routes-btn":
|
|
await self.refresh_routes()
|
|
elif event.button.id == "import-gpx-btn":
|
|
await self.show_file_upload()
|
|
elif event.button.id == "browse-gpx-btn":
|
|
await self.browse_gpx_files()
|
|
elif event.button.id == "analyze-routes-btn":
|
|
await self.analyze_routes()
|
|
|
|
except Exception as e:
|
|
self.log(f"Button press error: {e}", severity="error")
|
|
|
|
async def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
|
|
"""Handle row selection in routes table."""
|
|
try:
|
|
if event.data_table.id == "routes-table":
|
|
# Get route index from row selection
|
|
row_index = event.row_key.value if hasattr(event.row_key, 'value') else event.cursor_row
|
|
|
|
if 0 <= row_index < len(self.routes):
|
|
selected_route = self.routes[row_index]
|
|
await self.show_route_visualization(selected_route)
|
|
|
|
except Exception as e:
|
|
self.log(f"Row selection error: {e}", severity="error")
|
|
|
|
async def on_directory_tree_file_selected(self, event) -> None:
|
|
"""Handle file selection from directory tree."""
|
|
try:
|
|
file_path = str(event.path)
|
|
if file_path.lower().endswith('.gpx'):
|
|
await self.load_gpx_file(file_path)
|
|
|
|
except Exception as e:
|
|
self.log(f"File selection error: {e}", severity="error")
|
|
|
|
async def show_route_visualization(self, route_data: Dict) -> None:
|
|
"""Show visualization for a route."""
|
|
try:
|
|
# Load additional route data if needed
|
|
async with AsyncSessionLocal() as db:
|
|
route_service = RouteService(db)
|
|
full_route_data = await route_service.load_gpx_file(
|
|
route_data.get("gpx_file_path", "")
|
|
)
|
|
|
|
self.selected_route = full_route_data
|
|
self.refresh()
|
|
|
|
# Switch to visualization tab
|
|
tabs = self.query_one(TabbedContent)
|
|
tabs.active = "route-viz-tab"
|
|
|
|
# Post message that route was selected
|
|
self.post_message(self.RouteSelected(route_data["id"]))
|
|
|
|
except Exception as e:
|
|
self.log(f"Error showing route visualization: {e}", severity="error")
|
|
|
|
async def refresh_routes(self) -> None:
|
|
"""Refresh the routes list."""
|
|
self.loading = True
|
|
self.refresh()
|
|
await self.load_routes_data()
|
|
|
|
async def show_file_upload(self) -> None:
|
|
"""Switch to the file upload tab."""
|
|
tabs = self.query_one(TabbedContent)
|
|
tabs.active = "upload-gpx-tab"
|
|
|
|
async def browse_gpx_files(self) -> None:
|
|
"""Browse for GPX files."""
|
|
try:
|
|
# Update status
|
|
status = self.query_one("#upload-status", Static)
|
|
status.update("Click on a .gpx file in the directory tree below")
|
|
|
|
except Exception as e:
|
|
self.log(f"Error browsing files: {e}", severity="error")
|
|
|
|
async def load_gpx_file(self, file_path: str) -> None:
|
|
"""Load a GPX file and create route visualization."""
|
|
try:
|
|
self.log(f"Loading GPX file: {file_path}", severity="info")
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
route_service = RouteService(db)
|
|
route_data = await route_service.load_gpx_file(file_path)
|
|
|
|
# Create visualization
|
|
self.selected_route = route_data
|
|
self.refresh()
|
|
|
|
# Switch to visualization tab
|
|
tabs = self.query_one(TabbedContent)
|
|
tabs.active = "route-viz-tab"
|
|
|
|
# Update upload status
|
|
status = self.query_one("#upload-status", Static)
|
|
status.update(f"Loaded: {route_data.get('name', 'Unknown Route')}")
|
|
|
|
# Post message about route upload
|
|
self.post_message(self.RouteUploaded(route_data))
|
|
|
|
except Exception as e:
|
|
self.log(f"Error loading GPX file: {e}", severity="error")
|
|
|
|
# Update status with error
|
|
try:
|
|
status = self.query_one("#upload-status", Static)
|
|
status.update(f"Error: {str(e)}")
|
|
except:
|
|
pass
|
|
|
|
async def analyze_routes(self) -> None:
|
|
"""Analyze all routes for insights."""
|
|
try:
|
|
if not self.routes:
|
|
self.log("No routes to analyze", severity="warning")
|
|
return
|
|
|
|
# Calculate route statistics
|
|
total_distance = sum(r.get("total_distance", 0) for r in self.routes) / 1000
|
|
total_elevation = sum(r.get("elevation_gain", 0) for r in self.routes)
|
|
avg_distance = total_distance / len(self.routes)
|
|
|
|
analysis = f"""Route Analysis:
|
|
• Total Routes: {len(self.routes)}
|
|
• Total Distance: {total_distance:.1f} km
|
|
• Total Elevation: {total_elevation:.0f} m
|
|
• Average Distance: {avg_distance:.1f} km
|
|
• Average Elevation: {total_elevation / len(self.routes):.0f} m"""
|
|
|
|
self.log("Route Analysis Complete", severity="info")
|
|
self.log(analysis, severity="info")
|
|
|
|
except Exception as e:
|
|
self.log(f"Error analyzing routes: {e}", severity="error")
|
|
|
|
def watch_loading(self, loading: bool) -> None:
|
|
"""React to loading state changes."""
|
|
if hasattr(self, '_mounted') and self._mounted:
|
|
self.refresh() |