feat: added gpio control for led

Makefile

@@ -1,3 +1,2 @@
 run:
-	chmod +x robot_pi_service/robot_pi_service.py
 	python3 robot_pi_service/robot_pi_service.py

README.md

@@ -2,6 +2,11 @@
 
 **Robot-pi-service** - сервис для управления роботом на orange pi, работающий на linux Armbian.
 
+**Для работы с GPIO:**
+
+1. `chmod +x gpio_setup.sh`
+2. `./gpio_setup.sh`
+
 **Запуск сервиса (как проекта python3)** - `make run`
 
 <details>

gpio_setup.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+chmod -R +x robot_pi_service/
+sudo groupadd gpio
+sudo usermod -aG gpio $USER
+echo 'SUBSYSTEM=="gpio", KERNEL=="gpiochip[0-9]*", GROUP="gpio", MODE="0660"' | sudo tee /etc/udev/rules.d/99-gpio.rules
+sudo udevadm control --reload-rules
+sudo udevadm trigger

requirements.txt

@@ -1,2 +1,3 @@
 websockets==15.0.1
 pydantic-settings==2.9.1
+gpiod==2.3.0

robot_pi_service/gpio_control.py

@@ -0,0 +1,34 @@
+from gpiod import request_lines, LineSettings
+from gpiod.line import Direction, Value
+
+
+class LedLineGpio:
+    """Класс для управления динией gpio для LED"""
+    
+    def __init__(self, line: int) -> None:
+        self.line = line
+        self._request = request_lines(
+            '/dev/gpiochip0',
+            consumer='led-blinker',
+            config={
+                line: LineSettings(
+                    direction=Direction.OUTPUT,
+                    output_value=Value.INACTIVE
+                )
+            }
+        )
+
+    def on(self) -> None:
+        """Метод для включения LED"""
+        
+        self._request.set_value(self.line, Value.ACTIVE)
+
+    def off(self) -> None:
+        """Метод для выключения LED"""
+        
+        self._request.set_value(self.line, Value.INACTIVE)
+
+    def close(self) -> None:
+        """Метод для освобождения ресурса"""
+        
+        self._request.release()

robot_pi_service/robot_pi_service.py

@@ -1,8 +1,10 @@
 import asyncio
 from websockets import serve, exceptions
 from websockets.legacy.server import WebSocketServerProtocol
+from json import loads, JSONDecodeError
 
 from settings import settings
+from gpio_control import LedLineGpio
 
 
 async def robot_control_gpio(websocket: WebSocketServerProtocol):
@@ -11,16 +13,28 @@ async def robot_control_gpio(websocket: WebSocketServerProtocol):
     """
 
     try:
+        led_line_gpio = LedLineGpio(line=6)
+        action = None
+
         while True:
             try:
                 command = await asyncio.wait_for(websocket.recv(), timeout=30.0)
+                data = loads(command)
             except asyncio.TimeoutError as err:
                 print('Таймаут ожидания команды от клиента')
                 continue  # продолжение цикла, чтобы не закрывать соединение
+            except JSONDecodeError:
+                print("Ошибка при декодировании JSON")
+                continue
             
-            match command:
+            match next(iter(data), None):
                 case settings.commands_robot.forward:
+                    if data.get(settings.commands_robot.forward):
                         action = 'Робот едет вперёд'
+                        led_line_gpio.on()
+                    else:
+                        action = 'Команда вперёд стоп'
+                        led_line_gpio.off()
                     print(action)
                     await websocket.send(message=action)
                 case settings.commands_robot.backward:
@@ -45,11 +59,15 @@ async def robot_control_gpio(websocket: WebSocketServerProtocol):
         message_err = f'{err.__class__.__name__}: {err}'
         print(message_err)
         await websocket.send(message=message_err)
+    finally:
+        led_line_gpio.close()
 
 
 async def start():
     """Асинхронная функция запуска вебсокета и бесконечного цикла"""
 
+    print('Старт сервиса робота для приёма команд.')
+
     async with serve(handler=robot_control_gpio, host=settings.websocket_host, port=settings.websocket_port):
         # бесконечный цикл
         await asyncio.Future()
@@ -61,7 +79,7 @@ def run_app():
     try:
         asyncio.run(start())
     except KeyboardInterrupt:
-        pass
+        print('Выключение сервиса робота для приёма команд.')
 
 
 if __name__ == '__main__':