Merge branch 'main' into https-implementation

docs/conf.py

@@ -116,7 +116,6 @@ napoleon_numpy_docstring = False
 import sphinx_rtd_theme
 
 html_theme = "sphinx_rtd_theme"
-html_theme_path = [sphinx_rtd_theme.get_html_theme_path(), "."]
 
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,

docs/examples.rst

@@ -355,6 +355,23 @@ but it is recommended as it makes it easier to handle multiple tasks. It can be
     :emphasize-lines: 12,20,65-72,88,99
     :linenos:
 
+Custom response types e.g. video streaming
+------------------------------------------
+
+The built-in response types may not always meet your specific requirements. In such cases, you can define custom response types and implement
+the necessary logic.
+
+The example below demonstrates a ``XMixedReplaceResponse`` class, which uses the ``multipart/x-mixed-replace`` content type to stream video frames
+from a camera, similar to a CCTV system.
+
+To ensure the server remains responsive, a global list of open connections is maintained. By running tasks asynchronously, the server can stream
+video to multiple clients while simultaneously handling other requests.
+
+.. literalinclude:: ../examples/httpserver_video_stream.py
+    :caption: examples/httpserver_video_stream.py
+    :emphasize-lines: 31-77,92
+    :linenos:
+
 SSL/TLS (HTTPS)
 ---------------
 

examples/httpserver_multiple_servers.py

@@ -42,11 +42,11 @@ def home(request: Request):
     return Response(request, "Hello from home!")
 
 
-id_address = str(wifi.radio.ipv4_address)
+ip_address = str(wifi.radio.ipv4_address)
 
 # Start the servers.
-bedroom_server.start(id_address, 5000)
-office_server.start(id_address, 8000)
+bedroom_server.start(ip_address, 5000)
+office_server.start(ip_address, 8000)
 
 while True:
     try:

examples/httpserver_video_stream.py

@@ -0,0 +1,128 @@
+# SPDX-FileCopyrightText: 2024 Michał Pokusa
+#
+# SPDX-License-Identifier: Unlicense
+
+try:
+    from typing import Dict, List, Tuple, Union
+except ImportError:
+    pass
+
+from asyncio import create_task, gather, run, sleep
+from random import choice
+
+import socketpool
+import wifi
+
+from adafruit_pycamera import PyCamera
+from adafruit_httpserver import Server, Request, Response, Headers, Status, OK_200
+
+
+pool = socketpool.SocketPool(wifi.radio)
+server = Server(pool, debug=True)
+
+
+camera = PyCamera()
+camera.display.brightness = 0
+camera.mode = 0  # JPEG, required for `capture_into_jpeg()`
+camera.resolution = "1280x720"
+camera.effect = 0  # No effect
+
+
+class XMixedReplaceResponse(Response):
+    def __init__(
+        self,
+        request: Request,
+        frame_content_type: str,
+        *,
+        status: Union[Status, Tuple[int, str]] = OK_200,
+        headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
+    ) -> None:
+        super().__init__(
+            request=request,
+            headers=headers,
+            cookies=cookies,
+            status=status,
+        )
+        self._boundary = self._get_random_boundary()
+        self._headers.setdefault(
+            "Content-Type", f"multipart/x-mixed-replace; boundary={self._boundary}"
+        )
+        self._frame_content_type = frame_content_type
+
+    @staticmethod
+    def _get_random_boundary() -> str:
+        symbols = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+        return "--" + "".join([choice(symbols) for _ in range(16)])
+
+    def send_frame(self, frame: Union[str, bytes] = "") -> None:
+        encoded_frame = bytes(
+            frame.encode("utf-8") if isinstance(frame, str) else frame
+        )
+
+        self._send_bytes(
+            self._request.connection, bytes(f"{self._boundary}\r\n", "utf-8")
+        )
+        self._send_bytes(
+            self._request.connection,
+            bytes(f"Content-Type: {self._frame_content_type}\r\n\r\n", "utf-8"),
+        )
+        self._send_bytes(self._request.connection, encoded_frame)
+        self._send_bytes(self._request.connection, bytes("\r\n", "utf-8"))
+
+    def _send(self) -> None:
+        self._send_headers()
+
+    def close(self) -> None:
+        self._close_connection()
+
+
+stream_connections: List[XMixedReplaceResponse] = []
+
+
+@server.route("/frame")
+def frame_handler(request: Request):
+    frame = camera.capture_into_jpeg()
+
+    return Response(request, body=frame, content_type="image/jpeg")
+
+
+@server.route("/stream")
+def stream_handler(request: Request):
+    response = XMixedReplaceResponse(request, frame_content_type="image/jpeg")
+    stream_connections.append(response)
+
+    return response
+
+
+async def send_stream_frames():
+    while True:
+        await sleep(0.1)
+
+        frame = camera.capture_into_jpeg()
+
+        for connection in iter(stream_connections):
+            try:
+                connection.send_frame(frame)
+            except BrokenPipeError:
+                connection.close()
+                stream_connections.remove(connection)
+
+
+async def handle_http_requests():
+    server.start(str(wifi.radio.ipv4_address))
+
+    while True:
+        await sleep(0)
+
+        server.poll()
+
+
+async def main():
+    await gather(
+        create_task(send_stream_frames()),
+        create_task(handle_http_requests()),
+    )
+
+
+run(main())