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 LIDAR_LIMIT = 2000 HTML = """ Inertial Tracker — Реальное время

X: —
Y: —
Z: —

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

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

""" def fetch_data(): conn = sqlite3.connect(f'file:{DB_PATH}?mode=ro', uri=True) try: cur = conn.execute( "SELECT x, y, z, roll, pitch, yaw FROM trajectory ORDER BY timestamp DESC LIMIT 1" ) row = cur.fetchone() if row: pos = {'x': row[0], 'y': row[1], 'z': row[2]} orient = {'roll': row[3] or 0.0, 'pitch': row[4] or 0.0, 'yaw': row[5] or 0.0} else: pos = {'x': 0.0, 'y': 0.0, 'z': 0.0} orient = {'roll': 0.0, 'pitch': 0.0, 'yaw': 0.0} cur = conn.execute( f"SELECT x, y, z FROM trajectory ORDER BY timestamp ASC 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], } 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, 'orient': orient, '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()