.env.exaple

@@ -0,0 +1,21 @@
+# Настройки для websocket
+WEBSOCKET_HOST=хост
+WEBSOCKET_PORT=порт
+
+# Команды
+FORWARD=вперёд
+BACKWARD=назад
+LEFT=влево 
+RIGHT=вправо
+STOP=стоп
+
+# Настройки для линий GPIO
+GPIO_PATH=путь до gpio
+LED_LINE=номер линии
+LED_CONSUMER=идентификатор клиента
+LEFT_MOTOR_LINE_IN1=номер линии для левого (вращение вперёд)
+LEFT_MOTOR_LINE_IN2=номер линии для левого (вращение назад)
+LEFT_MOTOR_CONSUMER=идентификатор клиента для левого
+RIGHT_MOTOR_LINE_IN1=номер линии для правого (вращение вперёд)
+RIGHT_MOTOR_LINE_IN2=номер линии для правого (вращение назад)
+RIGHT_MOTOR_CONSUMER=идентификатор клиента для правого

robot_pi_service/gpio_control.py

@@ -1,17 +1,19 @@
 from gpiod import request_lines, LineSettings
 from gpiod.line import Direction, Value
 
+from settings import settings
+
 
 class LedLineGpio:
-    """Класс для управления динией gpio для LED"""
+    """Класс для управления линией gpio для LED"""
     
-    def __init__(self, line: int) -> None:
+    def __init__(self, line: int, gpio_path: str, consumer: str) -> None:
         self.line = line
         self._request = request_lines(
-            '/dev/gpiochip0',
-            consumer='led-blinker',
+            path=gpio_path,
+            consumer=consumer,
             config={
-                line: LineSettings(
+                self.line: LineSettings(
                     direction=Direction.OUTPUT,
                     output_value=Value.INACTIVE
                 )
@@ -32,3 +34,113 @@ class LedLineGpio:
         """Метод для освобождения ресурса"""
         
         self._request.release()
+
+
+class MotorDCLineGpio:
+    """Класс для управления линией GPIO для DC мотора"""
+
+    def __init__(self, line_in1: int, line_in2: int, gpio_path: str, consumer: str) -> None:
+        self.line_in1 = line_in1
+        self._request_in1 = request_lines(
+            path=gpio_path,
+            consumer=consumer,
+            config={
+                self.line_in1: LineSettings(
+                    direction=Direction.OUTPUT,
+                    output_value=Value.INACTIVE
+                )
+            }
+        )
+
+        self.line_in2 = line_in2
+        self._request_in2 = request_lines(
+            path=gpio_path,
+            consumer=consumer,
+            config={
+                self.line_in2: LineSettings(
+                    direction=Direction.OUTPUT,
+                    output_value=Value.INACTIVE
+                )
+            }
+        )
+
+    def forward_motor(self) -> None:
+        """Метод для вращения мотора вперёд"""
+
+        self._request_in1.set_value(self.line_in1, Value.ACTIVE)
+        self._request_in2.set_value(self.line_in2, Value.INACTIVE)
+
+    def backward_motor(self) -> None:
+        """Метод для вращения мотора назад"""
+
+        self._request_in1.set_value(self.line_in1, Value.INACTIVE)
+        self._request_in2.set_value(self.line_in2, Value.ACTIVE)
+
+    def stop_motor(self) -> None:
+        """Метод для остановки мотора"""
+
+        self._request_in1.set_value(self.line_in1, Value.INACTIVE)
+        self._request_in2.set_value(self.line_in2, Value.INACTIVE)
+
+    def close(self) -> None:
+        """Метод для освобождения ресурса"""
+
+        self._request_in1.release()
+        self._request_in2.release()
+
+
+class RobotControl:
+    """Класс для управления роботом"""
+    
+    def __init__(self) -> None:
+        self._left_motor = MotorDCLineGpio(
+            line_in1=settings.gpio_lines.left_motor_line_in1,
+            line_in2=settings.gpio_lines.left_motor_line_in2,
+            gpio_path=settings.gpio_lines.gpio_path,
+            consumer=settings.gpio_lines.left_motor_consumer
+        )
+        
+        self._right_motor = MotorDCLineGpio(
+            line_in1=settings.gpio_lines.right_motor_line_in1,
+            line_in2=settings.gpio_lines.right_motor_line_in2,
+            gpio_path=settings.gpio_lines.gpio_path,
+            consumer=settings.gpio_lines.right_motor_consumer
+        )
+
+        self.stop()
+
+    def forward(self) -> None:
+        """Движение робота вперёд"""
+        
+        self._left_motor.forward_motor()
+        self._right_motor.forward_motor()        
+
+    def backward(self) -> None:
+        """Движение робота назад"""
+        
+        self._left_motor.backward_motor()
+        self._right_motor.backward_motor()
+
+    def left(self) -> None:
+        """Поворот робота налево"""
+
+        self._right_motor.forward_motor()
+        self._left_motor.backward_motor()
+
+    def right(self) -> None:
+        """Поворот робота направо"""
+
+        self._right_motor.backward_motor()
+        self._left_motor.forward_motor()
+
+    def stop(self) -> None:
+        """Остановка робота"""
+
+        self._left_motor.stop_motor()
+        self._right_motor.stop_motor()
+
+    def close(self) -> None:
+        """Освобождение ресурса"""
+        
+        self._left_motor.close()
+        self._right_motor.close()

robot_pi_service/response_data.py

@@ -0,0 +1,27 @@
+from enum import Enum
+
+
+class FormResponse(Enum):
+    """Класс для формирования ответа"""
+
+    NOT_FOUND_COMMAND = (404, 'Unknown command!')
+    OK_COMMAND = (200, 'Command executed')
+    SERVER_ERR = (500, 'Internal Server Error!')
+
+    @property
+    def response(self) -> dict[str, int | str]:
+        """Словарь для формирования ответа серверу"""
+
+        return {
+            'status': self.value[0],
+            'message': self.value[1]
+        }
+
+    def get_response_err(self, name_err: str, message_err: str) -> dict[str, int | str]:
+        """Метод для формирования ответа серверу с ошибкой"""
+        
+        return {
+            'status': self.value[0],
+            'name_error': name_err,
+            'message': message_err
+        }

robot_pi_service/robot_pi_service.py

@@ -1,79 +1,84 @@
 import asyncio
+from socket import gethostbyname
 from websockets import serve, exceptions
 from websockets.legacy.server import WebSocketServerProtocol
-from json import loads, JSONDecodeError
+from json import loads, dumps, JSONDecodeError
 
 from settings import settings
-from gpio_control import LedLineGpio
+from gpio_control import RobotControl
+from response_data import FormResponse
 
 
-async def robot_control_gpio(websocket: WebSocketServerProtocol):
+async def robot_control_gpio(websocket: WebSocketServerProtocol) -> None:
     """
     Асинхронная функция для управлением gpio робота (через websocket)
     """
 
     try:
-        led_line_gpio = LedLineGpio(line=6)
-        action = None
+        robot_control = RobotControl()
+        commands_status = FormResponse
 
         while True:
             try:
                 command = await asyncio.wait_for(websocket.recv(), timeout=30.0)
                 data = loads(command)
+                command_name = data.get('command')
             except asyncio.TimeoutError as err:
                 print('Таймаут ожидания команды от клиента')
                 continue  # продолжение цикла, чтобы не закрывать соединение
             except JSONDecodeError:
-                print("Ошибка при декодировании JSON")
+                print('Ошибка при декодировании JSON!')
                 continue
-            
-            match next(iter(data), None):
+
+            match command_name:
                 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)
+                    robot_control.forward()
                 case settings.commands_robot.backward:
-                    action = 'Робот едет назад'
-                    print(action)
-                    await websocket.send(message=action)
+                    robot_control.backward()
                 case settings.commands_robot.left:
-                    action = 'Робот едет налево'
-                    print(action)
-                    await websocket.send(message=action)
+                    robot_control.left()
                 case settings.commands_robot.right:
-                    action = 'Робот едет направо'
-                    print(action)
-                    await websocket.send(message=action)
+                    robot_control.right()
+                case settings.commands_robot.stop:
+                    robot_control.stop()
+                case _:
+                    data.update(commands_status.NOT_FOUND_COMMAND.response)
+                    await websocket.send(message=dumps(data))
+
+            if settings.commands_robot.is_command(command=command_name):
+                data.update(commands_status.OK_COMMAND.response)
+                await websocket.send(message=dumps(data))
     except exceptions.ConnectionClosed:
         pass
     except (exceptions.ConnectionClosedOK, exceptions.InvalidMessage, exceptions.InvalidState) as err:
         message_err = f'{err.__class__.__name__}: {err}'
         print(message_err)
-        await websocket.send(message=message_err)
-    except Exception as err:
-        message_err = f'{err.__class__.__name__}: {err}'
-        print(message_err)
-        await websocket.send(message=message_err)
+        data.update(
+            commands_status.SERVER_ERR.get_response_err(
+                name_err=err.__class__.__name__, 
+                message_err=err
+            )
+        )
+        await websocket.send(message=dumps(data))
     finally:
-        led_line_gpio.close()
+        robot_control.close()
 
 
-async def start():
+async def start() -> None:
     """Асинхронная функция запуска вебсокета и бесконечного цикла"""
 
     print('Старт сервиса робота для приёма команд.')
 
-    async with serve(handler=robot_control_gpio, host=settings.websocket_host, port=settings.websocket_port):
+    async with serve(
+        handler=robot_control_gpio, 
+        host=gethostbyname(settings.websocket_host), 
+        port=settings.websocket_port
+    ):
         # бесконечный цикл
         await asyncio.Future()
 
 
-def run_app():
+def run_app() -> None:
     """Функция старата приложения"""
 
     try:

robot_pi_service/settings.py

@@ -18,6 +18,26 @@ class CommandsRobot(ModelConfig):
     backward: str
     left: str
     right: str
+    stop: str
+
+    def is_command(self, command: str) -> bool:
+        """Есть ли команда в командах"""
+
+        return command in self.model_dump().values()
+
+
+class GpioLines(ModelConfig):
+    """Класс для линий GPIO"""
+
+    gpio_path: str
+    led_line: int
+    led_consumer: str
+    left_motor_line_in1: int
+    left_motor_line_in2: int
+    left_motor_consumer: str
+    right_motor_line_in1: int
+    right_motor_line_in2: int
+    right_motor_consumer: str
 
 
 class Settings(ModelConfig):
@@ -26,6 +46,7 @@ class Settings(ModelConfig):
     websocket_host: str
     websocket_port: int
     commands_robot: CommandsRobot = CommandsRobot()
+    gpio_lines: GpioLines = GpioLines()
 
 
 settings = Settings()