From 72af2bc2f9b78156646a959c9daf86121fd0f3a0 Mon Sep 17 00:00:00 2001 From: Arduinum628 Date: Tue, 16 Sep 2025 09:57:02 +0300 Subject: [PATCH] feat: added classes for work motors, added class FormResponse, added logic for motor control in service; docs: updated .env.example for commands motors --- .env.exaple | 21 +++++ robot_pi_service/gpio_control.py | 122 +++++++++++++++++++++++++-- robot_pi_service/response_data.py | 27 ++++++ robot_pi_service/robot_pi_service.py | 73 ++++++++-------- robot_pi_service/settings.py | 21 +++++ 5 files changed, 225 insertions(+), 39 deletions(-) create mode 100644 .env.exaple create mode 100644 robot_pi_service/response_data.py diff --git a/.env.exaple b/.env.exaple new file mode 100644 index 0000000..41472c9 --- /dev/null +++ b/.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=идентификатор клиента для правого \ No newline at end of file diff --git a/robot_pi_service/gpio_control.py b/robot_pi_service/gpio_control.py index 85a5328..3b80da3 100644 --- a/robot_pi_service/gpio_control.py +++ b/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() diff --git a/robot_pi_service/response_data.py b/robot_pi_service/response_data.py new file mode 100644 index 0000000..0b92b83 --- /dev/null +++ b/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 + } diff --git a/robot_pi_service/robot_pi_service.py b/robot_pi_service/robot_pi_service.py index 387ff13..554096f 100755 --- a/robot_pi_service/robot_pi_service.py +++ b/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: diff --git a/robot_pi_service/settings.py b/robot_pi_service/settings.py index 47e1905..4faa168 100644 --- a/robot_pi_service/settings.py +++ b/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()