Inertial Tracker — Реальное время

import sqlite3 import json import os from http.server import HTTPServer, BaseHTTPRequestHandler DB_PATH = os.path.expanduser('~/inertial_control/inertial_data.db') PORT = 8050 TRAJ_LIMIT = 5000 # показываем последние 5000 точек траектории (~60 сек движения) LIDAR_LIMIT = 2000 # последние точки лидара HTML = """ Inertial Tracker — Реальное время

X: —
Y: —
Z: —

◆ Начальная точка (0, 0, 0)
━━ Траектория движения
●●● Карта окружения (лидар)
● Текущее положение

Оси X, Y, Z — перемещение в метрах.
Масштаб одинаковый по всем осям.
Вращайте сцену мышью.

Подключение...

""" def fetch_data(): """Читает свежие данные из БД в read-only режиме (не мешает писателям).""" conn = sqlite3.connect(f'file:{DB_PATH}?mode=ro', uri=True) try: # Текущая позиция cur = conn.execute( "SELECT x, y, z FROM trajectory ORDER BY timestamp DESC LIMIT 1" ) row = cur.fetchone() pos = {'x': row[0], 'y': row[1], 'z': row[2]} if row else {'x': 0.0, 'y': 0.0, 'z': 0.0} # Последние N точек траектории (разворачиваем в хронологический порядок) cur = conn.execute( f"SELECT x, y, z FROM trajectory ORDER BY timestamp DESC LIMIT {TRAJ_LIMIT}" ) traj_rows = cur.fetchall()[::-1] traj = { 'x': [r[0] for r in traj_rows], 'y': [r[1] for r in traj_rows], 'z': [r[2] for r in traj_rows], } # Последние N точек лидара cur = conn.execute( f"SELECT x, y, z FROM lidar_points ORDER BY rowid DESC LIMIT {LIDAR_LIMIT}" ) lidar_rows = cur.fetchall() lidar = { 'x': [r[0] for r in lidar_rows], 'y': [r[1] for r in lidar_rows], 'z': [r[2] for r in lidar_rows], } finally: conn.close() return {'pos': pos, 'traj': traj, 'lidar': lidar} class Handler(BaseHTTPRequestHandler): def log_message(self, format, *args): pass def do_GET(self): if self.path == '/': self.send_response(200) self.send_header('Content-type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(HTML.encode('utf-8')) elif self.path == '/plotly-latest.min.js': self.send_response(200) self.send_header('Content-type', 'application/javascript') self.end_headers() with open('plotly-latest.min.js', 'rb') as f: self.wfile.write(f.read()) elif self.path.startswith('/api/data'): try: data = fetch_data() payload = json.dumps(data).encode('utf-8') self.send_response(200) self.send_header('Content-type', 'application/json') self.send_header('Cache-Control', 'no-store') self.end_headers() self.wfile.write(payload) except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(str(e).encode('utf-8')) elif self.path.startswith('/api/pos'): try: conn = sqlite3.connect(f'file:{DB_PATH}?mode=ro', uri=True) cur = conn.execute( "SELECT x, y, z FROM trajectory ORDER BY timestamp DESC LIMIT 1" ) row = cur.fetchone() conn.close() pos = {'x': row[0], 'y': row[1], 'z': row[2]} if row else {'x': 0, 'y': 0, 'z': 0} self.send_response(200) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps(pos).encode('utf-8')) except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(str(e).encode('utf-8')) else: self.send_response(404) self.end_headers() if __name__ == '__main__': print("=" * 50) print("INERTIAL TRACKER — реальное время") print(f"Открыть: http://192.168.0.106:{PORT}") print("=" * 50) HTTPServer(('0.0.0.0', PORT), Handler).serve_forever()