nucleo-wb55-playground/backend/command.py

import atexit
import logging
import os
import time
from dataclasses import dataclass
from enum import Enum
from multiprocessing import Process
from multiprocessing import Queue
from typing import Optional

from flask import Response
from flask import blueprints
from flask import request
from flask_api import status
from serial import Serial

from .container import get_container
from .container import initialize_container

bp = blueprints.Blueprint("command", __name__)

_process: Optional[Process] = None
_queue: Queue = Queue()
_logger = logging.getLogger(__name__)


@bp.route("/command", methods=["POST", "GET"])
def command():
    logger = logging.getLogger("test")
    command_id = request.args.get("command-id")

    if _queue is not None and command_id is not None:
        logger.info(f"put in queue: {command_id}")
        _queue.put(command_id)

    return Response(status=status.HTTP_200_OK)


def _end_running_process():
    if _process is not None:
        _process.kill()


def worker_process(
    queue: Queue,
    config,
    # container: Container
):
    logging.basicConfig(
        level=logging.DEBUG,
        format="[%(asctime)s] [%(name)-20s] [%(levelname)-8s] --- %(message)s",
    )
    logger = logging.getLogger("worker_process")

    initialize_container(config)
    container = get_container()

    counter = 1
    with container.serial() as serial:
        while True:
            logger.info(f"Ping {counter} {os.getpid()}")
            counter += 1
            time.sleep(0.5)
            receive_and_log(
                serial=serial,
                header_size=container.config.header_size(),
            )

            if queue is not None:
                while not queue.empty():
                    command_id = queue.get()
                    logger.debug(f"device_id {container.config.device_id()}")
                    logger.info(f"would execute command: {command_id}")


class CommandId(Enum):
    command_none = 0
    command_log = 0xFF


@dataclass
class CommandMeta:
    command_id: CommandId
    data_length: int


@dataclass
class LogCommand:
    """Command ID: command_log"""

    level: int
    message: str

    HEADER_SIZE = 1  # log level

    def __init__(
        self,
        data: bytes,
    ) -> None:
        self._logger = logging.getLogger(self.__class__.__name__)
        self.received_logger = logging.getLogger("stm32wb55")
        self.received_logger.setLevel(logging.DEBUG)

        self._logger.setLevel(logging.INFO)

        level = int(data[0])
        self._logger.debug(f"level: {level}")

        message = data[self.HEADER_SIZE :]
        self._logger.debug("Message: " + str(message))

        self.level = level
        self.message = message.decode()

    def execute(self):
        self.received_logger.log(level=self.level, msg=self.message)


def receive_and_log(
    serial: Serial,
    header_size: int,
):
    logger = logging.getLogger("receive_and_log")
    logger.setLevel(logging.INFO)

    bytes_read = serial.read(serial.in_waiting)

    while bytes_read:

        logger.debug(f"bytes: {bytes_read.hex()}")

        command_id = int(bytes_read[0])
        logger.debug(f"command_id: {command_id}")

        data_length = (int(bytes_read[1]) << 4) + int(bytes_read[2])
        logger.debug(f"data_length: {data_length}")

        meta = CommandMeta(
            command_id=CommandId(command_id),
            data_length=data_length,
        )

        if meta.command_id == CommandId.command_log:
            command = LogCommand(
                data=bytes_read[header_size : header_size + meta.data_length],
            )
            command.execute()
        else:
            return

        stop_byte = bytes_read[header_size + meta.data_length]
        logger.debug(f"stop_byte: {stop_byte}")
        assert stop_byte == 0xFF

        bytes_read = bytes_read[header_size + meta.data_length + 1 :]


def start_backgroup_process():
    _logger.warning("start_backgroup_process called")
    global _process

    _process = Process(
        target=worker_process,
        args=(
            _queue,
            get_container().config(),
        ),
    )
    _process.start()
    atexit.register(_end_running_process)