Merge pull request #3 from gurgleapps/esp

ESP support and new features like file list and ip blinking. Minimise memory usage.

Author: gurgleapps

README.md

@@ -6,17 +6,22 @@ This repository contains code to control Raspberry Pi Pico, ESP8266, ESP32 or ot
 
 ## Features
 
-- Serve static web pages from your Raspberry Pi Pico
+- Serve static and dynamic web pages from your Raspberry Pi Pico
 - Run Python functions on your microcontroller device from a web browser
 - Create dynamic web pages with live data from your Pico or other Microcontroller
+- Blink the IP address using the built-in LED, handy when you're out in the field with no screen or computer
+- Display a file and folder list of your root directory with an attractive and responsive user interface
+- End-to-end examples showcasing various functionalities
+- Easily customizable codebase
 
 ## Setup
 
 1. Make sure you have MicroPython on your Pico
 2. Clone this repository
 3. Copy the code to your Pico
-4. Edit `config.py` with your Wi-Fi details
+4. Edit `config.py` with your Wi-Fi details and IP blink options
 5. Run `main.py` and look for the IP address of your web server
+6. Point your browsers to http://<YOUR_IP> 
 
 ## Documentation
 

config.py

@@ -1,2 +1,4 @@
 WIFI_SSID = "your_wifi_ssid"
-WIFI_PASSWORD = "your_wifi_password"
+WIFI_PASSWORD = "your_wifi_password"
+BLINK_IP = False
+BLINK_LAST_ONLY = False

gurgleapps_webserver.py

@@ -21,6 +21,7 @@ class GurgleAppsWebserver:
 
     def __init__(self, wifi_ssid, wifi_password, port=80, timeout=20, doc_root="/www", log_level=0):
         print("GurgleApps.com Webserver")
+        self.ip_address = '1.1.1.1'
         self.port = port
         self.timeout = timeout
         self.wifi_ssid = wifi_ssid
@@ -50,16 +51,17 @@ def __init__(self, wifi_ssid, wifi_password, port=80, timeout=20, doc_root="/www
             print('waiting for connection...')
             time.sleep(1)
 
-        #if self.wlan.status() != 3:
+        # if self.wlan.status() != 3:
         if self.wlan.isconnected() == False:
             raise RuntimeError('network connection failed')
         else:
             print('connected')
             status = self.wlan.ifconfig()
             print('ip = ' + status[0])
         self.serving = True
+        self.ip_address = status[0]
         print('point your browser to http://', status[0])
-        #asyncio.new_event_loop()
+        # asyncio.new_event_loop()
         print("exit constructor")
 
     # async def start_server(self):
@@ -96,7 +98,7 @@ async def serve_request(self, reader, writer):
             post_data = None
             while True:
                 line = await reader.readline()
-                #print("line: "+str(line))
+                # print("line: "+str(line))
                 line = line.decode('utf-8').strip()
                 if line == "":
                     break
@@ -109,7 +111,7 @@ async def serve_request(self, reader, writer):
                 method = match.group(1)
                 url = match.group(2)
                 print(method, url)
-            else: # regex didn't match, try splitting the request line
+            else:  # regex didn't match, try splitting the request line
                 request_parts = request_raw.split(" ")
                 if len(request_parts) > 1:
                     method = request_parts[0]
@@ -144,14 +146,22 @@ async def serve_request(self, reader, writer):
             file_path = self.doc_root + url
             if self.log_level > 0:
                 print("file_path: "+str(file_path))
-            #if uos.stat(file_path)[6] > 0:
+            # if uos.stat(file_path)[6] > 0:
             if self.file_exists(file_path):
                 content_type = self.get_content_type(url)
                 if self.log_level > 1:
                     print("content_type: "+str(content_type))
                 await response.send_file(file_path, content_type=content_type)
                 return
-            print("file not found")
+            if url == "/":
+                print("root")
+                files_and_folders = self.list_files_and_folders(self.doc_root)
+                await response.send_iterator(self.generate_root_page_html(files_and_folders))
+                return
+                html = self.generate_root_page_html(files_and_folders)
+                await response.send(html)
+                return
+            print("file not found "+url)
             await response.send(self.html % "page not found "+url, status_code=404)
             if (url == "/shutdown"):
                 self.serving = False
@@ -163,7 +173,7 @@ def dir_exists(self, filename):
             return (os.stat(filename)[0] & 0x4000) != 0
         except OSError:
             return False
-        
+
     def file_exists(self, filename):
         try:
             return (os.stat(filename)[0] & 0x4000) == 0
@@ -220,8 +230,7 @@ def get_file_extension(self, file_path):
             return file_parts[-1]
         return ''
 
-
-    def get_content_type(self,file_path):
+    def get_content_type(self, file_path):
         extension = self.get_file_extension(file_path)
         content_type_map = {
             'html': 'text/html',
@@ -231,6 +240,127 @@ def get_content_type(self,file_path):
             'jpeg': 'image/jpeg',
             'png': 'image/png',
             'gif': 'image/gif',
-            'ico': 'image/x-icon'
+            'webp': 'image/webp',
+            'ico': 'image/x-icon',
+            'svg': 'image/svg+xml',
+            'json': 'application/json',
+            'xml': 'application/xml',
+            'pdf': 'application/pdf',
+            'zip': 'application/zip',
+            'txt': 'text/plain',
+            'csv': 'text/csv',
+            'mp3': 'audio/mpeg',
+            'mp4': 'video/mp4',
+            'wav': 'audio/wav',
+            'ogg': 'audio/ogg',
+            'webm': 'video/webm',
         }
         return content_type_map.get(extension, 'text/plain')
+
+    # long pause for dots 4 quick blinks for zero 2 quick for a dot
+    async def blink_ip(self, led_pin, ip=None, repeat=2, delay_between_digits=0.9, last_only=False):
+        delay_between_repititions = 2
+        if ip == None:
+            ip = self.ip_address
+        print("blink_ip: " + str(ip))
+
+        def blink_element(element, pin, duration=0.27):
+            if element == '-':
+                blinks = 9
+                duration = 0.1
+            elif element == '.':
+                blinks = 2
+                duration = 0.1
+            elif element == 0:
+                blinks = 4
+                duration = 0.1
+            else:
+                blinks = element
+
+            for _ in range(blinks):
+                pin.on()
+                time.sleep(duration)
+                pin.off()
+                time.sleep(duration)
+
+        ip_digits_and_dots = []
+        ip_parts = ip.split('.')
+        if last_only:
+            # Only blink the last part of the IP address
+            ip_parts = [ip_parts[-1]]
+
+        for part in ip_parts:
+            for digit in part:
+                ip_digits_and_dots.append(int(digit))
+            # Add a dot to the list to represent the separator
+            ip_digits_and_dots.append('.')
+        ip_digits_and_dots.pop()  # Remove the last dot
+        # Add a dash to the list to represent the end of the IP address
+        ip_digits_and_dots.append('-')
+
+        for _ in range(repeat):
+            for element in ip_digits_and_dots:
+                blink_element(element, led_pin)
+                await asyncio.sleep(delay_between_digits if element != '.' else 2 * delay_between_digits)
+            await asyncio.sleep(delay_between_repititions)
+
+    def list_files_and_folders(self, path):
+        entries = uos.ilistdir(path)
+        files_and_folders = []
+        for entry in entries:
+            name = entry[0]
+            mode = entry[1]
+            if mode & 0o170000 == 0o040000:  # Check if it's a directory
+                files_and_folders.append({"name": name, "type": "directory"})
+            elif mode & 0o170000 == 0o100000:  # Check if it's a file
+                files_and_folders.append({"name": name, "type": "file"})
+        return files_and_folders
+
+    def generate_root_page_html(self, files_and_folders):
+        yield """
+       <!DOCTYPE html>
+        <html>
+            <head>
+                <title>GurgleApps.com Webserver</title>
+                <meta name="viewport" content="width=device-width, initial-scale=1.0">
+                <link href="/styles.css" rel="stylesheet">
+            </head>
+            <body class="bg-gray-100">
+        """
+        yield """
+        <div class="relative flex min-h-screen flex-col justify-center overflow-hidden bg-gray-50 py-6 sm:py-12">
+        <div class="relative bg-white px-6 pb-8 pt-10 shadow-xl ring-1 ring-gray-900/5 sm:mx-auto sm:max-w-lg sm:rounded-lg sm:px-10">
+        <div class="mx-auto max-w-md">
+        <img src="/img/logo.svg" class="h-12 w-auto" alt="GurgleApps.com">
+        """
+        yield """
+        <div class="divide-y divide-gray-300/50">
+        <div class="space-y-6 py-8 text-base leading-7 text-gray-600">
+          <h1 class="text-lg font-semibold">Welcome to GurgleApps.com Webserver</h1>
+          <h12 class="text-base font-semibold">File List:</h2>
+          <ul class="space-y-2 mt-3">
+        """
+        folder_icon_svg = """
+        <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="currentColor" class="w-6 h-6  fill-indigo-800">
+        <path d="M19.5 21a3 3 0 003-3v-4.5a3 3 0 00-3-3h-15a3 3 0 00-3 3V18a3 3 0 003 3h15zM1.5 10.146V6a3 3 0 013-3h5.379a2.25 2.25 0 011.59.659l2.122 2.121c.14.141.331.22.53.22H19.5a3 3 0 013 3v1.146A4.483 4.483 0 0019.5 9h-15a4.483 4.483 0 00-3 1.146z" />
+        </svg>
+        """
+        file_icon_svg = """
+        <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="currentColor" class="w-6 h-6 fill-indigo-800">
+        <path d="M5.625 1.5c-1.036 0-1.875.84-1.875 1.875v17.25c0 1.035.84 1.875 1.875 1.875h12.75c1.035 0 1.875-.84 1.875-1.875V12.75A3.75 3.75 0 0016.5 9h-1.875a1.875 1.875 0 01-1.875-1.875V5.25A3.75 3.75 0 009 1.5H5.625z" />
+        <path d="M12.971 1.816A5.23 5.23 0 0114.25 5.25v1.875c0 .207.168.375.375.375H16.5a5.23 5.23 0 013.434 1.279 9.768 9.768 0 00-6.963-6.963z" />
+        </svg>
+        """
+        for file_or_folder in files_and_folders:
+            icon = folder_icon_svg if file_or_folder['type'] == 'directory' else file_icon_svg
+            yield f"<li class='border-t pt-1'><a href='/{file_or_folder['name']}' class='flex items-center font-semibold text-slate-800 hover:text-indigo-800'>{icon}<p class='ml-2'>{file_or_folder['name']}</p></a></li>"
+        yield "</ul>"
+        # Closing tags for the body and container div
+        yield """
+        </div>
+        <div class="pt-3 text-base font-semibold leading-7">
+        <p class="text-gray-900">More information</p><p><a href="https://gurgleapps.com/learn/projects/micropython-web-server-control-raspberry-pi-pico-projects" class="text-indigo-500 hover:text-sky-600">Project Home &rarr;</a>
+        </p></div></div></div></div></div></body></html>
+        """
+       
+

main.py

@@ -87,15 +87,23 @@ async def background_task():
 
 
 async def run():
+    #await(server.blink_ip(led_pin=led))
+    if config.BLINK_IP:
+        await(server.blink_ip(led_pin = led, last_only = config.BLINK_LAST_ONLY))
+
     await asyncio.gather(main(), background_task())
 
 server = GurgleAppsWebserver(config.WIFI_SSID, config.WIFI_PASSWORD, port=80, timeout=20, doc_root="/www", log_level=2)
 server.add_function_route("/set-delay/<delay>", set_delay)
-server.add_function_route("/set-blink-pattern/<on_time>/<off_time>", set_blink_pattern)
+server.add_function_route(
+    "/set-blink-pattern/<on_time>/<off_time>",
+    set_blink_pattern
+)
 server.add_function_route("/stop", stop_flashing)
 server.add_function_route("/start", start_flashing)
 server.add_function_route("/status", send_status)
 server.add_function_route("/example/func/<param1>/<param2>", example_func)
 server.add_function_route("/hello/<name>", say_hello)
 
+
 asyncio.run(run())

response.py

@@ -29,7 +29,8 @@ async def send_file(self, file_path, status_code=200, content_type="text/html"):
             file_size = os.stat(file_path)[6]
             await self.send_headers(status_code, content_type, content_length=file_size)
 
-            chunk_size = 1024  # Adjust the chunk size as needed
+            #chunk_size = 1024  # Adjust the chunk size as needed
+            chunk_size = 512  # Adjust the chunk size as needed
             with open(file_path, "rb") as f:
                 while True:
                     chunk = f.read(chunk_size)
@@ -43,3 +44,11 @@ async def send_file(self, file_path, status_code=200, content_type="text/html"):
         except Exception as e:
             print("Error sending file:", e)
             await self.send('', status_code=404)
+
+    async def send_iterator(self, iterator, status_code=200, content_type="text/html"):
+        await self.send_headers(status_code=status_code, content_type=content_type)
+        for chunk in iterator:
+            self.writer.write(chunk)
+            await self.writer.drain()
+        await self.writer.wait_closed()
+