From f1f9a5524a48e56825f44ad25db8c62be9d56654 Mon Sep 17 00:00:00 2001 From: Justin Myers Date: Wed, 24 Apr 2024 19:58:33 -0700 Subject: [PATCH] Remove wsgiserver --- README.rst | 64 ----- .../adafruit_wiznet5k_wsgiserver.py | 240 ------------------ docs/api.rst | 2 - examples/wiznet5k_wsgiserver.py | 140 ---------- 4 files changed, 446 deletions(-) delete mode 100644 adafruit_wiznet5k/adafruit_wiznet5k_wsgiserver.py delete mode 100644 examples/wiznet5k_wsgiserver.py diff --git a/README.rst b/README.rst index e5adb59..79ed6f8 100644 --- a/README.rst +++ b/README.rst @@ -108,70 +108,6 @@ wifitest.adafruit.com. print("Done!") -This example demonstrates a simple web server that allows setting the Neopixel color. - -.. code-block:: python - - import board - import busio - import digitalio - import neopixel - - from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K - import adafruit_wiznet5k.adafruit_wiznet5k_wsgiserver as server - from adafruit_wsgi.wsgi_app import WSGIApp - - print("Wiznet5k Web Server Test") - - # Status LED - led = neopixel.NeoPixel(board.NEOPIXEL, 1) - led.brightness = 0.3 - led[0] = (0, 0, 255) - - # W5500 connections - cs = digitalio.DigitalInOut(board.D10) - spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) - - # Initialize Ethernet interface with DHCP - eth = WIZNET5K(spi_bus, cs) - - # Here we create our application, registering the - # following functions to be called on specific HTTP GET requests routes - - web_app = WSGIApp() - - @web_app.route("/led///") - def led_on(request, r, g, b): - print("LED handler") - led.fill((int(r), int(g), int(b))) - return ("200 OK", [], ["LED set!"]) - - @web_app.route("/") - def root(request): - print("Root handler") - return ("200 OK", [], ["Root document"]) - - @web_app.route("/large") - def large(request): - print("Large pattern handler") - return ("200 OK", [], ["*-.-" * 2000]) - - - # Here we setup our server, passing in our web_app as the application - server.set_interface(eth) - wsgiServer = server.WSGIServer(80, application=web_app) - - print("Open this IP in your browser: ", eth.pretty_ip(eth.ip_address)) - - # Start the server - wsgiServer.start() - while True: - # Our main loop where we have the server poll for incoming requests - wsgiServer.update_poll() - # Maintain DHCP lease - eth.maintain_dhcp_lease() - # Could do any other background tasks here, like reading sensors - Documentation ============= diff --git a/adafruit_wiznet5k/adafruit_wiznet5k_wsgiserver.py b/adafruit_wiznet5k/adafruit_wiznet5k_wsgiserver.py deleted file mode 100644 index d4625e3..0000000 --- a/adafruit_wiznet5k/adafruit_wiznet5k_wsgiserver.py +++ /dev/null @@ -1,240 +0,0 @@ -# Based on ESP32 code Copyright (c) 2019 Matt Costi for Adafruit Industries -# SPDX-FileCopyrightText: Copyright (c) 2020 Patrick Van Oosterwijck -# -# SPDX-License-Identifier: MIT - -""" -`adafruit_wiznet5k_wsgiserver` -================================================================================ - -A simple WSGI (Web Server Gateway Interface) server that interfaces with the W5500. -Opens a listening port on the W5500 to listen for incoming HTTP Requests and -Accepts an Application object that must be callable, which gets called -whenever a new HTTP Request has been received. - -The Application MUST accept 2 ordered parameters: - 1. environ object (incoming request data) - 2. start_response function. Must be called before the Application - callable returns, in order to set the response status and headers. - -The Application MUST return strings in a list, which is the response data - -Requires update_poll being called in the applications main event loop. - -For more details about Python WSGI see: -https://www.python.org/dev/peps/pep-0333/ - -* Author(s): Matt Costi, Patrick Van Oosterwijck -""" -# pylint: disable=no-name-in-module -from __future__ import annotations - -try: - from typing import TYPE_CHECKING, Optional, List, Tuple, Dict - - if TYPE_CHECKING: - from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K -except ImportError: - pass - -import io -import gc -from micropython import const -import adafruit_wiznet5k as wiznet5k -import adafruit_wiznet5k.adafruit_wiznet5k_socket as socket - -_the_interface: Optional[WIZNET5K] = None # pylint: disable=invalid-name - - -def set_interface(iface: WIZNET5K) -> None: - """ - Helper to set the global internet interface. - - :param wiznet5k.adafruit_wiznet5k.WIZNET5K: Ethernet interface. - """ - global _the_interface # pylint: disable=global-statement, invalid-name - _the_interface = iface - socket.set_interface(iface) - - -# pylint: disable=invalid-name -class WSGIServer: - """A simple server that implements the WSGI interface.""" - - def __init__( - self, - port: int = 80, - debug: bool = False, - application: Optional[callable] = None, - ) -> None: - """ - :param int port: WSGI server port, defaults to 80. - :param bool debug: Enable debugging, defaults to False. - :param Optional[callable] application: Application to call in response to a HTTP request. - """ - self.application = application - self.port = port - self._timeout = 20 - self._client_sock = [] - self._debug = debug - - self._response_status = None - self._response_headers = [] - if _the_interface.chip == "w5100s": - self.MAX_SOCK_NUM = const(2) - else: - self.MAX_SOCK_NUM = const(6) - if self._debug: - print("Max sockets: ", self.MAX_SOCK_NUM) - - def start(self) -> None: - """ - Start the server and listen for incoming connections. - - Call update_poll in the main loop for the application callable to be - invoked on receiving an incoming request. - """ - for _ in range(self.MAX_SOCK_NUM): - new_sock = socket.socket() - new_sock.settimeout(self._timeout) - new_sock.bind((None, self.port)) - new_sock.listen() - self._client_sock.append(new_sock) - if self._debug: - ip = _the_interface.pretty_ip(_the_interface.ip_address) - print("Server available at {0}:{1}".format(ip, self.port)) - - def update_poll(self) -> None: - """ - Check for new incoming client requests. - - Call this method inside your main event loop to get the server - check for new incoming client requests. When a request comes in, - the application callable will be invoked. - """ - for sock in self._client_sock: - if sock._available(): # pylint: disable=protected-access - environ = self._get_environ(sock) - result = self.application(environ, self._start_response) - self.finish_response(result, sock) - self._client_sock.remove(sock) - break - for sock in self._client_sock: - if ( - sock._status # pylint: disable=protected-access - == wiznet5k.adafruit_wiznet5k.SNSR_SOCK_CLOSED - ): - self._client_sock.remove(sock) - for _ in range(len(self._client_sock), self.MAX_SOCK_NUM): - try: - new_sock = socket.socket() - new_sock.settimeout(self._timeout) - new_sock.bind((None, self.port)) - new_sock.listen() - self._client_sock.append(new_sock) - except RuntimeError: - pass - - def finish_response(self, result: str, client: socket.socket) -> None: - """ - Called after the application callable returns result data to respond with. - Creates the HTTP Response payload from the response_headers and results data, - and sends it back to client. - - :param str result: the data string to send back in the response to the client. - :param socket.socket client: the socket to send the response to. - """ - try: - response = "HTTP/1.1 {0}\r\n".format(self._response_status) - for header in self._response_headers: - response += "{0}: {1}\r\n".format(*header) - response += "\r\n" - client.send(response.encode("utf-8")) - for data in result: - if not isinstance(data, bytes): - data = data.encode("utf-8") - if len(data) < 0x800: - client.send(data) - else: - # split to chunks of 2 kb - data_chunks = [ - data[i : i + 0x800] for i in range(0, len(data), 0x800) - ] - for data_chunk in data_chunks: - client.send(data_chunk) - gc.collect() - finally: - client._disconnect() # pylint: disable=protected-access - client.close() - - def _start_response( - self, status: str, response_headers: List[Tuple[str, str]] - ) -> None: - """ - The application callable will be given this method as the second param - This is to be called before the application callable returns, to signify - the response can be started with the given status and headers. - - :param str status: a status string including the code and reason. ex: "200 OK" - :param List[Tuple[str, str]] response_headers: a list of tuples to represent the headers. - ex ("header-name", "header value") - """ - self._response_status = status - self._response_headers = [("Server", "w5kWSGIServer")] + response_headers - - def _get_environ(self, client: socket.socket) -> Dict: - """ - The application callable will be given the resulting environ dictionary. - It contains metadata about the incoming request and the request body ("wsgi.input") - - :param socket.socket client: Socket to read the request from. - - :return Dict: Data for the application callable. - """ - env = {} - line = str(client._readline(), "utf-8") # pylint: disable=protected-access - (method, path, ver) = line.rstrip("\r\n").split(None, 2) - - env["wsgi.version"] = (1, 0) - env["wsgi.url_scheme"] = "http" - env["wsgi.multithread"] = False - env["wsgi.multiprocess"] = False - env["wsgi.run_once"] = False - - env["REQUEST_METHOD"] = method - env["SCRIPT_NAME"] = "" - env["SERVER_NAME"] = _the_interface.pretty_ip(_the_interface.ip_address) - env["SERVER_PROTOCOL"] = ver - env["SERVER_PORT"] = self.port - if path.find("?") >= 0: - env["PATH_INFO"] = path.split("?")[0] - env["QUERY_STRING"] = path.split("?")[1] - else: - env["PATH_INFO"] = path - - headers = {} - while True: - header = str( - client._readline(), "utf-8" # pylint: disable=protected-access - ) - if header == "": - break - title, content = header.split(": ", 1) - headers[title.lower()] = content - - if "content-type" in headers: - env["CONTENT_TYPE"] = headers.get("content-type") - if "content-length" in headers: - env["CONTENT_LENGTH"] = headers.get("content-length") - body = client.recv(int(env["CONTENT_LENGTH"])) - env["wsgi.input"] = io.StringIO(body) - else: - body = client.recv(0) - env["wsgi.input"] = io.StringIO(body) - for name, value in headers.items(): - key = "HTTP_" + name.replace("-", "_").upper() - if key in env: - value = "{0},{1}".format(env[key], value) - env[key] = value - - return env diff --git a/docs/api.rst b/docs/api.rst index 560c4b0..83d7f49 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -16,5 +16,3 @@ .. automodule:: adafruit_wiznet5k.adafruit_wiznet5k_dns :members: -.. automodule:: adafruit_wiznet5k.adafruit_wiznet5k_wsgiserver - :members: diff --git a/examples/wiznet5k_wsgiserver.py b/examples/wiznet5k_wsgiserver.py deleted file mode 100644 index 5b3d2c2..0000000 --- a/examples/wiznet5k_wsgiserver.py +++ /dev/null @@ -1,140 +0,0 @@ -# SPDX-FileCopyrightText: 2021 Patrick Van Oosterwijck @ Silicognition LLC -# -# SPDX-License-Identifier: MIT -# -# This demo was tested with the PoE-FeatherWing, which contains a 24AA02E48 -# chip to provide a globally unique MAC address, but can also work without -# this chip for testing purposes by using a hard coded MAC. -# -# It also contains a `get_static_file` function that demonstrates how to -# use a generator to serve large static files without using up too much -# memory. To avoid having to put extra files in the repo, it just serves -# `code.py` which isn't very large, but to properly test it, adjust the code -# to serve an image of several 100 kB to see how it works. -# -# There's also an endpoint that demonstrates that `requests` can be used to -# get data from another socket and serve it. -# - -import board -import busio -import digitalio -import neopixel - -import adafruit_connection_manager -import adafruit_requests -from adafruit_wsgi.wsgi_app import WSGIApp -from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K -import adafruit_wiznet5k.adafruit_wiznet5k_wsgiserver as server - - -print("Wiznet5k Web Server Test") - - -def get_mac(i2c_obj): - "Read MAC from 24AA02E48 chip and return it" - mac_addr = bytearray(6) - while not i2c_obj.try_lock(): - pass - i2c_obj.writeto(0x50, bytearray((0xFA,))) - i2c_obj.readfrom_into(0x50, mac_addr, start=0, end=6) - i2c_obj.unlock() - return mac_addr - - -def get_static_file(filename): - "Static file generator" - with open(filename, "rb") as f: - b = None - while b is None or len(b) == 2048: - b = f.read(2048) - yield b - - -# Status LED -led = neopixel.NeoPixel(board.NEOPIXEL, 1) -led.brightness = 0.3 -led[0] = (255, 0, 0) - -# Chip Select for PoE-FeatherWing and Adafruit Ethernet FeatherWing -cs = digitalio.DigitalInOut(board.D10) -# Chip Select for Particle Ethernet FeatherWing -# cs = digitalio.DigitalInOut(board.D5) - -# Initialize SPI bus -spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO) - -try: - # Initialize the I2C bus to read the MAC - i2c = busio.I2C(board.SCL, board.SDA) - # Read the MAC from the 24AA02E48 chip - mac = get_mac(i2c) -except (RuntimeError, OSError): - # Hard coded MAC if there is no 24AA02E48 - mac = b"\xFE\xED\xDE\xAD\xBE\xEF" - -# Initialize Ethernet interface with DHCP -eth = WIZNET5K(spi_bus, cs, mac=mac) - -# Initialize a requests session -pool = adafruit_connection_manager.get_radio_socketpool(eth) -ssl_context = adafruit_connection_manager.get_radio_ssl_context(eth) -requests = adafruit_requests.Session(pool, ssl_context) - - -# Here we create our application, registering the -# following functions to be called on specific HTTP GET requests routes - -web_app = WSGIApp() - - -@web_app.route("/led///") -def led_on(request, r, g, b): # pylint: disable=unused-argument - print("LED handler") - led.fill((int(r), int(g), int(b))) - return ("200 OK", [], ["LED set!"]) - - -@web_app.route("/") -def root(request): # pylint: disable=unused-argument - print("Root WSGI handler") - return ("200 OK", [], ["Root document"]) - - -@web_app.route("/large") -def large(request): # pylint: disable=unused-argument - print("Large pattern handler") - return ("200 OK", [], ["*-.-" * 2000]) - - -@web_app.route("/code") -def code(request): # pylint: disable=unused-argument - print("Static file code.py handler") - return ("200 OK", [], get_static_file("code.py")) - - -@web_app.route("/btc") -def btc(request): # pylint: disable=unused-argument - print("BTC handler") - r = requests.get("http://api.coindesk.com/v1/bpi/currentprice/USD.json") - result = r.text - r.close() - return ("200 OK", [], [result]) - - -# Here we setup our server, passing in our web_app as the application -server.set_interface(eth) -wsgiServer = server.WSGIServer(80, application=web_app) - -print("Open this IP in your browser: ", eth.pretty_ip(eth.ip_address)) - -# Start the server -wsgiServer.start() -led[0] = (0, 0, 255) - -while True: - # Our main loop where we have the server poll for incoming requests - wsgiServer.update_poll() - # Maintain DHCP lease - eth.maintain_dhcp_lease() - # Could do any other background tasks here, like reading sensors