WebserverRequest & WebserverResponse refactor to HTTPRequest, HTTPResponse based on HTTPHeader object

Author: JannickOste

src/Device.py

@@ -0,0 +1,30 @@
+class __Enviroment:
+ @property
+ def PUBLIC_FOLDER(self):
+ return "public"
+ 
+ @property
+ def POWER_LED_PIN(self):
+ return -1
+ 
+ @property
+ def WIFI_LED_PIN(self):
+ return -2
+ 
+ @property
+ def ADHOC_LED_PIN(self):
+ return -3
+ 
+ @property
+ def DEBUG_MODE(self):
+ return True
+ 
+ @property
+ def CAMERA_INTERVAL(self):
+ mins = 15
+ ms = (mins)*60*1000
+ sec = (ms / 1000)
+ 
+ return int(sec)
+ 
+Enviroment = __Enviroment()

src/Enviroment.py

@@ -1,68 +1,48 @@
 
 from net.www.Webserver import Webserver
+from Enviroment import Enviroment
+from utils.Logger import Logger
+from utils.Device import Device
 import uasyncio
 import machine
-import gc
-
-gc.collect()
-
 
 www = None
 
-
+# https://github.com/lemariva/micropython-camera-driver
 def __init():
  """
- Devicie initializiation 
+ Device initializiation 
  """
- # Gebruik globale variabele niet enkel in deze scope.
+ 
  global www
- # Initalizeer noodzakelijke componenten 
  www = Webserver(port=80, maxClients=1)
- # Maakt WLAN en ADHOC connectie aan (Wifi/AccessPoint)
- 
+
 async def __main():
  """
  Main device execution function
  """
- # Gebruik globale variabele niet enkel in deze scope.
  global www
 
  loop = uasyncio.get_event_loop()
- 
- # Voeg de webserver start functie als task toe aan asyncio voor uit te voeren.
- loop.create_task(www.start())
- loop.create_task(__update())
- 
- loop.run_forever()
- 
 
-async def __update(): 
- """
- Reserved for device updates, wifi checks, etc...
- """
+ await loop.create_task(Device.update())
+ await loop.create_task(www.start())
 
- wifi_pin = machine.Pin(16, machine.Pin.OUT)
- adhoc_pin = machine.Pin(17, machine.Pin.OUT)
- while True:
- if Wifi.adhoc_ip_adress is not None:
- adhoc_pin.value(1)
- if Wifi.wifi_connected:
- wifi_pin.value(1)
+ loop.run_forever()
 
- await uasyncio.sleep_ms(1000)
 
 # Main entry point, indien python interpeter naar dit bestand is gericht.
 if __name__ == "__main__":
- power_pin = machine.Pin(0, machine.Pin.OUT)
- power_pin.value(1)
-
  from net.Wifi import Wifi 
  try:
  __init()
  uasyncio.run(__main()) # Start main functie met async handler "uasyncio" => asyncio alternatief micropython.
-
+ 
  except OSError as e:
- if e.errno == 98:
- machine.reset()
- 
- print(e)
+ Logger.error("main", f"An exception occured: {e}")
+ 
+ if Enviroment.DEBUG_MODE:
+ if e.errno == 98:
+ machine.reset()
+ else:
+ raise(e)

src/Logger.py

@@ -1,59 +1,55 @@
- 
-from re import search as searchPattern
-
-from components.scripts.Redirect import Redirect
-from components.layout.forms.WifiSelector import WifiSelector
 from net.www.WebserverRoute import WebserverRoute
+from components.layout.forms.WifiSelector import WifiSelector
 from net.Wifi import Wifi
-
+from utils.Device import Device
+from net.WifiConfig import WifiConfig
 
 class Routes:
  @WebserverRoute.get("/")
  def index(request, response):
-  return response.send(Redirect("/wifi", 0))
- return response.render("/index")
-
+ 
+ return response.getFile("/index") 
+ 
  @WebserverRoute.get("/wifi")
- def wifi_get(request, response):
- return response.render("/wifi", **{
- "SSID_SELECTOR": WifiSelector(*[network["ssid"] for network in Wifi.networks]), # todo: WLAN netwerk scan => Arg parse erin
- "MESSAGE": "Failed to connect to wifi, please connect to a nearby access point"
+ def wifi(request, response):
+ response.setParams(**{
+ "MESSAGE": "Already connected to wifi network" if Wifi.wifi_connected else "No wifi network specified, or failed to connect",
+ "SSID_SELECTOR": WifiSelector(*[network["ssid"] for network in Wifi.networks])
  })
-
- @WebserverRoute.post("/wifi")
- def wifi_post(request, response):
- params = request.PARAMS
- message = """
- <b>Attempting to connect to {ssid}.. page will reload in a second.</b>
- {redirect}
- """.format(ssid=" ".join(params["ssid"].split("+")), redirect=Redirect("/", 5000))
- 
- if not len(request.PARAMS):
- message = "Wifi not connected, please connect to a wifi access point."
- elif not all(i in params.keys() for i in ["ssid", "password"]):
- message = "Wifi SSID or password was not set"
 
- return response.render("/wifi", **{
- "SSID_SELECTOR": WifiSelector(*[network["ssid"] for network in Wifi.networks]), # todo: WLAN netwerk scan => Arg parse erin
- "MESSAGE": message
+ return response.getFile("/wifi") 
+ 
+ @WebserverRoute.post("/wifi")
+ def wifi(request, response):
+ print(response.parameters)
+ response.setParams(**{
+ "MESSAGE": "Already connected to wifi network" if Wifi.wifi_connected else "No wifi network specified, or failed to connect",
+ "SSID_SELECTOR": WifiSelector(*[network["ssid"] for network in Wifi.networks])
  })
-
+ if all(key in response.parameters.keys() for key in ["ssid", "password"]):
+ WifiConfig.wlan_ssid = response.parameters["ssid"]
+ WifiConfig.wlan_password = response.parameters["password"]
+ Device.hard_reset()
+ 
+ return response.getFile("/wifi")
+ 
  @WebserverRoute.get("/change_adhoc")
- def adhoc_set_get(request, response):
- return response.render("/change_adhoc", **{
- "MESSAGE": "Failed to connect to wifi, please connect to a nearby access point"
+ def adhoc(request, response):
+ response.setParams(**{
+ "MESSAGE": ""
  })
-
+ return response.getFile("/change_adhoc") 
+ 
  @WebserverRoute.post("/change_adhoc")
- def adhoc_set_post(request, response):
- params = request.PARAMS
-
- message = "<b>Assigning new access point password.</b>"
- if not len(request.PARAMS) or not all(i in params.keys() for i in ["password", "password_verify"]):
-  message = "Invalid request"
- elif params["password"] != params["password_verify"]:
- message = "Passwords did not match"
-
- return response.render("/change_adhoc", **{
-  "MESSAGE": message
-  })
+ def adhoc(request, response):
+ if any([i not in response.parameters.keys() for i in ["password_verify", "password"]]):
+ response.setParams(MESSAGE="Invalid request")
+ elif response.parameters["password"] != response.parameters["password_verify"]:
+  response.setParams(MESSAGE="Password did not match")
+ else:
+  response.setParams(MESSAGE="ADHOC password changed succesfully")
+ Device.hard_reset() 
+ 
+ return response.getFile("/change_adhoc") 
+ 
+ 

src/Main.py

@@ -0,0 +1,22 @@
+import gc
+from machine import Pin
+from Enviroment import Enviroment
+from utils.Logger import Logger
+
+# Enable garbage collection
+gc.collect()
+
+# Enable power out
+try:
+ power_pin = Pin(Enviroment.POWER_LED_PIN, Pin.OUT)
+ power_pin.value(1)
+except ValueError as e:
+ if e.errno == "invalid pin":
+ Logger.error("boot", "Failed to setup power pin: (Invalid enviroment -> POWER_LED_PIN value)")
+ else:
+ if Enviroment.DEBUG_MODE:
+ raise(e)
+ else:
+ Logger.error("boot", f"Failed to setup power pin unkown error: {e}")
+
+from net.Wifi import Wifi