chore: fix pylint warnings for boot.py, code.py and logs.py

This commit is contained in:
Himadri Bhattacharjee
2023-06-09 09:42:46 +05:30
parent 5f91021b82
commit ead3afeaaf
6 changed files with 121 additions and 43 deletions

View File

@@ -1,14 +1,29 @@
"""
Builder script to compile .py files to .mpy bytecode using mpy-cross
"""
import glob
from subprocess import PIPE, Popen
from os import listdir, makedirs
from os.path import join, splitext
from shutil import copytree, copy, rmtree
from errno import ENOTDIR
SRC = 'src'
DST = 'build'
def cp(src, dst):
def recursive_copy(src: str, dst: str):
"""
Copy a file or directory from src to dst.
Parameters:
src (str): The path of the source file or directory.
dst (str): The path of the destination file or directory.
Returns:
None
"""
try:
copytree(src, dst)
except OSError as exc:
@@ -18,28 +33,52 @@ def cp(src, dst):
raise
def to_compile(s: str):
name, ext = splitext(s)
if name not in ("code", "boot") and ext == ".py":
return name
def to_compile(name: str) -> str:
"""
Check if a given file is a Python source file that needs to be compiled.
Parameters:
name (str): The name of the file.
Returns:
str: The name of the file without the extension if it need compilation,
otherwise None.
"""
base, ext = splitext(name)
if base not in ("code", "boot") and ext == ".py":
return base
return None
rmtree(DST, ignore_errors=True)
makedirs(DST, exist_ok=True)
mpy_cross_bin = join(".", glob.glob("mpy-cross.static*")[0])
def main():
"""
Use mpy-cross to compile .py files to .mpy bytecode
"""
for entry in listdir(SRC):
src_path = join(SRC, entry)
if name := to_compile(entry):
Popen([
mpy_cross_bin,
"-o",
join(DST, f'{name}.mpy'),
src_path,
],
stdout=PIPE,
).communicate()
else:
dst_path = join(DST, entry)
cp(src_path, dst_path)
# Remove the build directory if it exists, then create it again
rmtree(DST, ignore_errors=True)
makedirs(DST, exist_ok=True)
# Find the path of the mpy-cross binary
mpy_cross_bin = join(".", glob.glob("mpy-cross.static*")[0])
# Process each entry in the source directory
for entry in listdir(SRC):
src_path = join(SRC, entry)
# If the entry is a Python source file that needs to be compiled
if name := to_compile(entry):
# Compile the file using mpy-cross
with Popen(
[mpy_cross_bin, "-o", join(DST, f"{name}.mpy"), src_path],
stdout=PIPE,
) as process:
process.communicate()
else:
# Copy the file or directory to the build directory
dst_path = join(DST, entry)
recursive_copy(src_path, dst_path)
if __name__ == "__main__":
main()

View File

@@ -1,12 +1,13 @@
import logs
import os
import json
from ducky import run_script_file, run_script
import os
import logs
from ducky import run_script, run_script_file
def create(path, contents=b""):
with open(path, "wb") as h:
h.write(contents)
with open(path, "wb") as file:
file.write(contents)
def handle(body, response):
@@ -22,8 +23,8 @@ def handle(body, response):
filename = body.get("filename")
path = f"payloads/{filename}"
if action == "load":
with open(path) as h:
response.send(json.dumps({"contents": h.read()}))
with open(path) as file:
response.send(json.dumps({"contents": file.read()}))
elif action == "store":
create(path, body["contents"].encode())
elif action == "delete":

View File

@@ -1,3 +1,9 @@
"""
Disable concurrent write protections on the storage.
This is especially necessary for dev builds since we need to change the files
on the board for hot reloading.
"""
import storage
storage.remount("/", readonly=False, disable_concurrent_write_protection=True)

View File

@@ -1,18 +1,31 @@
import microcontroller
import wifi
import socketpool
"""
The entrypoint for our circuitpython board.
"""
import asyncio
import os
import json
from api import handle
from adafruit_httpserver.server import HTTPServer
from adafruit_httpserver.request import HTTPRequest
from adafruit_httpserver.response import HTTPResponse
import os
import microcontroller
import socketpool
import wifi
from adafruit_httpserver.methods import HTTPMethod
from adafruit_httpserver.mime_type import MIMEType
from adafruit_httpserver.request import HTTPRequest
from adafruit_httpserver.response import HTTPResponse
from adafruit_httpserver.server import HTTPServer
from api import handle
async def main():
"""
Begin a wifi access point defined by the SSID and PASSWORD environment
variables.
Spawn a socketpool on this interface.
Serve the web interface over this socketpool indefinitely using an HTTP
server.
"""
wifi.radio.start_ap(ssid=os.getenv("SSID"), password=os.getenv("PASSWORD"))
pool = socketpool.SocketPool(wifi.radio)
server = HTTPServer(pool)
@@ -28,7 +41,7 @@ async def main():
response.send_file("static/main.css")
@server.route("/script.js")
def js(request: HTTPRequest):
def javascript(request: HTTPRequest):
with HTTPResponse(request, content_type=MIMEType.TYPE_JS) as response:
response.send_file("static/script.js")
@@ -39,6 +52,7 @@ async def main():
server.serve_forever(str(wifi.radio.ipv4_address_ap))
if __name__ == "__main__":
try:
asyncio.run(main())

View File

@@ -1,18 +1,20 @@
import time
import usb_hid
from adafruit_hid.keyboard import Keyboard
# comment out these lines for non_US keyboards
from adafruit_hid.keyboard_layout_us import KeyboardLayoutUS as KeyboardLayout
from adafruit_hid.keycode import Keycode
from board import LED
from logs import info, warn
# uncomment these lines for non_US keyboards
# replace LANG with appropriate language
# from keyboard_layout_win_LANG import KeyboardLayout
# from keycode_win_LANG import Keycode
import time
from board import LED
from logs import info, warn
kbd = Keyboard(usb_hid.devices)
layout = KeyboardLayout(kbd)
@@ -86,5 +88,5 @@ def run_script_file(path: str):
try:
with open(path, "r", encoding="utf-8") as handle:
run_script(handle.read())
except OSError as e:
warn(f"unable to open file {path}: {e}")
except OSError as error:
warn(f"unable to open file {path}: {error}")

View File

@@ -1,16 +1,32 @@
"""
A very bare-bones logging implementation
for the bottom pane of the Web UI.
"""
import json
logs = []
def consume() -> str:
"""
Convert all the log entries from the module's global mutable
list to json return them, clearing the list after the dump.
"""
dump = json.dumps(logs)
logs.clear()
return dump
def info(message: str):
"""
Add a log entry with the message prepended with the info marker
"""
logs.append("info: " + message)
def warn(message: str):
"""
Add a log entry with the message prepended with the warning marker
"""
logs.append("warning: " + message)