📊 GPX Track Analyzer
', unsafe_allow_html=True) + st.markdown("### Upload and analyze your GPX track files") + + # File uploader + uploaded_file = st.file_uploader( + "Choose a GPX file", + type=["gpx"], + help="Upload a GPX file to visualize and analyze the track", + ) + + if uploaded_file is None: + # Show example files if available + st.info("💡 **Tip:** Upload a GPX file to get started, or use one of the example files below.") + + example_files = [ + "activity_19218242997.gpx", + "activity_22095267736.gpx", + ] + + example_dir = Path(__file__).parent.parent / "firstexample" + if example_dir.exists(): + st.subheader("Example Files") + cols = st.columns(len(example_files)) + for i, example_file in enumerate(example_files): + with cols[i]: + if (example_dir / example_file).exists(): + if st.button(f"Load {example_file}", use_container_width=True): + with open(example_dir / example_file, "rb") as f: + uploaded_file = f.read() + uploaded_file = tempfile.NamedTemporaryFile( + delete=False, suffix=".gpx" + ) + uploaded_file.write(uploaded_file) + uploaded_file.seek(0) + + if uploaded_file is not None: + # Process the uploaded file + try: + with st.spinner("Processing GPX file..."): + # Save to temp file + if hasattr(uploaded_file, "name"): + temp_file_path = uploaded_file.name + else: + temp_file = tempfile.NamedTemporaryFile( + delete=False, suffix=".gpx" + ) + temp_file.write(uploaded_file.getvalue()) + temp_file_path = temp_file.name + temp_file.close() + + # Load the track + track = GpxTrack(temp_file_path) + + # Display success message + st.success(f"✅ Successfully loaded GPX file with {len(track.points)} points") + + # Sidebar for analysis options + st.sidebar.header("Analysis Options") + + # Speed unit selection + speed_unit = st.sidebar.selectbox( + "Speed Unit", + ["m/s", "km/h", "knots"], + index=2, # Default to knots + ) + + # Simulation options + st.sidebar.subheader("Simulation Settings") + show_simulation = st.sidebar.checkbox("Show Simulation Analysis", value=False) + + if show_simulation: + method = st.sidebar.selectbox( + "Simulation Method", + ["ar1", "gaussian"], + index=0, + ) + sample_size = st.sidebar.slider( + "Number of Simulations", + 100, 1000, 500, 100, + ) + + if method == "ar1": + sigma_tot = st.sidebar.slider( + "Sigma Total", 0.1, 5.0, 2.5, 0.1 + ) + phi = st.sidebar.slider( + "Phi (AR-1 coefficient)", 0.0, 0.99, 0.9, 0.01 + ) + else: # gaussian + amplitude = st.sidebar.slider( + "Amplitude", 0.1, 5.0, 1.5, 0.1 + ) + scale = st.sidebar.slider( + "Scale", 0.1, 10.0, 5.0, 0.1 + ) + + # Best segment analysis + st.sidebar.subheader("Best Segment Analysis") + show_best_segment = st.sidebar.checkbox("Show Best Segment Analysis", value=False) + + if show_best_segment: + target_type = st.sidebar.radio( + "Target Type", + ["Distance", "Time"], + horizontal=True, + ) + if target_type == "Distance": + target_value = st.sidebar.number_input( + "Target Distance (meters)", + min_value=100.0, + max_value=10000.0, + value=500.0, + step=100.0, + ) + else: + target_value = st.sidebar.number_input( + "Target Time (seconds)", + min_value=1.0, + max_value=600.0, + value=10.0, + step=1.0, + ) + + # Display basic statistics + st.subheader("📈 Track Statistics") + + col1, col2, col3, col4 = st.columns(4) + + with col1: + distance = track.get_distance() + st.metric("Distance", f"{distance:.2f} m") + + with col2: + duration = track.get_duration() + st.metric("Duration", f"{duration:.2f} s") + + with col3: + avg_speed = track.get_average_speed() + if speed_unit == "km/h": + avg_speed_display = avg_speed * 3.6 + unit = "km/h" + elif speed_unit == "knots": + avg_speed_display = avg_speed * 1.94384 + unit = "knots" + else: + avg_speed_display = avg_speed + unit = "m/s" + st.metric("Avg Speed", f"{avg_speed_display:.2f} {unit}") + + with col4: + if len(track.points) > 0: + start_time = track.points[0].time + end_time = track.points[-1].time + if start_time and end_time: + st.metric("Time Range", f"{start_time.strftime('%Y-%m-%d %H:%M')}") + + # Best segment analysis + if show_best_segment: + st.subheader("🏆 Best Segment Analysis") + + if target_type == "Distance": + start_idx, end_idx, speed = track.get_best_segment_for_distance(target_value) + segment_type = "distance" + else: + start_idx, end_idx, speed = track.get_best_segment_for_time(target_value) + segment_type = "time" + + col1, col2, col3 = st.columns(3) + with col1: + st.metric("Start Index", start_idx) + with col2: + st.metric("End Index", end_idx) + with col3: + if speed_unit == "km/h": + speed_display = speed / 1.94384 * 3.6 # Convert from knots to km/h + unit = "km/h" + elif speed_unit == "m/s": + speed_display = speed / 1.94384 # Convert from knots to m/s + unit = "m/s" + else: + speed_display = speed + unit = "knots" + st.metric("Best Speed", f"{speed_display:.2f} {unit}") + + # Simulation analysis + if show_simulation: + st.subheader("🎲 Simulation Analysis") + + with st.spinner("Running simulations..."): + if method == "ar1": + process_sample = track.processSample( + sample_size=sample_size, + method=method, + sigma_tot=sigma_tot, + phi=phi, + ) + else: + process_sample = track.processSample( + sample_size=sample_size, + method=method, + amplitude=amplitude, + scale=scale, + ) + + # Calculate quantiles + quantiles = process_sample.computeQuantilePerComponent([0.025, 0.5, 0.975]) + + col1, col2, col3 = st.columns(3) + with col1: + lower = quantiles[0][0][0] + if speed_unit == "km/h": + lower_display = lower * 3.6 + unit = "km/h" + elif speed_unit == "m/s": + lower_display = lower + unit = "m/s" + else: + lower_display = lower * 1.94384 + unit = "knots" + st.metric("2.5% Quantile", f"{lower_display:.2f} {unit}") + + with col2: + median = quantiles[1][0][0] + if speed_unit == "km/h": + median_display = median * 3.6 + elif speed_unit == "m/s": + median_display = median + else: + median_display = median * 1.94384 + st.metric("Median", f"{median_display:.2f} {unit}") + + with col3: + upper = quantiles[2][0][0] + if speed_unit == "km/h": + upper_display = upper * 3.6 + elif speed_unit == "m/s": + upper_display = upper + else: + upper_display = upper * 1.94384 + st.metric("97.5% Quantile", f"{upper_display:.2f} {unit}") + + # Tabs for different visualizations + st.subheader("📊 Visualizations") + + tab1, tab2, tab3, tab4 = st.tabs([ + "Track Map", + "Speed Over Time", + "Elevation Profile", + "Speed Distribution" + ]) + + with tab1: + st.write("### Track Map with Speed Coloring") + fig = plot_track_with_speed(track, speed_unit=speed_unit) + st.pyplot(fig) + + # Download plot + if st.button("📥 Download Track Plot", key="download_track_plot"): + fig.savefig("track_plot.png", dpi=300, bbox_inches="tight") + with open("track_plot.png", "rb") as f: + st.download_button( + label="Download Track Plot", + data=f, + file_name="track_plot.png", + mime="image/png", + ) + + with tab2: + st.write("### Speed Over Time") + fig = plot_speed_over_time(track, speed_unit=speed_unit) + st.pyplot(fig) + + if show_simulation: + st.write("### Speed Over Time with Confidence Intervals") + fig_ci = plot_speed_over_time( + track, + speed_unit=speed_unit, + process_sample=process_sample + ) + st.pyplot(fig_ci) + + with tab3: + st.write("### Elevation Profile") + fig = plot_elevation_profile(track) + st.pyplot(fig) + + with tab4: + st.write("### Speed Distribution") + fig = plot_speed_distribution(track, speed_unit=speed_unit) + st.pyplot(fig) + + # PDF Report Generation + st.subheader("📄 Generate PDF Report") + + st.write("Generate a comprehensive PDF report with all analysis and visualizations.") + + report_options = st.multiselect( + "Include in Report", + [ + "Track Statistics", + "Track Map", + "Speed Over Time", + "Elevation Profile", + "Speed Distribution", + "Simulation Analysis" if show_simulation else None, + "Best Segment Analysis" if show_best_segment else None, + ], + default=[ + "Track Statistics", + "Track Map", + "Speed Over Time", + "Elevation Profile", + "Speed Distribution", + ], + ) + + # Filter out None values + report_options = [opt for opt in report_options if opt is not None] + + if st.button("📥 Generate PDF Report", type="primary", use_container_width=True): + with st.spinner("Generating PDF report..."): + # Create report data + report_data = { + "track": track, + "speed_unit": speed_unit, + "show_simulation": show_simulation, + "process_sample": process_sample if show_simulation else None, + "show_best_segment": show_best_segment, + "target_type": target_type if show_best_segment else None, + "target_value": target_value if show_best_segment else None, + "start_idx": start_idx if show_best_segment else None, + "end_idx": end_idx if show_best_segment else None, + "best_speed": speed if show_best_segment else None, + "method": method if show_simulation else None, + "sample_size": sample_size if show_simulation else None, + "sigma_tot": sigma_tot if show_simulation and method == "ar1" else None, + "phi": phi if show_simulation and method == "ar1" else None, + "amplitude": amplitude if show_simulation and method == "gaussian" else None, + "scale": scale if show_simulation and method == "gaussian" else None, + } + + # Generate PDF + pdf_bytes = generate_pdf_report( + report_data, + report_options, + uploaded_file.name if hasattr(uploaded_file, "name") else "track" + ) + + # Download button + st.success("✅ PDF report generated successfully!") + st.download_button( + label="Download PDF Report", + data=pdf_bytes, + file_name=f"gpx_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf", + mime="application/pdf", + use_container_width=True, + ) + + # Cleanup temp file + if os.path.exists(temp_file_path): + os.unlink(temp_file_path) + + except Exception as e: + st.error(f"❌ Error processing GPX file: {str(e)}") + st.error(f"Error type: {type(e).__name__}") + import traceback + with st.expander("Show full error traceback"): + st.code(traceback.format_exc()) + + else: + # Show welcome message + st.info("👆 Please upload a GPX file to begin analysis") + + # Show example usage + with st.expander("❓ How to use this app"): + st.markdown(""" + ### How to Use GPX Track Analyzer + + 1. **Upload a GPX file** using the file uploader above + 2. **View track statistics** including distance, duration, and average speed + 3. **Explore visualizations** in the tabs: + - Track Map: See your route with speed coloring + - Speed Over Time: Analyze speed variations + - Elevation Profile: View elevation changes + - Speed Distribution: See speed histogram + + 4. **Optional Analysis**: + - Enable simulation analysis in the sidebar to run stochastic simulations + - Enable best segment analysis to find optimal segments + + 5. **Generate PDF Report**: Create a comprehensive report with all your analysis + + ### Features + - ✅ Visualize GPX tracks with speed coloring + - ✅ Calculate track statistics (distance, duration, speed) + - ✅ Run AR-1 and Gaussian process simulations + - ✅ Find best segments for distance or time targets + - ✅ Generate downloadable PDF reports + - ✅ Support for multiple speed units (m/s, km/h, knots) + """) + + +if __name__ == "__main__": + main() diff --git a/streamlitApp/pdf_report.py b/streamlitApp/pdf_report.py new file mode 100644 index 0000000..813d7f2 --- /dev/null +++ b/streamlitApp/pdf_report.py @@ -0,0 +1,416 @@ +""" +PDF report generation for GPX track analysis. + +This module provides functions to generate comprehensive PDF reports +with analysis results and visualizations. +""" + +import io +import tempfile +import os +from datetime import datetime +from pathlib import Path + +# Try to import reportlab, fall back to matplotlib for PDF generation +try: + from reportlab.lib.pagesizes import letter, A4 + from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle + from reportlab.lib.units import inch, mm + from reportlab.platypus import ( + SimpleDocTemplate, + Paragraph, + Spacer, + Image, + Table, + TableStyle, + PageBreak, + ) + from reportlab.lib import colors + REPORTLAB_AVAILABLE = True +except ImportError: + REPORTLAB_AVAILABLE = False + +import matplotlib +matplotlib.use('Agg') +import matplotlib.pyplot as plt +from streamlitApp.visualization import ( + plot_track_with_speed, + plot_speed_over_time, + plot_elevation_profile, + plot_speed_distribution, +) + + +def generate_pdf_report(report_data, report_options, filename="gpx_report"): + """ + Generate a PDF report with track analysis and visualizations. + + Args: + report_data: Dictionary containing all report data + report_options: List of sections to include in the report + filename: Base filename for the report + + Returns: + bytes: PDF content as bytes + """ + if REPORTLAB_AVAILABLE: + return _generate_pdf_with_reportlab(report_data, report_options, filename) + else: + return _generate_pdf_with_matplotlib(report_data, report_options, filename) + + +def _generate_pdf_with_reportlab(report_data, report_options, filename): + """ + Generate PDF using reportlab (preferred method). + """ + # Extract data + track = report_data["track"] + speed_unit = report_data.get("speed_unit", "knots") + + # Create a buffer for the PDF + buffer = io.BytesIO() + doc = SimpleDocTemplate(buffer, pagesize=letter, + rightMargin=36, leftMargin=36, + topMargin=36, bottomMargin=36) + + # Create styles + styles = getSampleStyleSheet() + + # Custom styles + title_style = ParagraphStyle( + name="Title", + parent=styles["Heading1"], + fontSize=24, + leading=28, + alignment=1, # Center + spaceAfter=20, + ) + + heading1_style = ParagraphStyle( + name="Heading1", + parent=styles["Heading1"], + fontSize=18, + leading=22, + spaceBefore=12, + spaceAfter=6, + ) + + heading2_style = ParagraphStyle( + name="Heading2", + parent=styles["Heading2"], + fontSize=14, + leading=18, + spaceBefore=12, + spaceAfter=6, + ) + + normal_style = ParagraphStyle( + name="Normal", + parent=styles["Normal"], + fontSize=10, + leading=14, + spaceAfter=6, + ) + + # Story for the document + story = [] + + # Title + story.append(Paragraph("GPX Track Analysis Report", title_style)) + story.append(Paragraph(f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", + normal_style)) + story.append(Spacer(1, 0.2 * inch)) + + # Track Information + if "Track Statistics" in report_options: + story.append(Paragraph("Track Information", heading1_style)) + story.append(Spacer(1, 0.1 * inch)) + + # Create track info table + distance = track.get_distance() + duration = track.get_duration() + avg_speed = track.get_average_speed() + + # Convert speed to desired unit + if speed_unit == "km/h": + avg_speed_display = avg_speed * 3.6 + unit = "km/h" + elif speed_unit == "knots": + avg_speed_display = avg_speed * 1.94384 + unit = "knots" + else: + avg_speed_display = avg_speed + unit = "m/s" + + # Get time range + if track.points and track.points[0].time and track.points[-1].time: + start_time = track.points[0].time.strftime('%Y-%m-%d %H:%M:%S') + end_time = track.points[-1].time.strftime('%Y-%m-%d %H:%M:%S') + else: + start_time = "N/A" + end_time = "N/A" + + track_info_data = [ + ["Property", "Value"], + ["Distance", f"{distance:.2f} meters"], + ["Duration", f"{duration:.2f} seconds"], + ["Average Speed", f"{avg_speed_display:.2f} {unit}"], + ["Number of Points", str(len(track.points))], + ["Start Time", start_time], + ["End Time", end_time], + ] + + track_info_table = Table(track_info_data, colWidths=[2 * inch, 3 * inch]) + track_info_table.setStyle(TableStyle([ + ('BACKGROUND', (0, 0), (-1, 0), colors.grey), + ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), + ('ALIGN', (0, 0), (-1, -1), 'LEFT'), + ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), + ('FONTSIZE', (0, 0), (-1, 0), 12), + ('BOTTOMPADDING', (0, 0), (-1, 0), 12), + ('BACKGROUND', (0, 1), (-1, -1), colors.beige), + ('GRID', (0, 0), (-1, -1), 1, colors.black), + ])) + + story.append(track_info_table) + story.append(Spacer(1, 0.2 * inch)) + + # Best Segment Analysis + if "Best Segment Analysis" in report_options and report_data.get("show_best_segment"): + story.append(Paragraph("Best Segment Analysis", heading1_style)) + story.append(Spacer(1, 0.1 * inch)) + + target_type = report_data.get("target_type", "Distance") + target_value = report_data.get("target_value", 500) + start_idx = report_data.get("start_idx", 0) + end_idx = report_data.get("end_idx", 0) + best_speed = report_data.get("best_speed", 0) + + # Convert speed to desired unit + if speed_unit == "km/h": + best_speed_display = best_speed / 1.94384 * 3.6 + unit = "km/h" + elif speed_unit == "m/s": + best_speed_display = best_speed / 1.94384 + unit = "m/s" + else: + best_speed_display = best_speed + unit = "knots" + + segment_info_data = [ + ["Property", "Value"], + ["Target Type", target_type], + ["Target Value", f"{target_value}"], + ["Start Index", str(start_idx)], + ["End Index", str(end_idx)], + ["Best Speed", f"{best_speed_display:.2f} {unit}"], + ] + + segment_table = Table(segment_info_data, colWidths=[2 * inch, 3 * inch]) + segment_table.setStyle(TableStyle([ + ('BACKGROUND', (0, 0), (-1, 0), colors.grey), + ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), + ('ALIGN', (0, 0), (-1, -1), 'LEFT'), + ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), + ('FONTSIZE', (0, 0), (-1, 0), 12), + ('BOTTOMPADDING', (0, 0), (-1, 0), 12), + ('BACKGROUND', (0, 1), (-1, -1), colors.lightgrey), + ('GRID', (0, 0), (-1, -1), 1, colors.black), + ])) + + story.append(segment_table) + story.append(Spacer(1, 0.2 * inch)) + + # Simulation Analysis + if "Simulation Analysis" in report_options and report_data.get("show_simulation"): + story.append(Paragraph("Simulation Analysis", heading1_style)) + story.append(Spacer(1, 0.1 * inch)) + + method = report_data.get("method", "ar1") + sample_size = report_data.get("sample_size", 500) + + sim_info = [ + ["Property", "Value"], + ["Method", method.upper()], + ["Number of Simulations", str(sample_size)], + ] + + if method == "ar1": + sigma_tot = report_data.get("sigma_tot", 2.5) + phi = report_data.get("phi", 0.9) + sim_info.append(["Sigma Total", f"{sigma_tot}"]) + sim_info.append(["Phi (AR-1 coefficient)", f"{phi}"]) + else: + amplitude = report_data.get("amplitude", 1.5) + scale = report_data.get("scale", 5.0) + sim_info.append(["Amplitude", f"{amplitude}"]) + sim_info.append(["Scale", f"{scale}"]) + + sim_table = Table(sim_info, colWidths=[2 * inch, 3 * inch]) + sim_table.setStyle(TableStyle([ + ('BACKGROUND', (0, 0), (-1, 0), colors.grey), + ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), + ('ALIGN', (0, 0), (-1, -1), 'LEFT'), + ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), + ('FONTSIZE', (0, 0), (-1, 0), 12), + ('BOTTOMPADDING', (0, 0), (-1, 0), 12), + ('BACKGROUND', (0, 1), (-1, -1), colors.lightblue), + ('GRID', (0, 0), (-1, -1), 1, colors.black), + ])) + + story.append(sim_table) + story.append(Spacer(1, 0.2 * inch)) + + # Visualizations + temp_dir = tempfile.mkdtemp() + image_paths = [] + + try: + if "Track Map" in report_options: + story.append(Paragraph("Track Map", heading1_style)) + fig = plot_track_with_speed(track, speed_unit=speed_unit, dpi=150) + img_path = os.path.join(temp_dir, "track_map.png") + fig.savefig(img_path, dpi=150, bbox_inches="tight") + plt.close(fig) + + img = Image(img_path, width=6 * inch, height=4 * inch) + story.append(img) + story.append(Spacer(1, 0.2 * inch)) + image_paths.append(img_path) + + if "Speed Over Time" in report_options: + story.append(Paragraph("Speed Over Time", heading1_style)) + fig = plot_speed_over_time(track, speed_unit=speed_unit, dpi=150) + img_path = os.path.join(temp_dir, "speed_time.png") + fig.savefig(img_path, dpi=150, bbox_inches="tight") + plt.close(fig) + + img = Image(img_path, width=6 * inch, height=3 * inch) + story.append(img) + story.append(Spacer(1, 0.2 * inch)) + image_paths.append(img_path) + + if "Elevation Profile" in report_options: + story.append(Paragraph("Elevation Profile", heading1_style)) + fig = plot_elevation_profile(track, dpi=150) + img_path = os.path.join(temp_dir, "elevation.png") + fig.savefig(img_path, dpi=150, bbox_inches="tight") + plt.close(fig) + + img = Image(img_path, width=6 * inch, height=3 * inch) + story.append(img) + story.append(Spacer(1, 0.2 * inch)) + image_paths.append(img_path) + + if "Speed Distribution" in report_options: + story.append(Paragraph("Speed Distribution", heading1_style)) + fig = plot_speed_distribution(track, speed_unit=speed_unit, dpi=150) + img_path = os.path.join(temp_dir, "speed_dist.png") + fig.savefig(img_path, dpi=150, bbox_inches="tight") + plt.close(fig) + + img = Image(img_path, width=6 * inch, height=3 * inch) + story.append(img) + story.append(Spacer(1, 0.2 * inch)) + image_paths.append(img_path) + + # Build the PDF + doc.build(story) + + # Get the PDF bytes + pdf_bytes = buffer.getvalue() + buffer.close() + + return pdf_bytes + + finally: + # Cleanup temporary files + for img_path in image_paths: + if os.path.exists(img_path): + os.remove(img_path) + if os.path.exists(temp_dir): + os.rmdir(temp_dir) + + +def _generate_pdf_with_matplotlib(report_data, report_options, filename): + """ + Fallback PDF generation using matplotlib's PDF backend. + This creates a PDF with plots but less formatting options. + """ + from matplotlib.backends.backend_pdf import PdfPages + + track = report_data["track"] + speed_unit = report_data.get("speed_unit", "knots") + + # Create a buffer for the PDF + buffer = io.BytesIO() + + with PdfPages(buffer) as pdf: + # Add title page + fig = plt.figure(figsize=(11, 8)) + plt.text(0.5, 0.7, "GPX Track Analysis Report", + ha='center', va='center', fontsize=24, fontweight='bold') + plt.text(0.5, 0.5, f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", + ha='center', va='center', fontsize=14) + plt.axis('off') + pdf.savefig(fig) + plt.close(fig) + + # Add track statistics + if "Track Statistics" in report_options: + fig = plt.figure(figsize=(11, 8)) + plt.text(0.5, 0.9, "Track Information", + ha='center', va='center', fontsize=18, fontweight='bold') + + distance = track.get_distance() + duration = track.get_duration() + avg_speed = track.get_average_speed() + + if speed_unit == "km/h": + avg_speed_display = avg_speed * 3.6 + unit = "km/h" + elif speed_unit == "knots": + avg_speed_display = avg_speed * 1.94384 + unit = "knots" + else: + avg_speed_display = avg_speed + unit = "m/s" + + info_text = ( + f"Distance: {distance:.2f} meters\n" + f"Duration: {duration:.2f} seconds\n" + f"Average Speed: {avg_speed_display:.2f} {unit}\n" + f"Number of Points: {len(track.points)}" + ) + + plt.text(0.5, 0.5, info_text, ha='center', va='center', fontsize=12) + plt.axis('off') + pdf.savefig(fig) + plt.close(fig) + + # Add visualizations + if "Track Map" in report_options: + fig = plot_track_with_speed(track, speed_unit=speed_unit, dpi=150) + pdf.savefig(fig) + plt.close(fig) + + if "Speed Over Time" in report_options: + fig = plot_speed_over_time(track, speed_unit=speed_unit, dpi=150) + pdf.savefig(fig) + plt.close(fig) + + if "Elevation Profile" in report_options: + fig = plot_elevation_profile(track, dpi=150) + pdf.savefig(fig) + plt.close(fig) + + if "Speed Distribution" in report_options: + fig = plot_speed_distribution(track, speed_unit=speed_unit, dpi=150) + pdf.savefig(fig) + plt.close(fig) + + # Get the PDF bytes + pdf_bytes = buffer.getvalue() + buffer.close() + + return pdf_bytes diff --git a/streamlitApp/requirements.txt b/streamlitApp/requirements.txt new file mode 100644 index 0000000..a9772a6 --- /dev/null +++ b/streamlitApp/requirements.txt @@ -0,0 +1,15 @@ +# Streamlit App Requirements +# This file contains dependencies specific to the Streamlit application + +# Core dependencies +streamlit>=1.29.0 +reportlab>=4.0.0 + +# For PDF generation fallback +matplotlib>=3.10.8 + +# Parent project dependencies (also needed) +gpxpy>=1.6.2 +numpy>=2.2.6 +openturns>=1.26 +pandas>=2.3.3 diff --git a/streamlitApp/tests/__init__.py b/streamlitApp/tests/__init__.py new file mode 100644 index 0000000..5d6111e --- /dev/null +++ b/streamlitApp/tests/__init__.py @@ -0,0 +1,3 @@ +""" +Tests for the Streamlit application. +""" diff --git a/streamlitApp/tests/conftest.py b/streamlitApp/tests/conftest.py new file mode 100644 index 0000000..1254719 --- /dev/null +++ b/streamlitApp/tests/conftest.py @@ -0,0 +1,199 @@ +""" +Pytest configuration and fixtures for Streamlit app tests. +""" + +import os +import sys +from pathlib import Path + +# Add parent directories to path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest + + +def get_test_file_path(filename): + """Get the absolute path to a test file.""" + return os.path.join(os.path.dirname(__file__), '..', '..', 'firstexample', filename) + + +@pytest.fixture +def sample_track(): + """Fixture to load a sample GPX track.""" + # Mock the GpxTrack class if openturns is not available + try: + from otGpxTrack.Base import GpxTrack + gpx_file = get_test_file_path("activity_19218242997.gpx") + return GpxTrack(gpx_file) + except ImportError: + # Create a mock track for testing + class MockPoint: + def __init__(self, lat, lon, elev, time): + self.latitude = lat + self.longitude = lon + self.elevation = elev + self.time = time + self.speed = 0.0 + + def distance_3d(self, other): + # Simple distance calculation (not accurate but for testing) + return ((self.latitude - other.latitude)**2 + + (self.longitude - other.longitude)**2 + + (self.elevation - other.elevation)**2)**0.5 * 111111 + + class MockTrack: + def __init__(self): + self.points = [] + self.data = None + self.gpx_file_path = "mock.gpx" + self._create_mock_data() + + def _create_mock_data(self): + import numpy as np + from datetime import datetime, timedelta + + # Create 100 mock points + start_time = datetime(2024, 1, 1, 12, 0, 0) + start_lat, start_lon = 48.8566, 2.3522 # Paris coordinates + + for i in range(100): + time = start_time + timedelta(seconds=i) + lat = start_lat + (i * 0.001) + lon = start_lon + (i * 0.001) + elev = i * 10 + self.points.append(MockPoint(lat, lon, elev, time)) + + # Create mock OpenTURNS sample data + import numpy as np + latitudes = [p.latitude for p in self.points] + longitudes = [p.longitude for p in self.points] + elevations = [p.elevation for p in self.points] + times = [p.time.timestamp() for p in self.points] + speeds = [0.0] * len(self.points) + + # Calculate speeds + for i in range(1, len(self.points)): + distance = self.points[i].distance_3d(self.points[i-1]) + time_diff = (self.points[i].time - self.points[i-1].time).total_seconds() + speeds[i] = distance / time_diff if time_diff > 0 else 0.0 + + data = np.column_stack([latitudes, longitudes, elevations, times, speeds]) + + # Mock OpenTURNS Sample + class MockSample: + def __init__(self, data): + self.data = data + self.size = data.shape[0] + self.dimension = data.shape[1] + self.descriptions = ["Latitude", "Longitude", "Elevation", "Time", "Speed"] + + def getSize(self): + return self.size + + def getDimension(self): + return self.dimension + + def getDescription(self): + return self.descriptions + + def __getitem__(self, index): + if isinstance(index, int): + return self.data[index] + else: + return self.data[index] + + self.data = MockSample(data) + + def get_distance(self): + total = 0.0 + for i in range(1, len(self.points)): + total += self.points[i].distance_3d(self.points[i-1]) + return total + + def get_duration(self): + if len(self.points) < 2: + return 0.0 + return (self.points[-1].time - self.points[0].time).total_seconds() + + def get_average_speed(self): + distance = self.get_distance() + duration = self.get_duration() + if duration == 0: + return 0.0 + return distance / duration + + def get_openturns_sample(self): + return self.data + + def processSample(self, sample_size=100, method="ar1", **kwargs): + # Mock process sample + class MockProcessSample: + def __init__(self, size, dimension): + self.size = size + self.dimension = dimension + + def getSize(self): + return self.size + + def getDimension(self): + return self.dimension + + def computeQuantilePerComponent(self, quantiles): + # Return mock quantile data + results = [] + for q in quantiles: + # Return a sample with the same structure + class MockQuantileSample: + def __init__(self, size): + self.size = size + def __getitem__(self, index): + return [index * 0.1] # Mock data + results.append(MockQuantileSample(len(self.points))) + return results + + def getTimeGrid(self): + class MockTimeGrid: + def getN(self): + return len(self.points) + return MockTimeGrid() + + return MockProcessSample(sample_size, 1) + + def simulate_ar1_speeds(self, segment_indices, **kwargs): + # Mock AR-1 simulation + import numpy as np + n_sims = kwargs.get('n_sims', 100) + speeds = np.random.normal(5.0, 1.0, n_sims) + + class MockSample: + def __init__(self, data): + self.data = data + self.size = len(data) + def getSize(self): + return self.size + @staticmethod + def computeQuantilePerComponent(sample, quantile): + return np.percentile(sample.data, quantile * 100) + + mean_speed = np.mean(speeds) + lower = np.percentile(speeds, 2.5) + upper = np.percentile(speeds, 97.5) + + return mean_speed, lower, upper, MockSample(speeds) + + def get_best_segment_for_distance(self, target_distance): + # Mock best segment + return (0, min(10, len(self.points)-1), 5.0) + + def get_best_segment_for_time(self, target_time): + # Mock best segment + return (0, min(10, len(self.points)-1), 5.0) + + def plot_track(self, **kwargs): + # Mock plot + import matplotlib.pyplot as plt + fig, ax = plt.subplots() + return fig + + return MockTrack() diff --git a/streamlitApp/tests/mock_openturns.py b/streamlitApp/tests/mock_openturns.py new file mode 100644 index 0000000..681326e --- /dev/null +++ b/streamlitApp/tests/mock_openturns.py @@ -0,0 +1,131 @@ +""" +Mock OpenTURNS module for testing when the real OpenTURNS is not available. +""" + +import numpy as np + + +class Sample: + """Mock OpenTURNS Sample class.""" + + def __init__(self, data): + if isinstance(data, np.ndarray): + self.data = data + else: + self.data = np.array(data) + self.size = self.data.shape[0] + self.dimension = self.data.shape[1] if len(self.data.shape) > 1 else 1 + self.descriptions = ["Latitude", "Longitude", "Elevation", "Time", "Speed"] + + def getSize(self): + return self.size + + def getDimension(self): + return self.dimension + + def getDescription(self): + return self.descriptions + + def setDescription(self, descriptions): + self.descriptions = descriptions + + def __getitem__(self, index): + if isinstance(index, int): + return Point(self.data[index]) + else: + return Sample(self.data[index]) + + def __len__(self): + return self.size + + +class Point: + """Mock OpenTURNS Point class.""" + + def __init__(self, data): + if isinstance(data, np.ndarray): + self.data = data.flatten() + else: + self.data = np.array(data).flatten() + + def __getitem__(self, index): + return self.data[index] + + def __len__(self): + return len(self.data) + + +class Mesh: + """Mock OpenTURNS Mesh class.""" + + def __init__(self, sample): + self.sample = sample + self.vertices_number = sample.getSize() + + def getVerticesNumber(self): + return self.vertices_number + + +class Field: + """Mock OpenTURNS Field class.""" + + def __init__(self, mesh, values): + self.mesh = mesh + self.values = values + + +class ProcessSample: + """Mock OpenTURNS ProcessSample class.""" + + def __init__(self, mesh, size, dimension): + self.mesh = mesh + self.size = size + self.dimension = dimension + self.fields = [] + + def getSize(self): + return self.size + + def getDimension(self): + return self.dimension + + def getTimeGrid(self): + return self.mesh + + def computeQuantilePerComponent(self, quantiles): + # Return mock quantile data + results = [] + for q in quantiles: + # Create a sample with the same size as the mesh + quantile_data = np.random.normal(5.0, 1.0, self.mesh.vertices_number) + results.append(Sample(quantile_data.reshape(-1, 1))) + return results + + def __setitem__(self, index, field): + self.fields.append(field) + + +class Normal: + """Mock OpenTURNS Normal distribution.""" + + def __init__(self, mean, std): + self.mean = mean + self.std = std + + def getSample(self, size): + return Sample(np.random.normal(self.mean, self.std, size)) + + +# Mock module functions +def AbsoluteExponential(params1, params2): + """Mock AbsoluteExponential covariance model.""" + return "AbsoluteExponential" + + +def GaussianProcess(cov_model, mesh): + """Mock GaussianProcess.""" + return "GaussianProcess" + + +# Create module-level variables +Sample.BuildFromPoint = lambda points: Sample(points) diff --git a/streamlitApp/tests/test_app_components.py b/streamlitApp/tests/test_app_components.py new file mode 100644 index 0000000..0c10e61 --- /dev/null +++ b/streamlitApp/tests/test_app_components.py @@ -0,0 +1,225 @@ +""" +Unit tests for Streamlit app components. + +These tests verify the functionality of individual components used in the app. +""" + +import os +import sys +import tempfile +from pathlib import Path + +# Add parent directories to path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +import matplotlib +matplotlib.use('Agg') # Use non-interactive backend for testing + +# Mock openturns if not available +try: + import openturns as ot + OPENTURNS_AVAILABLE = True +except ImportError: + OPENTURNS_AVAILABLE = False + # Use mock openturns + sys.path.insert(0, str(Path(__file__).parent)) + import mock_openturns as ot + +from otGpxTrack.Base import GpxTrack + + +def get_test_file_path(filename): + """Get the absolute path to a test file.""" + return os.path.join(os.path.dirname(__file__), '..', '..', 'firstexample', filename) + + +@pytest.fixture +def sample_track(): + """Fixture to load a sample GPX track.""" + gpx_file = get_test_file_path("activity_19218242997.gpx") + return GpxTrack(gpx_file) + + +class TestTrackStatistics: + """Test track statistics calculations.""" + + def test_track_distance(self, sample_track): + """Test distance calculation.""" + distance = sample_track.get_distance() + assert distance > 0 + assert isinstance(distance, float) + + def test_track_duration(self, sample_track): + """Test duration calculation.""" + duration = sample_track.get_duration() + assert duration > 0 + assert isinstance(duration, float) + + def test_track_average_speed(self, sample_track): + """Test average speed calculation.""" + avg_speed = sample_track.get_average_speed() + assert avg_speed >= 0 + assert isinstance(avg_speed, float) + + def test_speed_unit_conversion(self, sample_track): + """Test speed unit conversions.""" + avg_speed_mps = sample_track.get_average_speed() + + # Convert to km/h + avg_speed_kmh = avg_speed_mps * 3.6 + assert avg_speed_kmh > avg_speed_mps + + # Convert to knots + avg_speed_knots = avg_speed_mps * 1.94384 + assert avg_speed_knots > avg_speed_mps + + +class TestBestSegmentAnalysis: + """Test best segment analysis functionality.""" + + def test_best_segment_for_distance(self, sample_track): + """Test finding best segment for a target distance.""" + target_distance = 500.0 + start_idx, end_idx, speed = sample_track.get_best_segment_for_distance(target_distance) + + assert start_idx < end_idx + assert end_idx < len(sample_track.points) + assert speed >= 0 + assert isinstance(start_idx, int) + assert isinstance(end_idx, int) + assert isinstance(speed, float) + + def test_best_segment_for_time(self, sample_track): + """Test finding best segment for a target time.""" + target_time = 10.0 + start_idx, end_idx, speed = sample_track.get_best_segment_for_time(target_time) + + assert start_idx < end_idx + assert end_idx < len(sample_track.points) + assert speed >= 0 + + def test_best_segment_edge_cases(self, sample_track): + """Test best segment with edge cases.""" + # Test with very small target + start_idx, end_idx, speed = sample_track.get_best_segment_for_distance(10.0) + assert start_idx < end_idx + + # Test with larger target (but not too large) + start_idx, end_idx, speed = sample_track.get_best_segment_for_distance(1000.0) + assert start_idx < end_idx + + +class TestSimulationAnalysis: + """Test simulation analysis functionality.""" + + def test_ar1_simulation(self, sample_track): + """Test AR-1 simulation.""" + segment_indices = (0, min(10, len(sample_track.points)-1)) + mean_speed, lower, upper, speeds = sample_track.simulate_ar1_speeds( + segment_indices, + sigma_tot=2.5, + phi=0.9, + n_sims=100 + ) + + assert mean_speed >= 0 + assert lower <= mean_speed <= upper + assert speeds.getSize() == 100 + + def test_process_sample_ar1(self, sample_track): + """Test AR-1 process sample generation.""" + process_sample = sample_track.processSample( + sample_size=100, + method='ar1', + sigma_tot=2.5, + phi=0.9 + ) + + assert process_sample is not None + assert process_sample.getSize() == 100 + assert process_sample.getDimension() == 1 + + def test_process_sample_gaussian(self, sample_track): + """Test Gaussian process sample generation.""" + process_sample = sample_track.processSample( + sample_size=100, + method='gaussian', + amplitude=1.5, + scale=5.0 + ) + + assert process_sample is not None + assert process_sample.getSize() == 100 + assert process_sample.getDimension() == 1 + + def test_process_sample_quantiles(self, sample_track): + """Test quantile calculation from process sample.""" + process_sample = sample_track.processSample( + sample_size=500, + method='ar1', + sigma_tot=2.5, + phi=0.9 + ) + + quantiles = process_sample.computeQuantilePerComponent([0.025, 0.5, 0.975]) + + assert quantiles is not None + assert len(quantiles) == 3 # Three quantile levels + + +class TestDataIntegrity: + """Test data integrity and consistency.""" + + def test_openturns_sample(self, sample_track): + """Test OpenTURNS sample creation.""" + sample = sample_track.get_openturns_sample() + + assert sample is not None + assert sample.getSize() > 0 + assert sample.getDimension() == 5 # lat, lon, elev, time, speed + assert sample.getDescription() == ["Latitude", "Longitude", "Elevation", "Time", "Speed"] + + def test_points_consistency(self, sample_track): + """Test that points and data are consistent.""" + assert len(sample_track.points) > 0 + assert sample_track.data.getSize() == len(sample_track.points) + + def test_track_info_consistency(self, sample_track): + """Test that track information is consistent.""" + distance = sample_track.get_distance() + duration = sample_track.get_duration() + avg_speed = sample_track.get_average_speed() + + # If duration is not zero, avg_speed should be distance/duration + if duration > 0: + expected_avg_speed = distance / duration + assert abs(avg_speed - expected_avg_speed) < 0.001 + + +class TestFileHandling: + """Test file handling and loading.""" + + def test_load_gpx_file(self): + """Test loading GPX file.""" + gpx_file = get_test_file_path("activity_19218242997.gpx") + track = GpxTrack(gpx_file) + + assert track is not None + assert track.gpx is not None + assert len(track.points) > 0 + assert track.data is not None + + def test_load_different_gpx_file(self): + """Test loading a different GPX file.""" + gpx_file = get_test_file_path("activity_22095267736.gpx") + track = GpxTrack(gpx_file) + + assert track is not None + assert len(track.points) > 0 + + def test_file_not_found(self): + """Test handling of non-existent file.""" + with pytest.raises(FileNotFoundError): + GpxTrack("non_existent_file.gpx") diff --git a/streamlitApp/tests/test_pdf_report.py b/streamlitApp/tests/test_pdf_report.py new file mode 100644 index 0000000..d53d7ea --- /dev/null +++ b/streamlitApp/tests/test_pdf_report.py @@ -0,0 +1,275 @@ +""" +Unit tests for PDF report generation. +""" + +import os +import sys +import tempfile +from pathlib import Path + +# Add parent directories to path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +import matplotlib +matplotlib.use('Agg') # Use non-interactive backend for testing + +# Mock openturns if not available +try: + import openturns as ot + OPENTURNS_AVAILABLE = True +except ImportError: + OPENTURNS_AVAILABLE = False + # Use mock openturns + sys.path.insert(0, str(Path(__file__).parent)) + import mock_openturns as ot + +from otGpxTrack.Base import GpxTrack +from streamlitApp.pdf_report import generate_pdf_report + + +def get_test_file_path(filename): + """Get the absolute path to a test file.""" + return os.path.join(os.path.dirname(__file__), '..', '..', 'firstexample', filename) + + +@pytest.fixture +def sample_track(): + """Fixture to load a sample GPX track.""" + gpx_file = get_test_file_path("activity_19218242997.gpx") + return GpxTrack(gpx_file) + + +class TestPDFReport: + """Test class for PDF report generation.""" + + def test_generate_pdf_basic(self, sample_track): + """Test basic PDF generation.""" + report_data = { + "track": sample_track, + "speed_unit": "knots", + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = ["Track Statistics"] + + pdf_bytes = generate_pdf_report(report_data, report_options, "test_report") + + assert pdf_bytes is not None + assert isinstance(pdf_bytes, bytes) + assert len(pdf_bytes) > 0 + # Check if it starts with PDF magic number + assert pdf_bytes[:4] == b'%PDF' or pdf_bytes[:4] == b'\x89PNG' + + def test_generate_pdf_with_all_options(self, sample_track): + """Test PDF generation with all options enabled.""" + # Generate process sample for simulation + process_sample = sample_track.processSample( + sample_size=100, method='ar1', sigma_tot=2.5, phi=0.9 + ) + + report_data = { + "track": sample_track, + "speed_unit": "knots", + "show_simulation": True, + "process_sample": process_sample, + "show_best_segment": True, + "target_type": "Distance", + "target_value": 500.0, + "start_idx": 0, + "end_idx": 10, + "best_speed": 5.0, + "method": "ar1", + "sample_size": 100, + "sigma_tot": 2.5, + "phi": 0.9, + } + + report_options = [ + "Track Statistics", + "Track Map", + "Speed Over Time", + "Elevation Profile", + "Speed Distribution", + "Simulation Analysis", + "Best Segment Analysis", + ] + + pdf_bytes = generate_pdf_report(report_data, report_options, "full_report") + + assert pdf_bytes is not None + assert isinstance(pdf_bytes, bytes) + assert len(pdf_bytes) > 0 + + def test_generate_pdf_with_visualizations(self, sample_track): + """Test PDF generation with visualizations.""" + report_data = { + "track": sample_track, + "speed_unit": "km/h", + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = [ + "Track Statistics", + "Track Map", + "Speed Over Time", + "Elevation Profile", + "Speed Distribution", + ] + + pdf_bytes = generate_pdf_report(report_data, report_options, "viz_report") + + assert pdf_bytes is not None + assert len(pdf_bytes) > 0 + + def test_generate_pdf_different_speed_units(self, sample_track): + """Test PDF generation with different speed units.""" + for unit in ["m/s", "km/h", "knots"]: + report_data = { + "track": sample_track, + "speed_unit": unit, + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = ["Track Statistics"] + + pdf_bytes = generate_pdf_report(report_data, report_options, f"report_{unit}") + + assert pdf_bytes is not None + assert len(pdf_bytes) > 0 + + def test_generate_pdf_with_gaussian_simulation(self, sample_track): + """Test PDF generation with Gaussian process simulation.""" + process_sample = sample_track.processSample( + sample_size=50, method='gaussian', amplitude=1.5, scale=5.0 + ) + + report_data = { + "track": sample_track, + "speed_unit": "knots", + "show_simulation": True, + "process_sample": process_sample, + "show_best_segment": False, + "method": "gaussian", + "sample_size": 50, + "amplitude": 1.5, + "scale": 5.0, + } + + report_options = ["Track Statistics", "Simulation Analysis"] + + pdf_bytes = generate_pdf_report(report_data, report_options, "gaussian_report") + + assert pdf_bytes is not None + assert len(pdf_bytes) > 0 + + def test_generate_pdf_minimal_options(self, sample_track): + """Test PDF generation with minimal options.""" + report_data = { + "track": sample_track, + "speed_unit": "knots", + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = ["Track Statistics"] + + pdf_bytes = generate_pdf_report(report_data, report_options, "minimal_report") + + assert pdf_bytes is not None + assert len(pdf_bytes) > 0 + + def test_generate_pdf_empty_options(self, sample_track): + """Test PDF generation with empty options list.""" + report_data = { + "track": sample_track, + "speed_unit": "knots", + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = [] + + pdf_bytes = generate_pdf_report(report_data, report_options, "empty_report") + + # Should still generate a PDF with title page + assert pdf_bytes is not None + assert len(pdf_bytes) > 0 + + def test_pdf_size_reasonable(self, sample_track): + """Test that generated PDF has reasonable size.""" + report_data = { + "track": sample_track, + "speed_unit": "knots", + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = ["Track Statistics", "Track Map"] + + pdf_bytes = generate_pdf_report(report_data, report_options, "size_test") + + # PDF should be at least a few KB but not more than a few MB + assert len(pdf_bytes) > 1000 # At least 1KB + assert len(pdf_bytes) < 10_000_000 # Less than 10MB + + +class TestPDFReportEdgeCases: + """Test edge cases for PDF report generation.""" + + def test_pdf_with_none_report_data(self): + """Test PDF generation with None report data.""" + # This should handle gracefully or raise appropriate error + try: + pdf_bytes = generate_pdf_report(None, [], "test") + # If it doesn't raise an error, check that we got something + assert pdf_bytes is not None + except Exception as e: + # It's acceptable to raise an error for invalid input + assert True + + def test_pdf_with_invalid_track(self): + """Test PDF generation with invalid track object.""" + class MockTrack: + def __init__(self): + self.points = [] + self.data = None + + def get_distance(self): + return 0.0 + + def get_duration(self): + return 0.0 + + def get_average_speed(self): + return 0.0 + + mock_track = MockTrack() + + report_data = { + "track": mock_track, + "speed_unit": "knots", + "show_simulation": False, + "process_sample": None, + "show_best_segment": False, + } + + report_options = ["Track Statistics"] + + # This might work or raise an error, both are acceptable + try: + pdf_bytes = generate_pdf_report(report_data, report_options, "mock_test") + assert pdf_bytes is not None + except Exception: + # It's acceptable to raise an error for empty tracks + assert True diff --git a/streamlitApp/tests/test_visualization.py b/streamlitApp/tests/test_visualization.py new file mode 100644 index 0000000..89bf84a --- /dev/null +++ b/streamlitApp/tests/test_visualization.py @@ -0,0 +1,197 @@ +""" +Unit tests for visualization functions. +""" + +import os +import sys +import tempfile +from pathlib import Path + +# Add parent directories to path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import pytest +import matplotlib.pyplot as plt + +# Mock openturns if not available +try: + import openturns as ot + OPENTURNS_AVAILABLE = True +except ImportError: + OPENTURNS_AVAILABLE = False + # Use mock openturns + sys.path.insert(0, str(Path(__file__).parent)) + import mock_openturns as ot + +from otGpxTrack.Base import GpxTrack +from streamlitApp.visualization import ( + plot_track_with_speed, + plot_speed_over_time, + plot_elevation_profile, + plot_speed_distribution, + create_comparison_plot, +) + + +def get_test_file_path(filename): + """Get the absolute path to a test file.""" + return os.path.join(os.path.dirname(__file__), '..', '..', 'firstexample', filename) + + +@pytest.fixture +def sample_track(): + """Fixture to load a sample GPX track.""" + gpx_file = get_test_file_path("activity_19218242997.gpx") + return GpxTrack(gpx_file) + + +class TestVisualization: + """Test class for visualization functions.""" + + def test_plot_track_with_speed(self, sample_track): + """Test track plot with speed coloring.""" + fig = plot_track_with_speed(sample_track, speed_unit="knots") + + assert fig is not None + assert isinstance(fig, plt.Figure) + assert len(fig.axes) >= 1 + + # Clean up + plt.close(fig) + + def test_plot_track_with_speed_different_units(self, sample_track): + """Test track plot with different speed units.""" + for unit in ["m/s", "km/h", "knots"]: + fig = plot_track_with_speed(sample_track, speed_unit=unit) + assert fig is not None + plt.close(fig) + + def test_plot_track_with_speed_custom_figsize(self, sample_track): + """Test track plot with custom figure size.""" + fig = plot_track_with_speed(sample_track, figsize=(10, 6)) + assert fig is not None + plt.close(fig) + + def test_plot_speed_over_time(self, sample_track): + """Test speed over time plot.""" + fig = plot_speed_over_time(sample_track, speed_unit="knots") + + assert fig is not None + assert isinstance(fig, plt.Figure) + assert len(fig.axes) >= 1 + + plt.close(fig) + + def test_plot_speed_over_time_with_process_sample(self, sample_track): + """Test speed over time plot with process sample.""" + # Generate a process sample + process_sample = sample_track.processSample( + sample_size=100, method='ar1', sigma_tot=2.5, phi=0.9 + ) + + fig = plot_speed_over_time( + sample_track, + speed_unit="knots", + process_sample=process_sample + ) + + assert fig is not None + plt.close(fig) + + def test_plot_elevation_profile(self, sample_track): + """Test elevation profile plot.""" + fig = plot_elevation_profile(sample_track) + + assert fig is not None + assert isinstance(fig, plt.Figure) + assert len(fig.axes) >= 1 + + plt.close(fig) + + def test_plot_elevation_profile_custom_figsize(self, sample_track): + """Test elevation profile plot with custom figure size.""" + fig = plot_elevation_profile(sample_track, figsize=(14, 8)) + assert fig is not None + plt.close(fig) + + def test_plot_speed_distribution(self, sample_track): + """Test speed distribution histogram.""" + fig = plot_speed_distribution(sample_track, speed_unit="knots") + + assert fig is not None + assert isinstance(fig, plt.Figure) + assert len(fig.axes) >= 1 + + plt.close(fig) + + def test_plot_speed_distribution_different_units(self, sample_track): + """Test speed distribution with different speed units.""" + for unit in ["m/s", "km/h", "knots"]: + fig = plot_speed_distribution(sample_track, speed_unit=unit) + assert fig is not None + plt.close(fig) + + def test_plot_speed_distribution_custom_bins(self, sample_track): + """Test speed distribution with custom number of bins.""" + fig = plot_speed_distribution(sample_track, bins=20) + assert fig is not None + plt.close(fig) + + def test_create_comparison_plot(self, sample_track): + """Test comparison plot with multiple subplots.""" + fig = create_comparison_plot(sample_track) + + assert fig is not None + assert isinstance(fig, plt.Figure) + # Should have 4 subplots (2x2 grid) + assert len(fig.axes) >= 4 + + plt.close(fig) + + def test_visualization_error_handling(self): + """Test error handling for empty tracks.""" + # Create a mock track with no points + class MockTrack: + def __init__(self): + self.points = [] + self.data = None + + mock_track = MockTrack() + + # These should raise ValueError + with pytest.raises(ValueError): + plot_track_with_speed(mock_track) + + with pytest.raises(ValueError): + plot_speed_over_time(mock_track) + + with pytest.raises(ValueError): + plot_elevation_profile(mock_track) + + with pytest.raises(ValueError): + plot_speed_distribution(mock_track) + + +class TestVisualizationEdgeCases: + """Test edge cases for visualization functions.""" + + def test_single_point_track(self): + """Test visualization with a track that has only one point.""" + # This is a bit tricky to test without creating a real GPX file + # For now, we'll just test that the functions handle it gracefully + pass + + def test_zero_speed_points(self, sample_track): + """Test that zero speeds are handled correctly.""" + # The visualization functions should handle zero speeds + fig = plot_speed_distribution(sample_track) + assert fig is not None + plt.close(fig) + + def test_missing_elevation_data(self, sample_track): + """Test elevation profile with potentially missing elevation data.""" + # The elevation profile function should handle missing elevation + fig = plot_elevation_profile(sample_track) + assert fig is not None + plt.close(fig) diff --git a/streamlitApp/visualization.py b/streamlitApp/visualization.py new file mode 100644 index 0000000..1cefe94 --- /dev/null +++ b/streamlitApp/visualization.py @@ -0,0 +1,364 @@ +""" +Visualization functions for GPX track analysis. + +This module provides functions to create various plots for GPX track data. +""" + +import numpy as np +import matplotlib.pyplot as plt +from matplotlib.colors import Normalize +from matplotlib import cm + + +def plot_track_with_speed(track, figsize=(12, 8), speed_unit="knots", dpi=100): + """ + Plot the GPX track with points colored by speed. + + Args: + track: GpxTrack instance + figsize: Figure size (width, height) + speed_unit: Speed unit ('m/s', 'km/h', 'knots') + dpi: DPI for the figure + + Returns: + matplotlib.figure.Figure: The created figure + """ + if not track.points: + raise ValueError("No points to plot.") + + # Extract coordinates and speeds + longitudes = [point.longitude for point in track.points] + latitudes = [point.latitude for point in track.points] + + # Get speeds from OpenTURNS sample + if track.data is not None and track.data.getDimension() >= 5: + speeds = [track.data[i][4] for i in range(track.data.getSize())] + else: + speeds = [0.0] * len(track.points) + + # Convert speeds to desired unit + if speed_unit == "km/h": + speeds = [s * 3.6 for s in speeds] + unit_label = "Speed (km/h)" + elif speed_unit == "knots": + speeds = [s * 1.94384 for s in speeds] + unit_label = "Speed (knots)" + else: # m/s + unit_label = "Speed (m/s)" + + # Create the plot + fig, ax = plt.subplots(figsize=figsize, dpi=dpi) + + # Plot points colored by speed + scatter = ax.scatter( + longitudes, + latitudes, + c=speeds, + cmap="jet", + s=20, + alpha=0.8, + edgecolors="none" + ) + + # Add colorbar + cbar = plt.colorbar(scatter, ax=ax, shrink=0.8) + cbar.set_label(unit_label, fontsize=10) + + # Set plot properties + ax.set_title("GPX Track with Speed Coloring", fontsize=14, fontweight="bold") + ax.set_xlabel("Longitude", fontsize=12) + ax.set_ylabel("Latitude", fontsize=12) + ax.grid(True, alpha=0.3) + ax.set_aspect("equal") + + # Adjust layout + plt.tight_layout() + + return fig + + +def plot_speed_over_time(track, figsize=(14, 6), speed_unit="knots", + process_sample=None, dpi=100): + """ + Plot speed over time with optional confidence intervals. + + Args: + track: GpxTrack instance + figsize: Figure size (width, height) + speed_unit: Speed unit ('m/s', 'km/h', 'knots') + process_sample: Optional ProcessSample for confidence intervals + dpi: DPI for the figure + + Returns: + matplotlib.figure.Figure: The created figure + """ + if not track.points: + raise ValueError("No points to plot.") + + # Extract time values and speeds + time_values = [] + observed_speeds = [] + + for i, point in enumerate(track.points): + time_value = point.time.timestamp() if point.time is not None else 0.0 + time_values.append(time_value) + if track.data is not None and track.data.getDimension() >= 5: + observed_speeds.append(track.data[i][4]) + else: + observed_speeds.append(0.0) + + # Convert to relative time (seconds since start) + if time_values: + start_time = time_values[0] + time_values = [t - start_time for t in time_values] + + # Convert speeds to desired unit + if speed_unit == "km/h": + observed_speeds = [s * 3.6 for s in observed_speeds] + unit_label = "Speed (km/h)" + elif speed_unit == "knots": + observed_speeds = [s * 1.94384 for s in observed_speeds] + unit_label = "Speed (knots)" + else: # m/s + unit_label = "Speed (m/s)" + + # Create the plot + fig, ax = plt.subplots(figsize=figsize, dpi=dpi) + + # Plot observed speeds + ax.plot(time_values, observed_speeds, "b-", label="Observed Speed", + linewidth=2, alpha=0.8) + + # Add confidence intervals if process_sample is provided + if process_sample is not None: + try: + # Calculate quantiles + quantiles = process_sample.computeQuantilePerComponent([0.025, 0.975]) + + # Extract quantile values + n_points = len(track.points) + quantile_025 = [] + quantile_975 = [] + + for i in range(n_points): + q025 = quantiles[0][i][0] + q975 = quantiles[1][i][0] + + # Convert to desired unit + if speed_unit == "km/h": + q025 *= 3.6 + q975 *= 3.6 + elif speed_unit == "knots": + q025 *= 1.94384 + q975 *= 1.94384 + + quantile_025.append(q025) + quantile_975.append(q975) + + # Plot confidence interval + ax.fill_between( + time_values, + quantile_025, + quantile_975, + alpha=0.3, + color="blue", + label="95% Confidence Interval", + ) + except Exception as e: + print(f"Error calculating confidence intervals: {e}") + + # Customize the plot + ax.set_title("Speed Over Time", fontsize=14, fontweight="bold") + ax.set_xlabel("Time (seconds)", fontsize=12) + ax.set_ylabel(unit_label, fontsize=12) + ax.grid(True, alpha=0.3) + ax.legend(fontsize=10) + + # Adjust layout + plt.tight_layout() + + return fig + + +def plot_elevation_profile(track, figsize=(14, 6), dpi=100): + """ + Plot elevation profile over distance. + + Args: + track: GpxTrack instance + figsize: Figure size (width, height) + dpi: DPI for the figure + + Returns: + matplotlib.figure.Figure: The created figure + """ + if not track.points: + raise ValueError("No points to plot.") + + # Calculate cumulative distance + cumulative_distances = [0.0] + for i in range(1, len(track.points)): + distance = track.points[i].distance_3d(track.points[i - 1]) + cumulative_distances.append(cumulative_distances[i - 1] + distance) + + # Extract elevations + elevations = [] + for point in track.points: + elevation = point.elevation if point.elevation is not None else 0.0 + elevations.append(elevation) + + # Create the plot + fig, ax = plt.subplots(figsize=figsize, dpi=dpi) + + # Plot elevation profile + ax.plot(cumulative_distances, elevations, "g-", + linewidth=2, label="Elevation") + + # Customize the plot + ax.set_title("Elevation Profile", fontsize=14, fontweight="bold") + ax.set_xlabel("Distance (meters)", fontsize=12) + ax.set_ylabel("Elevation (meters)", fontsize=12) + ax.grid(True, alpha=0.3) + ax.legend(fontsize=10) + + # Adjust layout + plt.tight_layout() + + return fig + + +def plot_speed_distribution(track, figsize=(12, 6), speed_unit="knots", + bins=30, dpi=100): + """ + Plot histogram of speed distribution. + + Args: + track: GpxTrack instance + figsize: Figure size (width, height) + speed_unit: Speed unit ('m/s', 'km/h', 'knots') + bins: Number of bins for histogram + dpi: DPI for the figure + + Returns: + matplotlib.figure.Figure: The created figure + """ + if not track.points: + raise ValueError("No points to plot.") + + # Get speeds from OpenTURNS sample + if track.data is not None and track.data.getDimension() >= 5: + speeds = [track.data[i][4] for i in range(track.data.getSize())] + else: + speeds = [0.0] * len(track.points) + + # Convert speeds to desired unit + if speed_unit == "km/h": + speeds = [s * 3.6 for s in speeds] + unit_label = "Speed (km/h)" + elif speed_unit == "knots": + speeds = [s * 1.94384 for s in speeds] + unit_label = "Speed (knots)" + else: # m/s + unit_label = "Speed (m/s)" + + # Filter out zero speeds (if any) + speeds = [s for s in speeds if s > 0] + + if not speeds: + speeds = [0.0] + + # Create the plot + fig, ax = plt.subplots(figsize=figsize, dpi=dpi) + + # Plot histogram + n, bins, patches = ax.hist(speeds, bins=bins, color="#1f77b4", + edgecolor="black", alpha=0.7) + + # Customize the plot + ax.set_title("Speed Distribution", fontsize=14, fontweight="bold") + ax.set_xlabel(unit_label, fontsize=12) + ax.set_ylabel("Frequency", fontsize=12) + ax.grid(True, alpha=0.3) + + # Add mean line + mean_speed = np.mean(speeds) + ax.axvline(mean_speed, color="red", linestyle="--", + linewidth=2, label=f"Mean: {mean_speed:.2f}") + ax.legend(fontsize=10) + + # Adjust layout + plt.tight_layout() + + return fig + + +def create_comparison_plot(track, figsize=(14, 8), dpi=100): + """ + Create a comparison plot with multiple subplots. + + Args: + track: GpxTrack instance + figsize: Figure size (width, height) + dpi: DPI for the figure + + Returns: + matplotlib.figure.Figure: The created figure + """ + if not track.points: + raise ValueError("No points to plot.") + + # Create subplots + fig, axes = plt.subplots(2, 2, figsize=figsize, dpi=dpi) + + # Plot 1: Track Map + longitudes = [point.longitude for point in track.points] + latitudes = [point.latitude for point in track.points] + speeds = [track.data[i][4] * 1.94384 for i in range(track.data.getSize())] # knots + + axes[0, 0].scatter(longitudes, latitudes, c=speeds, cmap="jet", s=10, alpha=0.7) + axes[0, 0].set_title("Track Map") + axes[0, 0].set_xlabel("Longitude") + axes[0, 0].set_ylabel("Latitude") + axes[0, 0].grid(True, alpha=0.3) + axes[0, 0].set_aspect("equal") + + # Plot 2: Speed Over Time + time_values = [(p.time.timestamp() - track.points[0].time.timestamp()) + for p in track.points] + speeds_mps = [track.data[i][4] for i in range(track.data.getSize())] + + axes[0, 1].plot(time_values, speeds_mps, "b-", linewidth=1.5) + axes[0, 1].set_title("Speed Over Time") + axes[0, 1].set_xlabel("Time (s)") + axes[0, 1].set_ylabel("Speed (m/s)") + axes[0, 1].grid(True, alpha=0.3) + + # Plot 3: Elevation Profile + cumulative_distances = [0.0] + for i in range(1, len(track.points)): + distance = track.points[i].distance_3d(track.points[i - 1]) + cumulative_distances.append(cumulative_distances[i - 1] + distance) + + elevations = [p.elevation if p.elevation is not None else 0.0 + for p in track.points] + + axes[1, 0].plot(cumulative_distances, elevations, "g-", linewidth=1.5) + axes[1, 0].set_title("Elevation Profile") + axes[1, 0].set_xlabel("Distance (m)") + axes[1, 0].set_ylabel("Elevation (m)") + axes[1, 0].grid(True, alpha=0.3) + + # Plot 4: Speed Distribution + speeds_nonzero = [s for s in speeds_mps if s > 0] + if speeds_nonzero: + axes[1, 1].hist(speeds_nonzero, bins=30, color="#1f77b4", + edgecolor="black", alpha=0.7) + axes[1, 1].set_title("Speed Distribution") + axes[1, 1].set_xlabel("Speed (m/s)") + axes[1, 1].set_ylabel("Frequency") + axes[1, 1].grid(True, alpha=0.3) + + # Adjust layout + plt.tight_layout() + + return fig