host.camera_visualizer.serve
Camera visualizer HTTP server.
Thin routing layer — business logic lives in routes/ and field_cache.py.
Usage: python -m host.camera_visualizer.serve python host/camera_visualizer/serve.py
1""" 2Camera visualizer HTTP server. 3 4Thin routing layer — business logic lives in routes/ and field_cache.py. 5 6Usage: 7 python -m host.camera_visualizer.serve 8 python host/camera_visualizer/serve.py 9""" 10 11import http.server 12import json 13import os 14import signal 15import sys 16import threading 17import webbrowser 18 19# Resolve project root (two levels up from this file) 20HERE = os.path.dirname(os.path.abspath(__file__)) 21PROJECT_ROOT = os.path.abspath(os.path.join(HERE, '..', '..')) 22sys.path.insert(0, PROJECT_ROOT) 23 24from host.camera_visualizer.field_cache import FieldCache # noqa: E402 25from host.camera_visualizer.routes import ( # noqa: E402 26 cad, config, fields, points, 27) 28 29PORT = 8070 30CAD_DIR = os.path.join(PROJECT_ROOT, 'host', 'cad_tools', 'models') 31CACHE_DIR = os.path.join(PROJECT_ROOT, 'cache', 'fields') 32BUNDLE_CACHE = os.path.join(PROJECT_ROOT, 'cache', 'frc_assets.zip') 33POINTS_DIR = os.path.join(PROJECT_ROOT, 'data', 'field_points') 34 35field_cache = FieldCache(CACHE_DIR, BUNDLE_CACHE) 36 37 38class Handler(http.server.BaseHTTPRequestHandler): 39 40 def do_GET(self): 41 from urllib.parse import urlparse 42 path = urlparse(self.path).path 43 44 # ── Page ────────────────────────────────────────────────── 45 if path in ('/', '/index.html'): 46 page = config.build_page(HERE) 47 self._respond(200, 'text/html; charset=utf-8', page, 48 cache='no-cache, no-store') 49 50 # ── API: robot config ───────────────────────────────────── 51 elif path == '/api/config': 52 self._respond_json(config.build_config_json()) 53 54 # ── API: field list ─────────────────────────────────────── 55 elif path == '/api/field-list': 56 self._respond_json(json.dumps(field_cache.get_field_list())) 57 58 # ── API: field tags (from robotpy_apriltag) ─────────────── 59 elif path == '/api/field-tags': 60 self._respond_json(json.dumps(fields.get_field_tags())) 61 62 # ── API: field GLB model ────────────────────────────────── 63 elif path.startswith('/api/field/') and path.endswith('.glb'): 64 field_id = path[len('/api/field/'):-4] 65 _, glb = field_cache.extract_field(field_id) 66 if glb is None: 67 self.send_error(404, f'Field {field_id} not found') 68 return 69 self._respond(200, 'model/gltf-binary', glb, 70 cache='public, max-age=86400') 71 72 # ── API: field config JSON ──────────────────────────────── 73 elif path.startswith('/api/field/') and path.endswith('.json'): 74 field_id = path[len('/api/field/'):-5] 75 cfg, _ = field_cache.extract_field(field_id) 76 if cfg is None: 77 self.send_error(404, f'Field {field_id} config not found') 78 return 79 self._respond_json(json.dumps(cfg)) 80 81 # ── API: saved points ───────────────────────────────────── 82 elif path == '/api/points/list': 83 self._respond_json(json.dumps(points.list_point_sets(POINTS_DIR))) 84 85 elif path == '/api/points': 86 set_name = self._query_param('set', 'default') 87 self._respond_json(points.load_points(POINTS_DIR, set_name)) 88 89 # ── API: CAD model list ─────────────────────────────────── 90 elif path == '/api/cad-models': 91 self._respond_json(json.dumps(cad.list_cad_models(CAD_DIR))) 92 93 # ── Static: CAD files ───────────────────────────────────── 94 elif path.startswith('/cad/'): 95 data, mime = cad.serve_cad_file(CAD_DIR, path[5:]) 96 if data is None: 97 self.send_error(404) 98 return 99 self._respond(200, mime, data, cache='public, max-age=3600') 100 101 # ── Static: JS modules ──────────────────────────────────── 102 elif path.endswith('.js') and '/' not in path[1:]: 103 fpath = os.path.join(HERE, path[1:]) 104 if not os.path.isfile(fpath): 105 self.send_error(404) 106 return 107 with open(fpath, 'rb') as f: 108 data = f.read() 109 self._respond(200, 'application/javascript', data, 110 cache='no-cache, no-store') 111 112 else: 113 self.send_error(404) 114 115 def do_POST(self): 116 from urllib.parse import urlparse 117 post_path = urlparse(self.path).path 118 if post_path == '/api/points': 119 set_name = self._query_param('set', 'default') 120 length = int(self.headers.get('Content-Length', 0)) 121 body = self.rfile.read(length) 122 points.save_points(POINTS_DIR, set_name, body) 123 self._respond_json('{"ok":true}') 124 else: 125 self.send_error(404) 126 127 def _respond(self, code, content_type, data, cache=None): 128 self.send_response(code) 129 self.send_header('Content-Type', content_type) 130 self.send_header('Content-Length', str(len(data))) 131 if cache: 132 self.send_header('Cache-Control', cache) 133 self.end_headers() 134 self.wfile.write(data) 135 136 def _query_param(self, key, default=''): 137 """Extract a query parameter from the request path.""" 138 from urllib.parse import urlparse, parse_qs 139 parsed = urlparse(self.path) 140 params = parse_qs(parsed.query) 141 return params.get(key, [default])[0] 142 143 def _respond_json(self, json_str): 144 data = json_str.encode('utf-8') if isinstance(json_str, str) else json_str 145 self._respond(200, 'application/json', data, cache='no-cache') 146 147 def log_message(self, format, *args): 148 pass 149 150 151def main(): 152 http.server.HTTPServer.allow_reuse_address = True 153 server = http.server.HTTPServer(('127.0.0.1', PORT), Handler) 154 url = f'http://127.0.0.1:{PORT}' 155 print(f'Camera visualizer running at {url}') 156 print('Press Ctrl+C to stop.') 157 158 t = threading.Thread(target=server.serve_forever, daemon=True) 159 t.start() 160 webbrowser.open(url) 161 162 stop = threading.Event() 163 signal.signal(signal.SIGINT, lambda *_: stop.set()) 164 stop.wait() 165 print('\nShutting down.') 166 server.server_close() 167 168 169if __name__ == '__main__': 170 main()
39class Handler(http.server.BaseHTTPRequestHandler): 40 41 def do_GET(self): 42 from urllib.parse import urlparse 43 path = urlparse(self.path).path 44 45 # ── Page ────────────────────────────────────────────────── 46 if path in ('/', '/index.html'): 47 page = config.build_page(HERE) 48 self._respond(200, 'text/html; charset=utf-8', page, 49 cache='no-cache, no-store') 50 51 # ── API: robot config ───────────────────────────────────── 52 elif path == '/api/config': 53 self._respond_json(config.build_config_json()) 54 55 # ── API: field list ─────────────────────────────────────── 56 elif path == '/api/field-list': 57 self._respond_json(json.dumps(field_cache.get_field_list())) 58 59 # ── API: field tags (from robotpy_apriltag) ─────────────── 60 elif path == '/api/field-tags': 61 self._respond_json(json.dumps(fields.get_field_tags())) 62 63 # ── API: field GLB model ────────────────────────────────── 64 elif path.startswith('/api/field/') and path.endswith('.glb'): 65 field_id = path[len('/api/field/'):-4] 66 _, glb = field_cache.extract_field(field_id) 67 if glb is None: 68 self.send_error(404, f'Field {field_id} not found') 69 return 70 self._respond(200, 'model/gltf-binary', glb, 71 cache='public, max-age=86400') 72 73 # ── API: field config JSON ──────────────────────────────── 74 elif path.startswith('/api/field/') and path.endswith('.json'): 75 field_id = path[len('/api/field/'):-5] 76 cfg, _ = field_cache.extract_field(field_id) 77 if cfg is None: 78 self.send_error(404, f'Field {field_id} config not found') 79 return 80 self._respond_json(json.dumps(cfg)) 81 82 # ── API: saved points ───────────────────────────────────── 83 elif path == '/api/points/list': 84 self._respond_json(json.dumps(points.list_point_sets(POINTS_DIR))) 85 86 elif path == '/api/points': 87 set_name = self._query_param('set', 'default') 88 self._respond_json(points.load_points(POINTS_DIR, set_name)) 89 90 # ── API: CAD model list ─────────────────────────────────── 91 elif path == '/api/cad-models': 92 self._respond_json(json.dumps(cad.list_cad_models(CAD_DIR))) 93 94 # ── Static: CAD files ───────────────────────────────────── 95 elif path.startswith('/cad/'): 96 data, mime = cad.serve_cad_file(CAD_DIR, path[5:]) 97 if data is None: 98 self.send_error(404) 99 return 100 self._respond(200, mime, data, cache='public, max-age=3600') 101 102 # ── Static: JS modules ──────────────────────────────────── 103 elif path.endswith('.js') and '/' not in path[1:]: 104 fpath = os.path.join(HERE, path[1:]) 105 if not os.path.isfile(fpath): 106 self.send_error(404) 107 return 108 with open(fpath, 'rb') as f: 109 data = f.read() 110 self._respond(200, 'application/javascript', data, 111 cache='no-cache, no-store') 112 113 else: 114 self.send_error(404) 115 116 def do_POST(self): 117 from urllib.parse import urlparse 118 post_path = urlparse(self.path).path 119 if post_path == '/api/points': 120 set_name = self._query_param('set', 'default') 121 length = int(self.headers.get('Content-Length', 0)) 122 body = self.rfile.read(length) 123 points.save_points(POINTS_DIR, set_name, body) 124 self._respond_json('{"ok":true}') 125 else: 126 self.send_error(404) 127 128 def _respond(self, code, content_type, data, cache=None): 129 self.send_response(code) 130 self.send_header('Content-Type', content_type) 131 self.send_header('Content-Length', str(len(data))) 132 if cache: 133 self.send_header('Cache-Control', cache) 134 self.end_headers() 135 self.wfile.write(data) 136 137 def _query_param(self, key, default=''): 138 """Extract a query parameter from the request path.""" 139 from urllib.parse import urlparse, parse_qs 140 parsed = urlparse(self.path) 141 params = parse_qs(parsed.query) 142 return params.get(key, [default])[0] 143 144 def _respond_json(self, json_str): 145 data = json_str.encode('utf-8') if isinstance(json_str, str) else json_str 146 self._respond(200, 'application/json', data, cache='no-cache') 147 148 def log_message(self, format, *args): 149 pass
HTTP request handler base class.
The following explanation of HTTP serves to guide you through the code as well as to expose any misunderstandings I may have about HTTP (so you don't need to read the code to figure out I'm wrong :-).
HTTP (HyperText Transfer Protocol) is an extensible protocol on top of a reliable stream transport (e.g. TCP/IP). The protocol recognizes three parts to a request:
- One line identifying the request type and path
- An optional set of RFC-822-style headers
- An optional data part
The headers and data are separated by a blank line.
The first line of the request has the form
where
The specification specifies that lines are separated by CRLF but for compatibility with the widest range of clients recommends servers also handle LF. Similarly, whitespace in the request line is treated sensibly (allowing multiple spaces between components and allowing trailing whitespace).
Similarly, for output, lines ought to be separated by CRLF pairs but most clients grok LF characters just fine.
If the first line of the request has the form
(i.e.
The reply form of the HTTP 1.x protocol again has three parts:
- One line giving the response code
- An optional set of RFC-822-style headers
- The data
Again, the headers and data are separated by a blank line.
The response code line has the form
where
This server parses the request and the headers, and then calls a
function specific to the request type (
do_SPAM()
Note that the request name is case sensitive (i.e. SPAM and spam are different requests).
The various request details are stored in instance variables:
client_address is the client IP address in the form (host, port);
command, path and version are the broken-down request line;
headers is an instance of email.message.Message (or a derived class) containing the header information;
rfile is a file object open for reading positioned at the start of the optional input data part;
wfile is a file object open for writing.
IT IS IMPORTANT TO ADHERE TO THE PROTOCOL FOR WRITING!
The first thing to be written must be the response line. Then follow 0 or more header lines, then a blank line, and then the actual data (if any). The meaning of the header lines depends on the command executed by the server; in most cases, when data is returned, there should be at least one header line of the form
Content-type:
where
41 def do_GET(self): 42 from urllib.parse import urlparse 43 path = urlparse(self.path).path 44 45 # ── Page ────────────────────────────────────────────────── 46 if path in ('/', '/index.html'): 47 page = config.build_page(HERE) 48 self._respond(200, 'text/html; charset=utf-8', page, 49 cache='no-cache, no-store') 50 51 # ── API: robot config ───────────────────────────────────── 52 elif path == '/api/config': 53 self._respond_json(config.build_config_json()) 54 55 # ── API: field list ─────────────────────────────────────── 56 elif path == '/api/field-list': 57 self._respond_json(json.dumps(field_cache.get_field_list())) 58 59 # ── API: field tags (from robotpy_apriltag) ─────────────── 60 elif path == '/api/field-tags': 61 self._respond_json(json.dumps(fields.get_field_tags())) 62 63 # ── API: field GLB model ────────────────────────────────── 64 elif path.startswith('/api/field/') and path.endswith('.glb'): 65 field_id = path[len('/api/field/'):-4] 66 _, glb = field_cache.extract_field(field_id) 67 if glb is None: 68 self.send_error(404, f'Field {field_id} not found') 69 return 70 self._respond(200, 'model/gltf-binary', glb, 71 cache='public, max-age=86400') 72 73 # ── API: field config JSON ──────────────────────────────── 74 elif path.startswith('/api/field/') and path.endswith('.json'): 75 field_id = path[len('/api/field/'):-5] 76 cfg, _ = field_cache.extract_field(field_id) 77 if cfg is None: 78 self.send_error(404, f'Field {field_id} config not found') 79 return 80 self._respond_json(json.dumps(cfg)) 81 82 # ── API: saved points ───────────────────────────────────── 83 elif path == '/api/points/list': 84 self._respond_json(json.dumps(points.list_point_sets(POINTS_DIR))) 85 86 elif path == '/api/points': 87 set_name = self._query_param('set', 'default') 88 self._respond_json(points.load_points(POINTS_DIR, set_name)) 89 90 # ── API: CAD model list ─────────────────────────────────── 91 elif path == '/api/cad-models': 92 self._respond_json(json.dumps(cad.list_cad_models(CAD_DIR))) 93 94 # ── Static: CAD files ───────────────────────────────────── 95 elif path.startswith('/cad/'): 96 data, mime = cad.serve_cad_file(CAD_DIR, path[5:]) 97 if data is None: 98 self.send_error(404) 99 return 100 self._respond(200, mime, data, cache='public, max-age=3600') 101 102 # ── Static: JS modules ──────────────────────────────────── 103 elif path.endswith('.js') and '/' not in path[1:]: 104 fpath = os.path.join(HERE, path[1:]) 105 if not os.path.isfile(fpath): 106 self.send_error(404) 107 return 108 with open(fpath, 'rb') as f: 109 data = f.read() 110 self._respond(200, 'application/javascript', data, 111 cache='no-cache, no-store') 112 113 else: 114 self.send_error(404)
116 def do_POST(self): 117 from urllib.parse import urlparse 118 post_path = urlparse(self.path).path 119 if post_path == '/api/points': 120 set_name = self._query_param('set', 'default') 121 length = int(self.headers.get('Content-Length', 0)) 122 body = self.rfile.read(length) 123 points.save_points(POINTS_DIR, set_name, body) 124 self._respond_json('{"ok":true}') 125 else: 126 self.send_error(404)
Log an arbitrary message.
This is used by all other logging functions. Override it if you have specific logging wishes.
The first argument, FORMAT, is a format string for the message to be logged. If the format string contains any % escapes requiring parameters, they should be specified as subsequent arguments (it's just like printf!).
The client ip and current date/time are prefixed to every message.
Unicode control characters are replaced with escaped hex before writing the output to stderr.
152def main(): 153 http.server.HTTPServer.allow_reuse_address = True 154 server = http.server.HTTPServer(('127.0.0.1', PORT), Handler) 155 url = f'http://127.0.0.1:{PORT}' 156 print(f'Camera visualizer running at {url}') 157 print('Press Ctrl+C to stop.') 158 159 t = threading.Thread(target=server.serve_forever, daemon=True) 160 t.start() 161 webbrowser.open(url) 162 163 stop = threading.Event() 164 signal.signal(signal.SIGINT, lambda *_: stop.set()) 165 stop.wait() 166 print('\nShutting down.') 167 server.server_close()