Основные библиотеки Python, используемые в ровере

На этой странице представлены основные использующиеся библиотеке в коде под ровер "Контакт". Больше библиотек прописано в описаниях кодов. Рекомендуется для проверки кодов на этой странице создать отдельную папку.

Папка для дальнейшего хранения файлов с примерами кодов

ПРИМЕЧАНИЕ: Библиотеки, которые вы используете в VS Code, необходимо так же докачивать на Orange Pi (подробнее об установке Python и pip смотрите в разделе "Работа с Orange Pi" в статьях "Установка Python 3.10" и "Установка pip"). Команды установки библиотек в VS Code и в Orange Pi не отличаются.

asyncio:

  • Описание: asyncio используется в коде ровера для написания асинхронных программ, используя корутины, задачи и событийные циклы.

  • Особенности:

    • Позволяет выполнять несколько задач ровера одновременно, не блокируя основной поток выполнения.

    • Используется для асинхронного ввода-вывода, что повышает производительность при работе с сетью.

  • Роль в проекте:

    • Асинхронная обработка соединений с клиентами через сокеты.

    • Асинхронное чтение файлов с помощью aiofiles.

    • Управление событиями и задачами в рамках основного цикла событий.

  • Пример кода с использование библиотеки asyncio:

import asyncio


async def move():
    print("Ровер начинает движение.")
    await asyncio.sleep(2)
    print("Ровер продолжает движение.")
    await asyncio.sleep(2)
    print("Ровер останавливается.")

async def turn(direction):
    print(f"Ровер начинает поворот {direction}.")
    await asyncio.sleep(2)
    print(f"Ровер завершает поворот {direction}.")

async def blink_lights():
    for _ in range(5):
        print("Фары мигают.")
        await asyncio.sleep(1)

async def main():
    await asyncio.gather(
        move(),
        turn("налево"),
        blink_lights()
    )

# Запуск событийного цикла
asyncio.run(main())

Описание кода:

  1. Функция move Имитирует движение ровера, выводя сообщения о начале, продолжении и остановке движения с 2-секундными интервалами.

  2. Функция turn Имитирует поворот ровера в заданном направлении (налево или направо), выводя сообщения о начале и завершении поворота с 2-секундными интервалами.

  3. Функция blink_lights Имитирует мигание фар ровера, выводя сообщение о мигании каждую секунду в течение 5 секунд.

  4. Функция main: Функция main использует asyncio.gather для одновременного запуска всех трех асинхронных задач (move(), turn("налево"), blink_lights()).Функция main использует asyncio.gather для одновременного запуска всех трех асинхронных задач (move(), turn("налево"), blink_lights()).

  5. asyncio.run(main()) - запуск кода

Итоговый вывод: 

Ровер начинает движение.
Ровер начинает поворот налево.
Фары мигают.
Фары мигают.
Ровер продолжает движение.
Фары мигают.
Ровер завершает поворот налево.
Фары мигают.
Ровер останавливается.
Фары мигают.

Процесс запуска кода:

  1. Откройте VS Code:

    • Запустите Visual Studio Code на вашем компьютере.

    Первое окно при запуске Visual Studio Code

  2. Откройте папку с проектом:

    • В VS Code выберите File -> Open Folder... и выберите папку, которую вы создали.

    Выбор действия во вкладе File
    Вот так выглядит окно иерархии сейчас

  3. Создайте файл с кодом:

    • Cоздайте новый файл с расширением .py, назовите его asynciolib.py, и вставьте в него предоставленный код.

Создание нового файла
Конечный вид окна иерархии

ПРИМЕЧАНИЕ: Когда в списке действий в иерархии вы нажимаете New File, изначально создается внутренняя папка, где в дальнейшем будут храниться наши файлы. Назовите ее как хотите и после, наведя курсор на папку, создайте файл таким же образом. Так же не забывайте в конце названия файла прописывать .py - без типа файла VS Code автоматически создает txt файл.

  1. Установите необходимые зависимости:

  • В данном случае библиотека предустановлена, так что в окно программирования вставляем наш код

Код в файле asynciolib.py

  1. Сохранение файла: перед запуском кода всегда сохраняйте свой файл комбинацией ctrl + S или через file - save.

  2. Запустите ваш код: в правом верхнем углу есть значок запуска кода. Нажав на него, у вас автоматически вылезет терминал, в который и будет выводиться результат. ВАЖНО: когда вы запустите код впервые, у вас попросят выбрать версию

Кнопка запуска программы

Вывод консоли и данных о проезде и состоянии фар ровера

contextlib:

  • Описание: contextlib используется в ровере для работы с контекстными менеджерами.

  • Особенности:

    • Облегчает создание и использование контекстных менеджеров.

    • Содержит полезные утилиты, такие как suppress, для обработки исключений.

  • Роль в проекте:

    • Управление контекстами и исключениями при выполнении задач, что позволяет обеспечить безопасное выполнение кода.

  • Пример кода с работой библиотеки contextlib:

import contextlib

@contextlib.contextmanager
def open_file(filename, mode):
    try:
        f = open(filename, mode)
        print(f"Файл {filename} открыт.")
        yield f
    finally:
        f.close()
        print(f"Файл {filename} закрыт.")

# Использование контекстного менеджера
with open_file('example.txt', 'w') as file:
    file.write('Hello, rover Contact!')
    print("Данные записаны в файл.")

# Чтение данных из файла
with open_file('example.txt', 'r') as file:
    data = file.read()
    print(f"Прочитанные данные: {data}")

Описание кода:

  1. @contextlib.contextmanager - Создание контекстного менеджера

  2. Функция open_file:

    • Функция open_file принимает два аргумента: filename (имя файла) и mode(режим открытия файла).

    • В блоке try, файл открывается и выводится сообщение о том, что файл открыт.

    • Ключевое слово yield используется для передачи управления обратно в блок with, где файл будет использоваться.

    • В блоке finally, файл закрывается и выводится сообщение о том, что файл закрыт.

  3. Использование контекстного менеджера: В этом блоке with, файл example.txt открывается в режиме записи ('w'), и в него записывается строка 'Привет, ровер Контакт!'.

  4. Чтение данных из файла: В этом блоке with, файл example.txt открывается в режиме чтения ('r'), и данные из файла считываются и выводятся на экран.

Итоговый вывод: 

Файл example.txt открыт.
Данные записаны в файл.
Файл example.txt закрыт.
Файл example.txt открыт.
Прочитанные данные: Привет, ровер Контакт!
Файл example.txt закрыт.

Процесс запуска кода:

  1. Откройте VS Code:

  • при открытии VS Code повторно открывается ваша папка. Если он у вас уже открыт, то переходите к следующему пункту.

Начальное окно теперь выглядит так

  1. Создайте новый файл .py:

  • создайте в окне иерархии новый .py файл и назовите его my_contextlib.py

ПРИМЕЧАНИЕ: Если назвать файл contextlib.py, python будет ругаться на несоответствие файловой конфигурации!

Новый файл my_contextlib.py

  1. Вставьте ваш код в окно программирования:

Окно иерархии и программирования

  1. Обязательно сохраните файл комбинацией ctrl + S или через file - save.

  2. Скачайте необходимые зависимости: в данном случае придется докачать библиотеку, так как она не идет стандартом в Python. Установка идет следующим образом:\

  • Откройте терминал: если терминал у вас не открыт, перейдите в Terminal - New terminal. Так же можно запустить через горячие клавиши Ctrl + Shift + `

    Запуск терминала

  • Пропишите в консоль pip install contextlib2и нажмите Enter, у вас запустится загрузка библиотеки

Установка библиотеки contextlib

  1. Запустите код:

Запуск кода происходит через эту кнопку

В консоль выведется следующая информация:

Информация о записи файла в example.txt

Так же в иерархии добавится файл example.txt, содержащий в себе текст, который мы передали ему.

Содержимое текстового файла example.txt

socket:

  • Описание: socket используется в ровере для работы с сетевыми соединениями.

  • Особенности:

    • Обеспечивает низкоуровневый интерфейс для работы с сетевыми сокетами.

  • Роль в проекте:

    • Создание серверных и клиентских сокетов для обмена данными с ровером через локальную сеть.

  • Пример кода с использованием библиотеки socket:

Код сервера:

import socket

def start_server():
    # Создаем сокет
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # Привязываем сокет к адресу и порту
    server_address = ('localhost', 10000)
    server_socket.bind(server_address)
    
    # Слушаем входящие соединения
    server_socket.listen(1)
    print("Сервер запущен и слушает на порту 10000...")
    
    while True:
        # Ждем соединения
        print("Ожидание соединения...")
        client_socket, client_address = server_socket.accept()
        print(f"Соединение установлено с {client_address}")
        
        try:
            # Получаем данные от клиента
            data = client_socket.recv(1024)
            print(f"Получено сообщение: {data.decode()}")
            
            # Отправляем ответ клиенту
            client_socket.sendall("Привет, ровер Контакт!")
        finally:
            # Закрываем соединение
            client_socket.close()
            print(f"Соединение с {client_address} закрыто")

if __name__ == "__main__":
    start_server()

Описание кода сервера:

  • Создает сокет с использованием socket.AF_INET (для IPv4) и socket.SOCK_STREAM (для TCP).

  • Привязывает сокет к адресу localhost и порту 10000.

  • Слушает входящие соединения с помощью метода listen.

  • Принимает соединение с помощью метода accept.

  • Получает данные от клиента и отправляет обратно ответ.

  • Закрывает соединение после обработки.

Код клиента:

import socket

def start_client():
    # Создаем сокет
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # Подключаемся к серверу
    server_address = ('localhost', 10000)
    client_socket.connect(server_address)
    
    try:
        # Отправляем сообщение серверу
        message = "Привет, сервер!"
        client_socket.sendall(message.encode())
        
        # Получаем ответ от сервера
        data = client_socket.recv(1024)
        print(f"Получено сообщение от сервера: {data.decode()}")
    finally:
        # Закрываем соединение
        client_socket.close()

if __name__ == "__main__":
    start_client()

Описание кода клиента:

  • Создает сокет с использованием socket.AF_INET (для IPv4) и socket.SOCK_STREAM (для TCP).

  • Подключается к серверу по адресу localhost и порту 10000.

  • Отправляет сообщение серверу.

  • Получает ответ от сервера и выводит его на экран.

  • Закрывает соединение после обработки.

Как запустить:

  1. Добавьте код сервера и клиента в Visual Studio Code

  2. Запустите код сервера

  3. Запустите код клиента

  4. Если все правильно запущено, в консоль сервера начнут выводиться следующие данные:

Сервер запущен и слушает на порту 10000...
Ожидание соединения...
Соединение установлено с ('localhost', 52346)
Получено сообщение: Привет, сервер!
Соединение с ('127.0.0.1', 52346) закрыто

в консоль клиента:

Получено сообщение от сервера: Привет, ровер Контакт!

Процесс запуска кода:

  1. Создайте два файла: server.py и client.py:

Окно иерархии сейчас

  1. Вставьте соответвующие кода в нужные файлы:

  • Файл клиент:

Файл клиент с кодом под клиента

  • Файл сервер:

файл сервер с кодом сервера

  1. Запустите код сервера: в данном случае нет необходимости качать дополнительные библиотеки, поэтому просто запускаем код сервера, в консоль выведется следующее:

В консоли выводится информация об ожидании подключения клиента

  1. Запустите код клиента: после запуска кода сервера нужно подключить клиент. Для этого перейдите в файл client.py и запустите код с помощью режима откладки:

Запуск кода с помощью режима откладки

  1. Осмотрите консоли клиента и сервера: в клиент было отправлено сообщение "Получено сообщение от сервера":

Консоль файла клиента

В сервер было отправлено следующее выражение:

Консоль файла сервера

ПРИМЕЧАНИЕ: Ошибки, которые вывелись в консоль - это более подробное описания процесса закрытия соединения с клиентом, так что не пугайтесь этих ошибок!

threading:

  • Описание: threading используется в ровере для работы с потоками.

  • Особенности:

    • Позволяет выполнять задачи параллельно в разных потоках.

    • Поддерживает блокировки и события для синхронизации потоков.

  • Роль в проекте:

    • Управление потоками для выполнения фоновых задач, таких как обработка видеопотока.

  • пример кода с использованием библиотеки threading:

import threading
import time

def move():
    print("Ровер начинает движение.")
    time.sleep(2)
    print("Ровер продолжает движение.")
    time.sleep(2)
    print("Ровер останавливается.")

def blink_lights():
    for _ in range(5):
        print("Фары мигают.")
        time.sleep(1)

# Создаем потоки
move_thread = threading.Thread(target=move)
blink_lights_thread = threading.Thread(target=blink_lights)

# Запускаем потоки
move_thread.start()
blink_lights_thread.start()

# Ждем завершения потоков
move_thread.join()
blink_lights_thread.join()

print("Все задачи завершены.")

Описание кода:

  1. Функции move и blink_lights:

    • move: Имитирует движение ровера, выводя сообщения о начале, продолжении и остановке движения. Использует time.sleep для задержек.

    • blink_lights: Имитирует мигание фар ровера, выводя сообщения о мигании фар и используя time.sleep для задержек.

  2. Создание потоков:

    • move_thread и blink_lights_thread: Создаются объекты потоков, которые будут выполнять функции move и blink_lights соответственно.

  3. Запуск потоков:

    • move_thread.start() и blink_lights_thread.start(): Запускают потоки, чтобы они начали выполнение своих функций параллельно.

  4. Ожидание завершения потоков:

    • move_thread.join() и blink_lights_thread.join(): Обеспечивают ожидание завершения выполнения потоков, чтобы основной поток не завершился раньше, чем потоки выполнят свои задачи.

  5. Сообщение о завершении:

    • print("Все задачи завершены."): Выводит сообщение после того, как все потоки завершили свою работу.

Процесс запуска кода:

  1. Создайте файл .py и назовите его threadinglib.py:

Окно иерархии с новым файлом threadinglib.py

  1. Вставьте код в окно программирования:

Окно программирования после добавления кода

  1. Запустите код: в данном случае устанавливать ничего не нужно. Просто запускаем код и смотрим на результат:

aiofiles:

  • Описание: aiofiles используется в ровере для асинхронной работы с файловой системой.

  • Особенности:

    • Обеспечивает асинхронное чтение и запись файлов.

    • Поддерживает интерфейс, аналогичный стандартной библиотеке open.

  • Роль в проекте:

    • Асинхронное чтение файлов, содержащих данные карты ArUco.

  • Пример кода с использованием библиотеки aiofiles:

```python
import asyncio
import aiofiles

async def log_data(data):
    async with aiofiles.open('rover_log.txt', mode='a') as file:
        await file.write(data + '\n')
        print(f"Записано: {data}")

async def main():
    tasks = [
        log_data("The rover started moving."),
        log_data("The rover continues to move."),
        log_data("The rover stopped."),
        log_data("The rover started flashing its headlights."),
        log_data("The rover has finished flashing its headlights.")
    ]
    await asyncio.gather(*tasks)

# Запуск событийного цикла
asyncio.run(main())
```

Описание кода:

  1. Функция log_data:

    • Принимает строку data и асинхронно записывает её в файл rover_log.txt.

    • Использует aiofiles.open для асинхронного открытия файла в режиме добавления (mode='a').

    • await file.write(data + '\n') асинхронно записывает данные в файл и добавляет перенос строки.

    • Выводит сообщение о записанных данных.

  2. Функция main:

    • Создает список задач tasks, каждая из которых представляет собой вызов функции log_data с различными сообщениями.

    • Использует asyncio.gather для одновременного выполнения всех задач.

  3. Запуск событийного цикла:

    • asyncio.run(main()) запускает асинхронную функцию main и управляет событийным циклом.

    Обратите внимание: в данном коде так же используется библиотека asyncio! Она описана выше!

Итоговый вывод:

Записано: Ровер начал движение.
Записано: Ровер продолжает движение.
Записано: Ровер остановился.
Записано: Ровер начал мигать фарами.
Записано: Ровер завершил мигание фарами.

Процесс запуска кода:

  1. Создайте файл aiofiles.py:

Окно иерархии

Пока что файл подсвечен желтым, он подсвечивается как только вы добавляете код в файл. Это потому, что библиотека не установлена. Если вас смущает то, что после установки библиотеки файл до сих пор помечен желтым, можете перезагрузить Visual Studio Code и проблема решится.

  1. Добавьте код в файл:

Окно программирования после добавленного кода

  1. Установите необходимые зависимости: в данном случае библиотеку aiofiles надо докачать. Открываем терминал и прописываем туда следующую команду: pip install aiofiles

Команда установки библиотеки и процесс скачивания

  1. Запустите код: после скачивания библиотеки можно приступать к тестированию кода. Запустите кода и посмотрите, что происходит:

вывод консоли

В консоль выводится информация о записанных данных. Так же в иерархии создался новый текстовый файл rover_log.txt

Текстовый файл rover_log.txt

logging:

  • Описание: loggingиспользуется в ровере для ведения логов.

  • Особенности:

    • Обеспечивает гибкую систему для создания логов различных уровней (информация, предупреждения, ошибки и т.д.).

    • Поддерживает вывод логов в консоль, файлы и другие обработчики.

  • Роль в проекте:

    • Ведение логов для отслеживания работы программы и отладки.

  • Пример кода с использованием библиотеки logging:

```python
import logging

# Настройка логгера
logging.basicConfig(
    filename='rover_log.log',
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# Создание логгера
logger = logging.getLogger('rover_logger')

# Примеры логирования сообщений
def log_rover_activity():
    logger.debug("The rover started moving.")
    logger.info("The rover continues to move.")
    logger.warning("The rover is approaching an obstacle.")
    logger.error("The rover encountered an obstacle.")
    logger.critical("The rover has lost contact with the user.")

# Вызов функции для логирования активности ровера
log_rover_activity()
```

Описание кода:

  1. Настройка логгера:

    • logging.basicConfig настраивает базовую конфигурацию логгера.

    • filename='rover_log.log' указывает, что логи будут записываться в файл rover_log.log.

    • level=logging.DEBUG устанавливает минимальный уровень логирования на DEBUG, что означает, что все сообщения от DEBUG и выше будут записаны.

    • format='%(asctime)s - %(levelname)s - %(message)s' задает формат записи логов, включая время, уровень сообщения и само сообщение.

    • datefmt='%Y-%m-%d %H:%M:%S' задает формат времени.

  2. Создание логгера:

    • logger = logging.getLogger('rover_logger') создает экземпляр логгера с именем rover_logger.

  3. Примеры логирования сообщений:

    • logger.debug, logger.info, logger.warning, logger.error, logger.critical используются для логирования сообщений различных уровней важности.

    • Вызов функции для логирования активности ровера:

  4. log_rover_activity() вызывает функцию, которая логирует различные сообщения от ровера.

Итоговый вывод:

В консоль данных не выводится. Вместо этого данные записываются в лог

2023-10-05 14:30:00 - DEBUG - Ровер начал движение.
2023-10-05 14:30:00 - INFO - Ровер продолжает движение.
2023-10-05 14:30:00 - WARNING - Ровер приближается к препятствию.
2023-10-05 14:30:00 - ERROR - Ровер столкнулся с препятствием.
2023-10-05 14:30:00 - CRITICAL - Ровер потерял связь с пользователем.

Процесс запуска кода:

  1. Создайте файл с названием logginglib.py:

Окно иерархии с новым файлом logginglib.py

  1. Добавьте код в окно программирования:

Окно программирования файла logginglib.py

  1. Запустите код: здесь не нужно докачивать библиотеки, поэтому просто запускаем код:

Результат выполнения кода в консоли

В консоль ничего не вывелось, однако в иерархии добавился файл rover_log.log:

Файл rover_log,py

queue:

  • Описание: queue используется в ровере для работы с очередями.

  • Особенности:

    • Поддерживает различные типы очередей (FIFO, LIFO, приоритетные).

    • Безопасна для использования в многопоточных программах.

  • Роль в проекте:

    • Хранение кадров видеопотока для дальнейшей обработки.

  • Пример кода с использованием библиотеки logging:

import queue
import threading
import time

# Создание очереди
message_queue = queue.Queue()

# Функция для добавления сообщений в очередь
def producer():
    for i in range(10):
        message = f"Сообщение {i}"
        message_queue.put(message)
        print(f"Ровер отправил: {message}")
        time.sleep(1)

# Функция для обработки сообщений из очереди
def consumer():
    while True:
        message = message_queue.get()
        if message is None:
            break
        print(f"Пользователь обработал: {message}")
        message_queue.task_done()

# Создание и запуск потоков
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

# Ожидание завершения работы производителя
producer_thread.join()

# Добавление сигнала о завершении для потребителя
message_queue.put(None)
consumer_thread.join()

print("Работа завершена.")

Описание кода:

  1. Создание очереди:

    • message_queue = queue.Queue() создает очередь для хранения сообщений.

  2. Функция для добавления сообщений в очередь:

    • producer функция добавляет 10 сообщений в очередь с интервалом в 1 секунду.

  3. Функция для обработки сообщений из очереди:

    • consumer функция бесконечно обрабатывает сообщения из очереди, пока не получит сигнал о завершении (None).

  4. Создание и запуск потоков:

    • producer_thread и consumer_thread создаются и запускаются для параллельной работы.

  5. Ожидание завершения работы производителя:

    • producer_thread.join() ожидает завершения работы потока производителя.

  6. Добавление сигнала о завершении для потребителя:

    • message_queue.put(None) добавляет сигнал о завершении для потребителя.

  7. Ожидание завершения работы потребителя:

    • consumer_thread.join() ожидает завершения работы потока потребителя.

Итоговый вывод:

Ровер отправил: Сообщение 0
Пользователь обработал: Сообщение 0
Ровер отправил: Сообщение 1
Пользователь обработал: Сообщение 1
Ровер отправил: Сообщение 2
Пользователь обработал: Сообщение 2
Ровер отправил: Сообщение 3
Пользователь обработал: Сообщение 3
Ровер отправил: Сообщение 4
Пользователь обработал: Сообщение 4
Ровер отправил: Сообщение 5
Пользователь обработал: Сообщение 5
Ровер отправил: Сообщение 6
Пользователь обработал: Сообщение 6
Ровер отправил: Сообщение 7
Пользователь обработал: Сообщение 7
Ровер отправил: Сообщение 8
Пользователь обработал: Сообщение 8
Ровер отправил: Сообщение 9
Пользователь обработал: Сообщение 9
Работа завершена.

Процесс запуска кода:

  1. Создайте файл queuelib.py:

Окно иерархии с новым файлом queuelib.py

  1. Добавьте в окно программирования код:

Окно программирования после добавления кода

  1. Запустите код: в данном случае ничего устанавливать не нужно, поэтому просто запускаем код:

Вывод в консоли после запуска.

В итоге наше окно иерархии выглядит так, в нем есть различные файлы и это далеко не всё, что можно сделать с этими библиотеками!

Конечный вид окна иерархии
PreviousФайл конфигурации ровера "Контакт"NextТехнология сокетов

Last updated