| import streamlit as st |
| import pandas as pd |
| from sqlalchemy import create_engine, text |
| from sshtunnel import SSHTunnelForwarder |
| import os |
| import tempfile |
| from datetime import datetime, timedelta |
|
|
| |
| st.set_page_config(page_title="GPS Viewer", layout="wide") |
| st.title("🚀 GPS Track on Google Maps") |
|
|
| |
| SSH_KEY = st.secrets["SSH_KEY"] |
| SSH_USER = st.secrets["SSH_USER"] |
| SSH_HOST = st.secrets["DB_HOST"] |
| REMOTE_BIND_HOST = st.secrets["REMOTE_BIND_HOST"] |
| LOCAL_BIND_HOST = st.secrets["LOCAL_BIND_HOST"] |
| MYSQL_USER = st.secrets["MYSQL_USER"] |
| MYSQL_PASSWORD = st.secrets["MYSQL_PASSWORD"] |
| MYSQL_DB = st.secrets["DB_NAME"] |
| MYSQL_PORT = int(st.secrets["MYSQL_PORT"]) |
| GOOGLE_MAPS_API_KEY = st.secrets["GOOGLE_MAPS_API_KEY"] |
|
|
| |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".pem") as temp_file: |
| temp_file.write(SSH_KEY.encode()) |
| temp_file_path = temp_file.name |
| os.chmod(temp_file_path, 0o600) |
|
|
| |
| @st.cache_resource |
| def start_ssh_tunnel(): |
| tunnel = SSHTunnelForwarder( |
| (SSH_HOST, 22), |
| ssh_username=SSH_USER, |
| ssh_pkey=temp_file_path, |
| remote_bind_address=(REMOTE_BIND_HOST, MYSQL_PORT), |
| local_bind_address=(LOCAL_BIND_HOST, MYSQL_PORT) |
| ) |
| tunnel.start() |
| return tunnel |
|
|
| server = start_ssh_tunnel() |
| if not server.is_active: |
| server.start() |
|
|
| |
| MYSQL_URL = f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{LOCAL_BIND_HOST}:{MYSQL_PORT}/{MYSQL_DB}" |
| engine = create_engine(MYSQL_URL, pool_pre_ping=True, pool_recycle=3600) |
|
|
| |
| def get_timestamp_range(): |
| try: |
| with engine.connect() as conn: |
| result = conn.execute(text("SELECT MIN(timestamp) as min_ts, MAX(timestamp) as max_ts FROM telemetry_data")).fetchone() |
| return result[0], result[1] |
| except Exception as e: |
| st.error(f"Error fetching timestamp range: {e}") |
| return None, None |
|
|
| min_ts, max_ts = get_timestamp_range() |
|
|
| |
| st.sidebar.header("Filter by Time Frame") |
| if min_ts and max_ts: |
| start_datetime = st.sidebar.slider("Start and End Time", min_value=min_ts, max_value=max_ts, |
| value=(min_ts, max_ts), format="YYYY-MM-DD HH:mm:ss") |
| start_ts, end_ts = start_datetime |
| else: |
| start_ts = end_ts = None |
|
|
| |
| def load_gps_data(from_ts, to_ts): |
| try: |
| query = text(""" |
| SELECT latitude, longitude, device_id |
| FROM telemetry_data |
| WHERE timestamp BETWEEN :from_ts AND :to_ts |
| ORDER BY timestamp |
| """) |
| with engine.connect() as conn: |
| df = pd.read_sql(query, conn, params={"from_ts": from_ts, "to_ts": to_ts}) |
| return df |
| except Exception as e: |
| st.error(f"Error loading GPS data: {e}") |
| return pd.DataFrame() |
|
|
| gps_data = pd.DataFrame() |
| if start_ts and end_ts: |
| gps_data = load_gps_data(start_ts, end_ts) |
|
|
| |
| if not gps_data.empty: |
| points = gps_data[["latitude", "longitude"]].to_dict(orient="records") |
| js_points = ",\n".join([f"{{ lat: {p['latitude']}, lng: {p['longitude']} }}" for p in points]) |
|
|
| st.components.v1.html(f""" |
| <div id=\"map\" style=\"height: 80vh; width: 100%;\"></div> |
| <div style=\"text-align: center; margin: 10px;\"> |
| <button onclick=\"startReplay()\">▶️ Start</button> |
| <button onclick=\"pauseReplay()\">⏸ Pause</button> |
| <button onclick=\"resetReplay()\">🔄 Reset</button> |
| <button onclick=\"stepBack()\">⏪ Back</button> |
| <button onclick=\"stepForward()\">⏩ Forward</button> |
| </div> |
| |
| <script src=\"https://maps.googleapis.com/maps/api/js?key={GOOGLE_MAPS_API_KEY}\"></script> |
| <script> |
| const pathCoords = [{js_points}]; |
| let map, marker, interval, index = 0, playing = false; |
| |
| function initMap() {{ |
| map = new google.maps.Map(document.getElementById(\"map\"), {{ |
| zoom: 14, |
| center: pathCoords[0], |
| mapTypeId: \"roadmap\" |
| }}); |
| |
| new google.maps.Polyline({{ |
| path: pathCoords, |
| geodesic: true, |
| strokeColor: \"#FF0000\", |
| strokeOpacity: 1.0, |
| strokeWeight: 2 |
| }}).setMap(map); |
| |
| marker = new google.maps.Marker({{ |
| position: pathCoords[0], |
| map: map, |
| icon: {{ |
| path: google.maps.SymbolPath.CIRCLE, |
| scale: 6, |
| fillColor: \"#0000FF\", |
| fillOpacity: 1.0, |
| strokeWeight: 1 |
| }} |
| }}); |
| }} |
| |
| function updateMarker() {{ |
| marker.setPosition(pathCoords[index]); |
| map.panTo(pathCoords[index]); |
| }} |
| |
| function playStep() {{ |
| if (index < pathCoords.length - 1) {{ |
| index++; |
| updateMarker(); |
| }} else {{ |
| clearInterval(interval); |
| playing = false; |
| }} |
| }} |
| |
| function startReplay() {{ |
| if (!playing) {{ |
| interval = setInterval(playStep, 200); |
| playing = true; |
| }} |
| }} |
| |
| function pauseReplay() {{ |
| clearInterval(interval); |
| playing = false; |
| }} |
| |
| function resetReplay() {{ |
| pauseReplay(); |
| index = 0; |
| updateMarker(); |
| }} |
| |
| function stepBack() {{ |
| pauseReplay(); |
| if (index > 0) {{ |
| index--; |
| updateMarker(); |
| }} |
| }} |
| |
| function stepForward() {{ |
| pauseReplay(); |
| if (index < pathCoords.length - 1) {{ |
| index++; |
| updateMarker(); |
| }} |
| }} |
| |
| window.onload = initMap; |
| </script> |
| """, height=750) |
|
|
| st.subheader("📊 Telemetry Data") |
| st.dataframe(gps_data[["device_id"]]) |
| else: |
| st.info("No GPS data available for selected time frame.") |
