Merge pull request #3 from MikeSchapp/wifi-configuration

feat(add wifi confiuration) enable computerless setup for BabyScout on the go

.picowgo

@@ -0,0 +1 @@
+{'info': 'This file is just used to identify a project folder.'}

.vscode/Pico-W-Stub

@@ -0,0 +1 @@
+/home/mschappacher/.config/Code/User/Pico-W-Stub

.vscode/extensions.json

@@ -0,0 +1,7 @@
+{
+    "recommendations": [
+        "ms-python.python",
+        "visualstudioexptteam.vscodeintellicode",
+        "ms-python.vscode-pylance"
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,13 @@
+{
+    "python.linting.enabled": true,
+    "python.analysis.typeshedPaths": [
+        ".vscode/Pico-W-Stub"
+    ],
+    "python.languageServer": "Pylance",
+    "python.analysis.typeCheckingMode": "basic",
+    "python.analysis.extraPaths": [
+        ".vscode/Pico-W-Stub/stubs"
+    ],
+    "picowgo.syncFolder": "firmware",
+    "picowgo.openOnStart": true
+}

firmware/lib/connection.py

@@ -3,49 +3,63 @@
 import utils
 import uos as os
 import network
+import socket
 from pin import onboard_led
 import time
+from request import Request
 
 
-def send_api_request(base_url, path, headers={}, data={}):
-    auth_variables = utils.retrieve_auth_variables(
-        utils.join_path(os.getcwd(), "secrets.json")
-    )["AUTHORIZATION"]
-    if headers:
-        auth_variables.update(headers)
-    if data:
-        data = json.dumps(data)
-        auth_variables["Content-Type"] = "application/json"
-        return json.loads(
-            requests.post(
-                url=base_url + path + "/", headers=auth_variables, data=data
-            ).content
-        )
-    return json.loads(
-        requests.get(url=base_url + path + "/", headers=auth_variables).content
-    )
+class PicoConnection:
+    def __init__(self):
+        self.wlan = self.wireless_client_setup()
+        self.nearby_access_points = self.scan_access_points()
 
+    @staticmethod
+    def wireless_client_setup():
+        wlan = network.WLAN(network.STA_IF)
+        wlan.active(True)
+        return wlan
 
-def connect_to_wifi(wlan_variables):
-    # Attempt to connect to WIFI
-    wlan = network.WLAN(network.STA_IF)
-    wlan.active(True)
-    wlan.connect(wlan_variables["SSID"], wlan_variables["PASSWORD"])
-    print(f"Attempting to connect to {wlan_variables['SSID']}")
-    while not wlan.isconnected():
-        onboard_led()
-        time.sleep(0.5)
-        onboard_led()
-        time.sleep(0.5)
-    print("WLAN Connected")
-    return wlan
+    @staticmethod
+    def access_point_wifi_setup():
+        ap = network.WLAN(network.AP_IF)
+        ap.config(essid="BabyScout", password="BabyBuddy")
+        ap.active(True)
+        print(ap.status())
+        print(ap)
+        return ap
 
+    def connect_to_access_point(self, ssid, password):
+        # Attempt to connect to WIFI
+        if ssid not in self.nearby_access_points:
+            return False
+        self.wlan.active(True)
+        self.wlan.connect(ssid, password)
+        print(f"Attempting to connect to {ssid}")
+        while not self.wlan.isconnected():
+            onboard_led()
+            time.sleep(0.5)
+            onboard_led()
+            time.sleep(0.5)
+        print("WLAN Connected")
+        return True
 
-def test_connection(url):
-    try:
-        status = requests.get(url=url).status_code
-        if status == 200:
-            return True
-        return False
-    except:
-        return False
+    def scan_access_points(self):
+        nearby_access_point_list = []
+        nearby_access_points = self.wlan.scan()
+        for ssid in nearby_access_points:
+            nearby_access_point_list.append(ssid[0].decode("utf-8"))
+        self.wlan.active(False)
+        while "" in nearby_access_point_list:
+            nearby_access_point_list.remove("")
+        print(nearby_access_point_list)
+        return nearby_access_point_list
+
+    def access_point_nearby(self, configured_ssids):
+        matching_ssids = []
+        for ssid in configured_ssids:
+            if ssid in self.nearby_access_points:
+                matching_ssids.append(ssid)
+        if matching_ssids:
+            return matching_ssids[0]
+        return None

firmware/lib/request.py

@@ -0,0 +1,34 @@
+class Request:
+    def __init__(self, request):
+        self.raw = request
+        self.method = None
+        self.path = None
+        self.query_strings = None
+        self.protocol = None
+        self.headers = None
+        self.body = None
+        self.parse_request()
+
+    def parse_request(self):
+        print(self.raw)
+        try:
+            request, self.body = self.raw.split("\r\n\r\n")
+        except ValueError:
+            request = self.raw
+        split_request = request.split("\r\n")
+        request_line = split_request.pop(0)
+        self.method, self.path, self.protocol = request_line.split(" ")
+        if "?" in self.path:
+            query_strings = {}
+            self.path, raw_query = self.path.split("?")
+            raw_query = raw_query.split("&")
+            for query in raw_query:
+                key, value = query.split("=")
+                query_strings[key] = value
+            self.query_strings = query_strings
+
+        headers = {}
+        for header in split_request:
+            key, value = header.split(": ")
+            headers[key] = value
+        self.headers = headers

firmware/lib/scout.py

@@ -1,9 +1,9 @@
+import utils
 import urequests as requests
 import ujson as json
 import uos as os
 import ussl as ssl
 from micropython import *
-from connection import send_api_request
 from pin import onboard_led
 import time
 
@@ -120,3 +120,22 @@ def connect_to_baby_buddy(base_url):
     onboard_led(0)
     print("Connected to BabyBuddy")
     return baby_buddy
+
+
+def send_api_request(base_url, path, headers={}, data={}):
+    auth_variables = utils.retrieve_auth_variables(
+        utils.join_path(os.getcwd(), "../secrets.json")
+    )["AUTHORIZATION"]
+    if headers:
+        auth_variables.update(headers)
+    if data:
+        data = json.dumps(data)
+        auth_variables["Content-Type"] = "application/json"
+        return json.loads(
+            requests.post(
+                url=base_url + path + "/", headers=auth_variables, data=data
+            ).content
+        )
+    return json.loads(
+        requests.get(url=base_url + path + "/", headers=auth_variables).content
+    )

firmware/lib/template.py

@@ -0,0 +1,5 @@
+def render_template(loaded_html, variables):
+    modified_html = loaded_html
+    for key, value in variables.items():
+        modified_html = modified_html.replace("{{ " + key + " }}", value)
+    return modified_html

firmware/lib/utils.py

@@ -2,8 +2,20 @@
 
 
 def retrieve_auth_variables(auth_path):
-    with open(auth_path) as open_auth:
-        return json.load(open_auth)
+    try:
+        with open(auth_path) as open_auth:
+            return json.load(open_auth)
+    except:
+        return None
+
+
+def auth_variables_valid(loaded_auth):
+    ssids = loaded_auth.get("SSIDS_PASSWORD")
+    base_url = loaded_auth.get("BASE_URL")
+    auth = loaded_auth.get("AUTHORIZATION")
+    if not ssids or not base_url or not auth:
+        return False
+    return True
 
 
 def get_base_path(path):

firmware/lib/webpage.py

@@ -0,0 +1,43 @@
+import uos as os
+import ujson as json
+import machine
+import template
+
+
+def load_webpage(location):
+    with open(location, "r") as website:
+        webpage = website.read()
+    return webpage
+
+
+def default_route(*args, **kwargs):
+    options = [
+        {"ssid": "SSID:"},
+        {"password": "Password:"},
+        {"babybuddy": "BabyBuddy URL:"},
+        {"babyauth": "BabyBuddy API Key"}]
+    render = ""
+    for option in options:
+        for key, value in option.items():
+            render += f''' <input type="text" id="{key}" name="{key}" placeholder="{value}"><br><br>'''
+    return template.render_template(load_webpage("webpages/default.html"), {"render": render})
+
+
+def config_route(*args, **kwargs):
+    request = kwargs.get("request")
+    secret_json = {}
+    if request.query_strings:
+        if "secrets.json" in os.listdir():
+            with open("secrets.json", "r+") as secret:
+                secret_json = json.loads(secret.read())
+        with open("secrets.json", "w") as secret:
+            secret_json["SSIDS_PASSWORD"][
+                request.query_strings.get("ssid", "")
+            ] = request.query_strings.get("password", "")
+            if request.query_strings.get("babybuddy"):
+                secret_json["BASE_URL"] = request.query_strings.get("babybuddy")
+            if request.query_strings.get("babyauth"):
+                secret_json["AUTHORIZATION"]["Authorization"] = request.query_strings.get("babyauth")
+            secret.write(json.dumps(secret_json))
+    machine.reset()
+

firmware/lib/webrouter.py

@@ -0,0 +1,63 @@
+from request import Request
+import socket
+import uos as os
+
+
+class WebRouter:
+    def __init__(self, ip, port, default, static_location="static"):
+        self.routes = {"default": default}
+        self.socket = self.open_socket(ip, port)
+        self.static_location = static_location
+        self.static_files = []
+        self.add_static()
+
+    def route(self, path):
+        def router(function):
+            def wrapper(*args, **kwargs):
+                self.routes[path] = function
+
+            return wrapper
+
+        return router
+
+    def add_static(self):
+        static_files = os.listdir(self.static_location)
+        if static_files:
+            for file in static_files:
+                self.static_files.append(f"/{file}")
+
+    @staticmethod
+    def open_socket(ip, port):
+        address = (ip, port)
+        new_socket = socket.socket()
+        new_socket.bind((ip, port))
+        new_socket.listen(1)
+        return new_socket
+
+    @staticmethod
+    def determine_mimetype(path):
+        mimetype = "text/html"
+        if path.endswith('.js'):
+            mimetype = "text/javascript"
+        if path.endswith('.css'):
+            mimetype = "text/css"
+
+
+    def serve(self):
+        while True:
+            client = self.socket.accept()[0]
+            request = client.recv(1024)
+            request = Request(request.decode("utf-8"))
+            path = request.path
+            header = 'HTTP/1.1 200 OK\n'
+            mimetype = self.determine_mimetype(path)
+            if path in self.routes.keys():
+                webpage = self.routes[path](request=request)
+            elif path in self.static_files:
+                with open(f"{self.static_location}/{path}", "rb") as static:
+                    webpage = static.read()
+            else:
+                webpage = self.routes["default"](request=request)
+            header += f"Content-Type: {mimetype}\n\n"
+            client.send(webpage)
+            client.close()