# Copyright 2022 The Pigweed Authors # # Licensed under the Apache License, Version 2.0 (the "License"); you may not # use this file except in compliance with the License. You may obtain a copy of # the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations under # the License. """Console HTTP Log Server functions.""" import logging from pathlib import Path import mimetypes import http.server from typing import Callable _LOG = logging.getLogger(__package__) def _start_serving(port: int, handler: Callable) -> bool: try: with http.server.HTTPServer(('', port), handler) as httpd: _LOG.debug('Serving on port %i', port) httpd.serve_forever() return True except OSError: _LOG.debug('Port %i failed.', port) return False def pw_console_http_server(starting_port: int, handler: Callable) -> None: for i in range(100): if _start_serving(starting_port + i, handler): break class ConsoleLogHTTPRequestHandler(http.server.BaseHTTPRequestHandler): """Request handler that serves files from pw_console.html package data.""" def __init__(self, html_files: dict[str, str], *args, **kwargs): self.html_files = html_files super().__init__(*args, **kwargs) def do_GET(self): # pylint: disable=invalid-name _LOG.debug( '%s: %s', self.client_address[0], self.raw_requestline.decode('utf-8').strip(), ) path = self.path if path == '/': path = '/index.html' if path not in self.html_files: self.send_error(http.server.HTTPStatus.NOT_FOUND, 'File not found') return content: str = self.html_files[path].encode('utf-8') content_type = 'application/octet-stream' mime_guess, _ = mimetypes.guess_type(Path(path).name) if mime_guess: content_type = mime_guess self.send_response(http.server.HTTPStatus.OK) self.send_header('Content-type', content_type) self.send_header('Content-Length', str(len(content))) self.send_header('Last-Modified', self.date_time_string()) self.end_headers() self.wfile.write(content) def log_message(self, *args, **kwargs): pass