blob: 4d5c45698459c9d2eb956a537956a9a2771928e9 [file] [log] [blame]
# 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 Dict, 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