Skip to content

Latest commit

 

History

History
248 lines (187 loc) · 7.45 KB

File metadata and controls

248 lines (187 loc) · 7.45 KB

Your current structure looks well-organized and covers most aspects of your PicoW LED Microcontroller project. However, there are a few areas you might consider refactoring or improving for better maintainability and scalability:

  1. Separate Configuration from Code:

    • The secrets.py file is currently used to store sensitive information like Wi-Fi credentials. Consider moving other configurable settings here as well. For example, default port numbers, firmware download links, etc.
  2. Use a Config Management Library:

    • Integrate a configuration management library like uconfig or another similar solution to handle configurations dynamically. This will make it easier to change settings without recompiling the code.
  3. Modularize Commands and Logic:

    • Ensure that commands and logic are modularized into separate modules, if they're not already. For instance, you could have a utils module for utility functions and constants.
  4. Error Handling and Logging:

    • Enhance error handling and logging. Use Python's built-in logging library to provide more detailed logs that can be useful for debugging in production.
  5. Unit Tests and Integration Tests:

    • Write unit tests for your commands and logic modules. This will help ensure that your code works as expected and catch regressions early.
  6. Documentation and README:

    • Ensure your README.md is up-to-date with installation instructions, configuration details, and any other important information for users.

Here's a refined version of your project structure with some suggested improvements:

picow-led-microcontroller/
├── .gitignore
├── CHANGELOG.md
├── README.md
├── src/
│   ├── commands.py
│   ├── constants.py
│   ├── gpio.py
│   ├── main.py
│   ├── picozero.py
│   └── utils.py  # New directory for utility functions and constants
└── secrets.py

Example Refactoring

  1. Add a utils Directory:
# src/utils/config_manager.py

import json
from typing import Any, Dict

class ConfigManager:
    def __init__(self, config_path: str):
        self.config_path = config_path
        self.config_data = self.load_config()

    def load_config(self) -> Dict[str, Any]:
        try:
            with open(self.config_path, 'r') as file:
                return json.load(file)
        except FileNotFoundError:
            return {}

    def save_config(self, data: Dict[str, Any]) -> None:
        with open(self.config_path, 'w') as file:
            json.dump(data, file, indent=4)

    def get(self, key: str, default: Any = None) -> Any:
        return self.config_data.get(key, default)

    def set(self, key: str, value: Any) -> None:
        self.config_data[key] = value
        self.save_config(self.config_data)
  1. Use ConfigManager in Your Main Script:
# src/main.py

import json
from socket import SO_REUSEADDR, SOL_SOCKET, socket
from sys import stderr
from time import sleep

import machine
import network

from commands import COMMANDS
from constants import END_BYTE, HOST, PORT
from secrets import PASS, SSID
from utils.config_manager import ConfigManager

config_path = 'config.json'
config_manager = ConfigManager(config_path)

def main():
    setup()

    server = socket()
    server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen(5)

    try:
        while True:
            print(
                f'[INFO ] Waiting for client on "{HOST}:{PORT}"',
                file=stderr,
            )
            client, address = server.accept()
            print(f'[INFO ] Connected client from "{address}"', file=stderr)

            try:
                handleClient(client)
            except Exception as e:
                print(f"[WARN ] Handle client exception: {str(e)}")
                client.close()

    finally:
        disable_led()


def setup():
    disable_led()

    sta_if = network.WLAN(network.STA_IF)

    if not sta_if.isconnected():
        sta_if.active(True)

        print("[DEBUG] Try to connect...", file=stderr)
        sta_if.connect(SSID, PASS)

        retry_count = 0
        while not sta_if.isconnected():
            if retry_count > 20:
                print("[ERROR] Connection failed, retry...")
                return setup()

            retry_count += 1

            print("[DEBUG] ...wait for connection...", file=stderr)
            sleep(0.5)

    enable_led()
    print(f"[DEBUG] ...ifconfig={sta_if.ifconfig()}", file=stderr)


def handleClient(client):
    while True:
        data = bytes()
        while True:
            chunk = client.recv(1)
            if chunk == END_BYTE:
                break

            if chunk:
                data += chunk
            else:
                break

        if len(data) == 0:
            client.close()
            return

        try:
            data = json.loads(data.strip())

        except Exception as ex:
            print(f"[WARN ] Convert data to JSON failed: {ex}", file=stderr)
            client.close()
            return

        try:
            request = {
                "id": data.get("id", 0),
                "group": data["group"],
                "type": data["type"],
                "command": data["command"],
                "args": data.get("args", []),
            }

            try:
                print(
                    f"[DEBUG] Run requested command: {request['group']} "
                    + f"{request['type']} {request['command']} "
                    + f"{request['args']}"
                )
                run(client, request)

            except Exception as ex:
                print(f"[ERROR] run failed: {ex}", file=stderr)
                client.close()
                return

        except Exception as ex:
            print(f"[WARN ] Invalid client request: {ex}", file=stderr)
            client.close()
            return


def run(client, request):
    response = {
        "id": request["id"],
        "error": None,
        "data": None,
    }

    try:
        if request["group"] not in COMMANDS:
            response["error"] = f'command group "{request["group"]}" not found'

        if request["type"] not in COMMANDS[request["group"]]:
            response["error"] = f'command type "{request["type"]}" not found'

        if request["command"] not in COMMANDS[request["group"]][request["type"]]:
            response["error"] = f'command "{request["command"]}" not found'

        if response["error"] is None:
            response["data"] = COMMANDS[request["group"]][request["type"]][
                request["command"]
            ](*request["args"])

    except AssertionError:
        message = "wrong args"
        print("[ERROR] " + message)
        response["error"] = message

    except Exception as e:
        message = f"command failed to run: {str(e)} (type: {type(e).__name__})"
        print("[ERROR] " + message + ": " + str(e))
        response["error"] = message

    if response["id"] != -1:
        data = json.dumps(response)
        client.settimeout(0.5)
        client.send(data.encode() + END_BYTE)
        client.settimeout(None)


def enable_led():
    led = machine.Pin("LED", machine.Pin.OUT)
    led.on()


def disable_led():
    led = machine.Pin("LED", machine.Pin.OUT)
    led.off()

This refactoring introduces a ConfigManager class to handle configuration loading and saving, making your main script cleaner and more modular. You can further extend this by adding more utility functions in the utils directory as needed.