reformat with black

config_server.py

@@ -13,16 +13,17 @@
 
 
 class DreamPiConfigurationService(BaseHTTPRequestHandler):
-
     def _get_post_data(self) -> Dict[str, List[str]]:
-        ctype, pdict = cgi.parse_header(self.headers['content-type'])
+        ctype, pdict = cgi.parse_header(self.headers["content-type"])
 
-        if ctype == 'multipart/form-data':
-            pdict = { k: v.encode() for k, v in pdict.items() }
+        if ctype == "multipart/form-data":
+            pdict = {k: v.encode() for k, v in pdict.items()}
             postvars = cgi.parse_multipart(self.rfile, pdict)
-        elif ctype == 'application/x-www-form-urlencoded':
-            length = int(self.headers['content-length'])
-            postvars = cgi.parse_qs(self.rfile.read(length).decode(), keep_blank_values=True)
+        elif ctype == "application/x-www-form-urlencoded":
+            length = int(self.headers["content-length"])
+            postvars = cgi.parse_qs(
+                self.rfile.read(length).decode(), keep_blank_values=True
+            )
         else:
             postvars = {}
 
@@ -39,11 +40,11 @@ def do_GET(self):
             with open(CONFIGURATION_FILE, "r") as f:
                 enabled_state = json.loads(f.read())["enabled"]
 
-        self.wfile.write(json.dumps({
-            "mac_address": hash_mac_address(),
-            "is_enabled": enabled_state
-        }).encode())
-
+        self.wfile.write(
+            json.dumps(
+                {"mac_address": hash_mac_address(), "is_enabled": enabled_state}
+            ).encode()
+        )
 
     def do_POST(self):
         enabled_state = True
@@ -54,30 +55,33 @@ def do_POST(self):
         self.end_headers()
 
         post_data = self._get_post_data()
-        if 'disable' in post_data:
+        if "disable" in post_data:
             enabled_state = False
         else:
             enabled_state = True
 
         with open(CONFIGURATION_FILE, "w") as f:
             f.write(json.dumps({"enabled": enabled_state}))
 
-        self.wfile.write(json.dumps({
-            "mac_address": hash_mac_address(),
-            "is_enabled": enabled_state
-        }).encode())
+        self.wfile.write(
+            json.dumps(
+                {"mac_address": hash_mac_address(), "is_enabled": enabled_state}
+            ).encode()
+        )
 
 
 server = None
 thread = None
 
+
 def start():
     global server
     global thread
-    server = HTTPServer(('0.0.0.0', 1998), DreamPiConfigurationService)
+    server = HTTPServer(("0.0.0.0", 1998), DreamPiConfigurationService)
     thread = threading.Thread(target=server.serve_forever)
     thread.start()
 
+
 def stop():
     global server
     global thread

dcnow.py

@@ -10,13 +10,13 @@
 import logging.handlers
 import urllib.request
 import urllib.parse
-import sh # type: ignore - sh module is dynamic
+import sh  # type: ignore - sh module is dynamic
 from typing import List, Optional
 from hashlib import sha256
 
 from uuid import getnode as get_mac
 
-logger = logging.getLogger('dcnow')
+logger = logging.getLogger("dcnow")
 
 API_ROOT = "https://dcnow-2016.appspot.com"
 UPDATE_END_POINT = "/api/update/{mac_address}/"
@@ -28,7 +28,9 @@
 
 def hash_mac_address():
     mac = get_mac()
-    return sha256(':'.join(("%012X" % mac)[i:i+2] for i in range(0, 12, 2)).encode()).hexdigest()
+    return sha256(
+        ":".join(("%012X" % mac)[i : i + 2] for i in range(0, 12, 2)).encode()
+    ).hexdigest()
 
 
 class DreamcastNowThread(threading.Thread):
@@ -42,7 +44,9 @@ def post_update():
             if not self._service.enabled:
                 return
 
-            lines: List[str] = list(sh.tail("/var/log/syslog", "-n", "10", _iter=True)) # type: ignore - sh has dynamic members
+            lines: List[str] = list(
+                sh.tail("/var/log/syslog", "-n", "10", _iter=True)
+            )  # type: ignore - sh has dynamic members
             dns_query = None
             for line in lines[::-1]:
                 line: str = line
@@ -52,21 +56,25 @@ def post_update():
 
                 if "query[A]" in line:
                     # We did a DNS lookup, what was it?
-                    remainder = line[line.find("query[A]") + len("query[A]"):].strip()
+                    remainder = line[line.find("query[A]") + len("query[A]") :].strip()
                     domain = remainder.split(" ", 1)[0].strip()
                     dns_query = sha256(domain.encode()).hexdigest()
                     break
 
-            user_agent = 'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT), Dreamcast Now'
-            header = { 'User-Agent' : user_agent }
+            user_agent = "Mozilla/4.0 (compatible; MSIE 5.5; Windows NT), Dreamcast Now"
+            header = {"User-Agent": user_agent}
             mac_address = self._service.mac_address_hash
             data = {}
             if dns_query:
                 data["dns_query"] = dns_query
 
             data = urllib.parse.urlencode(data).encode()
-            req = urllib.request.Request(API_ROOT + UPDATE_END_POINT.format(mac_address=mac_address), data, header)
-            urllib.request.urlopen(req) # Send POST update
+            req = urllib.request.Request(
+                API_ROOT + UPDATE_END_POINT.format(mac_address=mac_address),
+                data,
+                header,
+            )
+            urllib.request.urlopen(req)  # Send POST update
 
         while self._running:
             try:
@@ -88,8 +96,10 @@ def __init__(self):
         self.reload_settings()
 
         logger.setLevel(logging.INFO)
-        syslog_handler = logging.handlers.SysLogHandler(address='/dev/log')
-        syslog_handler.setFormatter(logging.Formatter('%(name)s[%(process)d]: %(levelname)s %(message)s'))
+        syslog_handler = logging.handlers.SysLogHandler(address="/dev/log")
+        syslog_handler.setFormatter(
+            logging.Formatter("%(name)s[%(process)d]: %(levelname)s %(message)s")
+        )
         logger.addHandler(syslog_handler)
 
     def update_mac_address(self):
@@ -116,11 +126,11 @@ def go_offline(self):
         if self._thread is not None:
             self._thread.stop()
             self._thread = None
-    
+
     @property
     def enabled(self):
         return self._enabled
-    
+
     @property
     def mac_address_hash(self):
         return self._mac_address_hash