Инженерный тур. 3 этап

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения задачи нужно:

1. Безошибочно подключить соединительные провода к электронным модулям беспилотного автомобиля.
2. Подключиться к бортовому компьютеру беспилотного автомобиля через SSH.
3. Скопировать файлы базового программного кода на бортовой компьютер и запустить их для исполнения.

Эти действия выполняются по предоставленным инструкциям.

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения поставленной задачи следует разобраться в базовом коде и добавить в него два алгоритма: алгоритм обнаружения запрещающего движение знака и алгоритм поиска зоны приема проб ЛВА.

Для поиска дорожного знака на изображении можно применять различные способы:

• обучение нейросети;
• использование каскада Хаара;
• поиск объекта по цвету.

Самый простой вариант решения — последний, однако этот метод требует четко подобранных порогов бинаризации.

В решении, представленном ниже, перед основным циклом обработки кадров расположен дополнительный цикл, который выполняется, пока запрещающий знак не исчезнет из кадра. Внутри цикла проводится бинаризация и поиск контуров. Для каждого найденного контура вычисляется минимальная окружность, которая его охватывает.

Если площадь контура составляет более 80% от площади этой окружности, считается, что знак найден.

Если круг не найден в течение 2 с, то цикл завершается и начинается основной цикл обработки кадров.

while True:
   _, frame = cap.read()
   height, width, _ = frame.shape
   new_width = int(width * 0.3)
   cropped_frame = frame[50:250, new_width : width - new_width]
   hsl_frame = cv2.cvtColor(cropped_frame, cv2.COLOR_BGR2HLS)
   binary = cv2.inRange(hsl_frame, (132, 86, 102), (179, 255, 255))
   contours, _ = cv2.findContours(binary, cv2.RETR_EXTERNAL,
    cv2.CHAIN_APPROX_SIMPLE)
   circle_found = False
   for contour in contours:
       (x, y), radius = cv2.minEnclosingCircle(contour)
       center = (int(x), int(y))
       radius = int(radius)
       try:
           if cv2.contourArea(contour) / (np.pi * radius * radius) > 0.8:
               cv2.circle(cropped_frame, center, radius, (0, 255, 0), 2)
               circle_found = True
               circle_last_seen = time.monotonic()
               print("Circle found")
               break
       except Exception as e:
           print(e)
   if not circle_found:
       print("Circle not found")
   if not circle_found and
                          (time.monotonic() - circle_last_seen) > 2:
       print("Circle not found for 2 seconds, exiting...")
       break

Вторая часть задания связана с остановкой в зоне приема проб ЛВА. Задача сводится к детектированию зеленого прямоугольника и отсчету времени для остановки беспилотника в нужный момент. В решении, представленном ниже, проводится бинаризация так, чтобы зеленые пиксели стали белыми, а все остальные – черными.

Подсчитывается число белых пикселей: если оно больше определенного значения, то линия считается обнаруженной (это должно произойти два раза подряд). Если линия обнаружена, то засекается время и через 0,514 с беспилотник останавливается.

Следующий фрагмент кода располагается в основном цикле обработки кадров.

mark_frame = frame[150:200, 216:316]
hsv_frame = cv2.cvtColor(mark_frame, cv2.COLOR_BGR2HSV_FULL)
bin_frame = cv2.inRange(hsv_frame, (0, 76, 0), (179, 255, 255))
green = np.count_nonzero(bin_frame)
# print(f"Current green: {green}")
if green > 600 and not green_flag:
   # print("Found green > 2000")
   green_flag = True
if green_flag and green < 600 and SUPER_FLAG == False:
   print("Green end")
   LAST_TIME_DETECTED = time.monotonic()
   SUPER_FLAG = True
if SUPER_FLAG == True and time.monotonic() > LAST_TIME_DETECTED + 0.514:
   STATE = STOP
   print("STOP")
   arduino.stop()

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Такие изображения накладываются на полигон для беспилотных автомобилей между линиями дорожной разметки. Базовый алгоритм движения по разметке не отличает друг от друга белые повреждения и линии дорожной разметки, из-за чего беспилотный автомобиль съезжает с трассы на поврежденных участках. Поэтому часть задачи — модернизация алгоритма движения по разметке.

В базовый алгоритм детектирования разметки можно добавить функцию, которая будет удалять белые повреждения с изображения участка дороги перед колесами автомобиля.

def fill_points(bin_frame):
   contours, _ = cv2.findContours(bin_frame,
                       cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
   new_mask = np.zeros_like(bin_frame)
   if contours:
       min_x = min(cv2.boundingRect(cnt)[0] for cnt in contours)
       max_x = max(cv2.boundingRect(cnt)[0] +
                   cv2.boundingRect(cnt)[2] for cnt in contours)
       tolerance = 50
       left_contours = [cnt for cnt in contours if
                      cv2.boundingRect(cnt)[0] <= min_x + tolerance]
       right_contours = [cnt for cnt in contours if
                         cv2.boundingRect(cnt)[0] +
                      cv2.boundingRect(cnt)[2] >= max_x - tolerance]
       cv2.drawContours(new_mask, left_contours, -1, 255,
        thickness=cv2.FILLED)
       cv2.drawContours(new_mask, right_contours, -1, 255,
         thickness=cv2.FILLED)
   return new_mask

Функция получает на вход бинаризованное изображение участка дороги, на котором производится поиск всех контуров, определяются координаты самого правого и левого контура. Контуры в интервале \(tolerance = 50\) от самого левого и правого переносятся на новое черно-белое изображение, идентичное по размерам исходному. Таким образом все контуры повреждений между линиями дорожной разметки исчезают. Полученное изображение можно передавать в базовый алгоритм движения по разметке.

Вторая часть задачи — обнаружение поврежденных участков дороги. Надежнее всего использовать для этой цели нейросетевой детектор. Необходимо собрать и разметить датасет со всеми видами поврежденных участков дорожного полотна — сфотографировать объекты с разнообразных ракурсов, под которыми беспилотный автомобиль может их увидеть.

Если предполагается обнаружение объектов на одной конкретной дистанции, то в датасет нужно добавлять только фотографии с объектами на этом конкретном удалении от камеры; в настройках обучения детектора нужно отключить масштабирование при аугментации данных.

На полученном датасете необходимо обучить детектор Yolo v4-tiny и выполнить квантование модели для запуска на бортовом компьютере беспилотного автомобиля. Ниже приведен код, использующий детектор для обнаружения поврежденных участков дорожного полотна. Для избежания множественных срабатываний детектора вводится задержка 2 с.

Перед циклом чтения кадров выполняются инструкции, загружающие модель нейронной сети из файла и настраивающие ее якорные рамки.

model = yolopy.Model('best.tmfile', use_uint8=True,
use_timvx=True, cls_num=1)
model.set_anchors([18, 33, 33, 48, 25, 71, 58, 76,
40, 113, 87, 140])

Следующая функция возвращает True, если в кадре появляется поврежденный участок дороги.

def detect_yolo(frame):
   frame = apply_perspective(frame, [(511, 392),
(725, 391), (406, 713), (841, 715)])
   classes, scores, boxes = model.detect(frame)
   return len(classes) > 0

В цикле обработки кадров добавлена проверка наличия поврежденных участков дороги. Если подобные участки есть и в течение последних 2 с подобных участков в кадре не было, то отправляется запрос MQTT-серверу, ответ от которого выводится в терминал.

if detect_yolo(frame):
   if time.time() - last_detect > 2:
       client.send()
       response = client.get_output()
       print(response)
   last_detect = time.time()

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

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

Ниже представлен код программы сервера, принимающего и хранящего данные о поврежденных участках дороги. Серверу передаются новые данные, после чего информация о поврежденных участках обновится.

У сервера можно запросить те данные, которые он сейчас хранит. Программа сервера использует FastAPI. Определен класс для хранения данных о точках — поврежденных участках дороги. Задан endpoint для отправки данных о точках на сервер и endpoint для получения данных обо всех точках, хранящихся на сервере.

import copy
import uvicorn
import time
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Dict
class PointsRequest(BaseModel):
   type: str
   number: int
   count: int
   x: int
   y: int
@app.post("/points")
async def post_points(request: List[PointsRequest]):
   global points
   points = []
   for point in request:
       points.append({"type": point.type, "number": point.number, "count": point.count, "x": point.x, "y": point.y})
   return {"status": "OK", "detail": None}
@app.get("/points")
async def get_points():
   print(points)
   return {"status": "OK", "detail": points}
if __name__ == "__main__":
   uvicorn.run(app, host="0.0.0.0", port=8000)

import requests
class Server:
   def __init__(self):
       self.base_url = "http://172.16.65.50:8000"
   def send_points(self, points: list) -> bool:
       ans = requests.post(f"{self.base_url}/points", json=points)
       if ans.status_code != 200:
           return False
       return ans.json()["status"] == "OK"
   def get_points(self) -> list:
       ans = requests.get(f"{self.base_url}/points")
       if ans.status_code != 200:
           return []
       return ans.json()["detail"]

class DamagePoint:
   def __init__(self, x, y, damage_type, visit_count, number):
       self.x: int = x
       self.y: int = y
       self.damage_type = damage_type
       self.visit_count: int = visit_count
       self.number: int = number
   def json(self):
       return {"type": self.damage_type, "number": self.number, "count": self.visit_count, "x": self.x, "y": self.y}

def send_all_points(points: List[DamagePoint]):
   data = []
   for point in points:
       data.append(point.json())
   if server.send_points(data):
       print("Data was sent to server!")
   else:
       print("Error in sanding data!")

image_path = "./image.jpg"
image = cv2.imread(image_path)
image = cv2.resize(image, (815, 600))
original_image = image.copy()
damage_points = []
def draw_points():
   global image
   image = original_image.copy()
   for idx, point in enumerate(damage_points):
       cv2.circle(image, (point.x, point.y), 5, (0, 0, 255), -1)
       text = f"{point.damage_type}/{point.visit_count}/
        {point.number}"
       cv2.putText(image, text, (point.x - 20, point.y - 15),
                      cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)
   cv2.imshow("Map", image)

def update_points():
   global damage_points, biggest_num
   damage_points = []
   server_points = server.get_points()
   biggest_num = 0
   for point in server_points:
       biggest_num = max(biggest_num, point["number"])
       damage_points.append(DamagePoint(point["x"], point["y"],
    point["type"], point["count"], point["number"]))
   draw_points()
def main():
   draw_points()
   while True:
       key = cv2.waitKey(1) & 0xFF
       update_points()
       if key == 27:
           break
   cv2.destroyAllWindows()
if __name__ == "__main__":
   main()

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для передачи данных между беспилотным автомобилем и компьютером, отображающим карту, удобно использовать FastAPI-сервер из предыдущей подзадачи, дополненный функциями для передачи задачи беспилотному автомобилю.

Задача — последовательность поворотов, которые необходимо совершить, и id поврежденых точек, которые нужно посетить. Кроме того, необходимо добавить функцию для увеличения на сервере счетчика посещений точки с конкретным id. Ниже представлен программный код, реализующий эти дополнения.

@app.post("/task")
async def post_task(request: List[str]):
   global tasks
   tasks = request
   return {"status": "OK", "detail": None}
@app.get("/task")
async def get_task():
   global tasks
   tasks_copy = copy.deepcopy(tasks)
   tasks = []
   return {"status": "OK", "detail": tasks_copy}
@app.post("/point")
async def visit_point(number: int):
   for i in range(len(points)):
       if points[i]["number"] == number:
           points[i]["count"] += 1
           return {"status": "OK", "detail": None}
   return {"status": "ER", "detail": "Number not found"}

К классу Server из предыдущей подзадачи добавлены методы для новых взаимодействий с сервером.

def send_task(self, tasks: list) -> bool:
   ans = requests.post(f"{self.base_url}/task", json=tasks)
   print(ans.json())
   if ans.status_code != 200:
       return False
   return ans.json()["status"] == "OK"
def get_task(self) -> list:
   ans = requests.get(f"{self.base_url}/task")
   if ans.status_code != 200:
       return []
   return ans.json()["detail"]
def visit_point(self, number) -> list:
   ans = requests.post(f"{self.base_url}/point?number={number}")
   if ans.status_code != 200:
       return []
   return ans.json()["status"]

Наибольшее количество изменений требуется внести в программу, отвечающую за отображение карты, а именно:

• возможность размещать точки на карте с помощью мыши;
• алгоритм для формирования задания для беспилотного автомобиля.

Новая функция mouse_callback обрабатывает события, связанные с мышью. При нажатии левой кнопки мыши добавляется новая поврежденная точка или удаляется существующая. При нажатии правой кнопки мыши устанавливаются стартовая и конечная точки для беспилотника.

robot_zone = None
stop_zone = None
freeze_zone = None
def mouse_callback(event, x, y, flags, param):
   global damage_points, robot_zone, biggest_num, stop_zone, freeze_zone
   if event == cv2.EVENT_LBUTTONDOWN:
       for i, point in enumerate(damage_points):
           if abs(point.x - x) < 10 and abs(point.y - y) < 10:
               print(f"Удаление точки: {point.damage_type}")
               del damage_points[i]
               draw_points()
               send_all_points(damage_points)
               return
       damage_type = input("Тип повреждения: ")
       damage_point = DamagePoint(x, y, damage_type, 0, biggest_num + 1)
       biggest_num += 1
       damage_points.append(damage_point)
       send_all_points(damage_points)
       draw_points()
   elif event == cv2.EVENT_RBUTTONDOWN:
       if not robot_zone:
           robot_zone, _ = DamagePoint(x, y, 0, 0, 0).get_zone()
           print(f"Робот установлен в зону {robot_zone}")
           freeze_zone = input("Зона для установки: ")
           print("Нажмите на точку остановки")
       else:
           stop_zone, _ = DamagePoint(x, y, 0, 0, 0).get_zone()
       if robot_zone and stop_zone:
           plan_robot_route()

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

def plan_robot_route():
   if not robot_zone:
       print("Зона беспилотника не установлена.")
       return
   target_zones = {dp.get_zone()[0] for dp in damage_points}
   if not target_zones:
       print("Точек нет")
       return
   route, visited = find_optimal_route(robot_zone, target_zones)
   vis_cnt = 0
   task = []
   for i in range(len(route) - 1):
       for neighbor, action in graph.get(route[i], []):
           if neighbor == route[i + 1]:
               task.append(action)
               while (vis_cnt < len(visited) and i + 1 ==
        visited[vis_cnt][0] - 1):
                   if freeze_zone == visited[vis_cnt][1]:
                       pass
                   else:
                       task.append(visited[vis_cnt][1])
                   vis_cnt += 1
               break
   task.append("f")
   print(task)
   print(visited)
   server.send_task(task)

В приведенном коде используется функция find_optimal_route, которая определяет последовательность поворотов, необходимых беспилотнику для посещения всех поврежденных участков дорожного покрытия.

def find_optimal_route(start, targets):
   print(targets)
   targets = set(targets)
   visited = []
   current_zone = start
   route = [current_zone]
   while targets:
       best_next_zone = None
       best_path = None
       min_turns = 1e9
       for target in targets:
           path = find_shortest_path(current_zone, target)
           turns = len(path)
           if turns < min_turns:
               min_turns = turns
               best_next_zone = target
               best_path = path
       if best_next_zone:
           route.extend(best_path[1:])
           for pt in find_all_points(best_next_zone, damage_points):
               visited.append((len(route), pt))
           targets.remove(best_next_zone)
           current_zone = best_next_zone
   path = find_shortest_path(current_zone, stop_zone)
   route.extend(path[1:])
   return route, visited

Для работы с точками и поиска пути через них используются зоны. Зоны — это абстракция, представляющая каждый участок дорожной полосы между двумя перекрестками. Чтобы определить, в какой именно зоне находится конкретная точка, используется заранее сформированный json-файл с соответствием пикселей изображения и id конкретных зон.

def get_zone(self):
   nearest_zone = None
   min_distance = 1e9
   percent_position = 0
   for zone_id, points in lines.items():
       points = np.array(points)
       distances = np.linalg.norm(points - np.array((self.x,
                                                   self.y)), axis=1)
       min_idx = np.argmin(distances)
       min_dist = distances[min_idx]
       if min_dist < min_distance:
           min_distance = min_dist
           nearest_zone = zone_id
           percent_position = (min_idx / (len(points) - 1)) * 100
   return nearest_zone, float(round(percent_position, 2))

Ниже приведен код еще двух функций find_shortest_path и find_all_points, они используются при формировании задания для беспилотника.

find_all_points находит все поврежденные точки в заданной зоне и возвращает их номера, отсортированные по порядку расположения в зоне.

find_shortest_path находит кратчайший путь от начальной точки до целевой точки, используя алгоритм поиска в ширину (BFS). Возвращает список зон, составляющих путь. Граф, в котором осуществляется поиск, задается словарем.

graph = {
   "6": [("10", "STRAIGHT"), ("9", "LEFT")],
   "7": [("15", "RIGHT")],
   "8": [("7", "RIGHT"), ("10", "LEFT")],
   "9": [("13", "RIGHT"), ("16", "LEFT")],
   "10": [("14", "LEFT")],
   "11": [("9", "RIGHT"), ("7", "STRAIGHT")],
   "13": [("11", "RIGHT")],
   "14": [("16", "STRAIGHT"), ("8", "LEFT")],
   "15": [("8", "RIGHT"), ("13", "STRAIGHT")],
   "16": [("6", "LEFT")]
}
def find_shortest_path(start, target):
   queue = deque([(start, [], None)])
   visited = set()
   while queue:
       current_zone, path, last_direction = queue.popleft()
       if current_zone == target:
           return path + [current_zone]
       if current_zone in visited:
           continue
       visited.add(current_zone)
       for neighbor, direction in graph.get(current_zone, []):
           if neighbor not in visited:
               queue.append((neighbor, path + [current_zone], direction))
   return []
def find_all_points(zone, damage_points):
   data = []
   for point in damage_points:
       if point.get_zone()[0] == zone:
           data.append((point.number, point.get_zone()[1]))
   return [f"{dt[0]}" for dt in sorted(data, key=lambda x: x[1])]

Таким образом, беспилотный автомобиль может запрашивать с сервера задание, состоящее из последовательности поворотов и id точек, которые необходимо посетить. Беспилотник делает это до тех пор, пока не получит его, а затем формирует список маневров, которые необходимо совершить, и список id точек для посещения.

data = server.get_task()
while len(data) == 0:
   data = server.get_task()
   time.sleep(0.5)
algorithm = []
points = []
points_cnt = 0
prev_r = 0
prev_l = 0
for step in data:
   if step == "STRAIGHT":
       algorithm.append(CROSS_STRAIGHT)
   elif step == "RIGHT":
       algorithm.append(CROSS_RIGHT)
   elif step == "LEFT":
       algorithm.append(CROSS_LEFT)
   else:
       points.append(step)

В основном цикле обработки кадров, аналогично БПА-3 выполнено обнаружение поврежденных участков дороги.

if detect_yolo(trans_perspective(frame, TRAP, RECT, SIZE)):
   detect_count += 1
   print("temp detected", detect_count)
   if time.time() - last_acur_detect > 2 and detect_count > 2:
       last_acur_detect = time.time()
       print("DETECTED!")
       server.visit_point(points[points_cnt])
       points_cnt += 1
       # Получение и вывод данных от акселерометра
       client.send()
       response = client.get_output()
       print(response)
   if detect_count > 2:
       last_acur_detect = time.time()
   last_detect = time.time()

При обнаружении очередного поврежденного участка дороги беспилотный автомобиль увеличивает счетчик посещенных участков — по номеру встреченного участка находит его идентификатор в заранее полученном списке, а затем отправляет серверу сообщение о его посещении. Сервер обновляет счетчик для данного участка, и эта информация отображается на карте, так как карта постоянно запрашивает сервер о статусе точек и обновляет их отображение.

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

В решение для БПА-5 необходимо внести изменение. Из всех точек (поврежденных участков) необходимо выделить ту, на которой необходимо остановиться и взять пробы дорожного полотна. После составлении задания для беспилотника ему передается список c id точек посещения. Точка, на которой необходимо взять пробы, отмечается отдельно! Метка добавляется в функции plan_robot_route.

while vis_cnt < len(visited) and i + 1 == visited[vis_cnt][0] - 1:
   if freeze_zone == visited[vis_cnt][1]:
       task.append(visited[vis_cnt][1] + "_f")
   else:
       task.append(visited[vis_cnt][1])
   vis_cnt += 1
break

Беспилотный автомобиль, выполняя задание, проверяет, к какой точке он прибыл, и если это точка для взятия проб, переходит в специальное состояние FREEZE.

if detect_yolo(trans_perspective(frame, TRAP, RECT, SIZE)) and not need_we_stop(frame):
   detect_count += 1
   print("temp detected", detect_count)
   if time.time() - last_acur_detect > 2 and detect_count > 4:
       last_acur_detect = time.time()
       print("DETECTED!")
       point = points[points_cnt]
       if "_f" in point:
           print("collecting samples of the road surface")
           point = point.split("_")[0]
           STATE = FREEZE
           freeze_start = time.time()
       server.visit_point(point)
       points_cnt += 1

Специальное состояние обрабатывается так, чтобы беспилотник остановился на 2 с.

if STATE == FREEZE and time.time() - freeze_start > 2:
   STATE = GO
if STATE != STOP and STATE != FREEZE:
   arduino.set_speed(CAR_SPEED)
   arduino.set_angle(angle)
else:
   arduino.stop()

Функции поиска поврежденных участков дороги аналогичны БПА-3. Функции обнаружения зоны приема проб ЛВА и остановки в ней аналогичны БПА-2.

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для определения оставшегося срока службы дорожного участка лучше всего использовать нейронные сети. Анализируя статистические наборы данных, становится ясно, что дорожное полотно разрушается по восьми различным сценариям.

Трудность заключается в том, что в некоторых сценариях на начальном или конечном этапе жизни дорожного полотна показания акселерометра неизменны. Когда половина данных не содержит вариаций, модель не получает полезных градиентов для обучения на этой части.

Такое положение может приводить к тому, что оптимизатор «перестанет обращать внимание» на эту часть данных или будет пытаться компенсировать отсутствие различий в сигналах за счет искусственного завышения весов на изменяемой части. Это снизит общую обобщающую способность модели и не позволит достичь необходимой точности. Для преодоления проблемы участникам следует удалить из датасета неинформативные данные.

Задачу можно решить с помощью единственной модели машинного обучения, которая будет одновременно выполнять классификацию сценариев и регрессию для каждого из них. На начальном этапе обработки данных можно объединить представление классификационной ветви с представлением энкодера через конкатенацию, что позволит регрессионной ветви адаптировать свои предсказания в зависимости от идентифицированного сценария разрушения. Однако такой подход требует глубоких знаний в области машинного обучения и большого объема данных для обучения.

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

Ниже представлен код для формирования архитектуры классификатора, его обучения и сохранения обученной модели. Параметры обучения подбираются экспериментально.

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, Dropout
# Параметры
SEQ_LENGTH = 200
NUM_CLASSES = 8
# Создаем модель классификатора с использованием одномерных сверточных слоев.
input_shape = (SEQ_LENGTH, 1)
model = Sequential(name="Conv1D_Classifier")
model.add(Conv1D(filters=32, kernel_size=3,
                     activation='relu', padding='same', input_shape=input_shape))
model.add(MaxPooling1D(pool_size=2))
model.add(Dropout(0.3))
model.add(Conv1D(filters=64, kernel_size=3, activation='relu', padding='same'))
model.add(MaxPooling1D(pool_size=2))
model.add(Dropout(0.3))
model.add(Conv1D(filters=128, kernel_size=3, activation='relu', padding='same'))
model.add(MaxPooling1D(pool_size=2))
model.add(Dropout(0.3))
model.add(Flatten())
model.add(Dense(64, activation='relu'))
# Выходной слой: NUM_CLASSES нейронов, softmax активация для классификации
model.add(Dense(NUM_CLASSES, activation='softmax'))
# learning_rate подбирается экспериментально
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
             loss='sparse_categorical_crossentropy',
             metrics=['accuracy'])
model.summary()
# Параметры обучения
epochs = 10
batch_size = 10
# Обучение модели
history = model.fit(
   X_train, Y_train,
   epochs=epochs,
   batch_size=batch_size,
   validation_data=(X_test, Y_test))
# Сохранение модели
model.save("Class.h5")

Модель различает восемь классов сценариев разрушения, и для каждого обучается регрессионная модель, при этом их архитектуры будут одинаковы. Параметры обучения для каждой модели подбираются экспериментально. Ниже представлен код создания, обучения и сохранения регрессионной модели.

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, Dropout
# Параметры генератора
SEQ_LENGTH = 200
BATCH_SIZE = 32
# Создаем модель для регрессии.
input_shape = (SEQ_LENGTH, 1)
model = Sequential(name="Conv1D_Regressor")
model.add(Conv1D(filters=32, kernel_size=3, activation='relu', padding='same', input_shape=input_shape))
model.add(MaxPooling1D(pool_size=2))
model.add(Dropout(0.3))
model.add(Conv1D(filters=64, kernel_size=3, activation='relu', padding='same'))
model.add(MaxPooling1D(pool_size=2))
model.add(Dropout(0.3))
model.add(Conv1D(filters=128, kernel_size=3, activation='relu', padding='same'))
model.add(MaxPooling1D(pool_size=2))
model.add(Dropout(0.3))
model.add(Flatten())
model.add(Dense(64, activation='relu'))
model.add(Dense(1, activation='linear'))  # выходной нейрон для регрессионного предсказания
# learning_rate подбирается экспериментально
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
             loss='mse',
             metrics=['mae'])
model.summary()
# Параметры обучения
steps_per_epoch = 100  # батчей на эпоху
epochs = 10
# Обучение модели
history = model.fit(
   X_train, Y_train,
   epochs=epochs,
   batch_size=batch_size,
   validation_data=(X_test, Y_test))
# Сохранение модели
model.save("Regress_0.h5")

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

Сначала беспилотник останавливается, затем загружается и используется классификатор. Он определяет, по какому сценарию разрушается участок дороги, далее используется регрессионная модель для конкретного сценария.

if detect_yolo(trans_perspective(frame, TRAP, RECT, SIZE)):
   detect_count += 1
   print("temp detected", detect_count)
   if time.time() - last_acur_detect > 2 and detect_count > 2:
       last_acur_detect = time.time()
       print("DETECTED!")
       server.visit_point(points[points_cnt])
       points_cnt += 1
       # Останавливаем беспилотник
       arduino.stop()
       # Получаем данные от акселерометра
       client.send()
       response = client.get_output()
       # print(response)
       # Загружаем модель классификатора из файла
       model = load_model('Class.h5')
       X_test = string_to_numpy_array(strintg_chis)
       X_test = np.expand_dims(X_test, axis=1)
       X_test = np.expand_dims(X_test, axis=0)
       # Определяем класс сценария разрушения
       y_pred = model.predict(X_test)
       predicted_classes = np.argmax(y_pred, axis=1)
       # print("Предсказанный класс:", predicted_classes[0])
       next_model_name = ["Regress_1.h5", "Regress_2.h5", "Regress_3.h5",
                          "Regress_4.h5", "Regress_5.h5", "Regress_6.h5",
                          "Regress_7.h5", "Regress_8.h5"]
       # Загружаем регрессионную модель для конкретного сценария
       model = load_model(next_model_name[predicted_classes[0]])
       # Определяем оставшийся срок службы и выводим в терминал
       y_pred = model.predict(X_test)
       y_pred = round(y_pred[0][0])
       print(y_pred)

При подготовке данных от акселерометра для передачи на вход нейронной сети используется функция, переводящая строку чисел разделенных пробелами в массив numpy:

# Функция для перевода данных от акселерометра к массиву numpy
def string_to_numpy_array(data_string: str, dtype=float) -> np.ndarray:
    string_numbers = data_string.split()
    numeric_values = list(map(dtype, string_numbers))
    numpy_array = np.array(numeric_values)
    return numpy_array

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для выполнения этого задания участникам необходимо разобраться в предоставленном им API управления промышленным манипулятором.

Для каждого действия, которое может совершать манипулятор, нужно организовать отдельные функции. Функция capture_sample применяется для захвата пробы с конкретным номером.

def capture_sample(robot, sample_number):
    #   Функция для захвата пробы с указанным номером
        sample_positions = {
       1: (450, -400, 95),
       2: (550,-400, 95),
       3: (650, -400, 95),
       4: (650,-300, 95),
       5: (550, -300, 95),
       6: (650,-200,95 ),
       7: (650, 260,95),
       8: (650,360,95),
       9: (550, 360,95),
       10: (650,460,95),
       11: (550, 460,95),
       12: (450,460,95),    }
   if sample_number not in sample_positions:
       print("Некорректный номер пробы")
       return False
   x, y, z = sample_positions[sample_number]
   robot.move("Robot1_1", 489, -131.1, 424, 0, 0)
   # Перемещение к пробе
   robot.move("Robot1_1", x, y, z, 0, 0)
   time.sleep(2)  # Ожидание завершения движения
   # Захват пробы
   robot.move("Robot1_1", x, y, z, 0, 1)  # Активировать захват
   time.sleep(1)
   return True

Функция transport_to_camera отвечает за перемещение пробы к объективу камеры.

def transport_to_camera(robot):
   #Функция для транспортировки пробы к камере.
   # Координаты камеры
   camera_position = (670, -511, 311.5)
   # Перемещение к камере
   robot.move("Robot1_1", *camera_position, 0, 1)
   time.sleep(2)
   # Получение изображения с камеры
   image_byte = robot.getCamera1Image()
   if image_byte:
       image_np = np.frombuffer(image_byte, np.uint8)
       image_np = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
       cv2.imshow("Camera Image", image_np)
       cv2.waitKey(1)
       time.sleep(5)  # Ожидание для просмотра изображения
       cv2.destroyAllWindows()
   else:
       print("Ошибка получения изображения с камеры")
       return False
   return True

Функция return_sample возвращающая пробу на исходную позицию.

    def return_sample(robot, sample_number):
    # Функция для возврата пробы на место.

   sample_positions = {
        1: (450, -400, 95),
       2: (550,-400, 95),
       3: (650, -400,95),
       4: (650,-300, 95),
       5: (550, -300, 95),
       6: (650,-200, 95),
       7: (650, 260,95),
       8: (650,360,95),
       9: (550, 360,95),
       10: (650,460,95),
       11: (550, 460,95),
       12: (450,460,95),}
   if sample_number not in sample_positions:
       print("Некорректный номер пробы")
       return False
   x, y, z = sample_positions[sample_number]
   # Перемещение к месту возврата
   robot.move("Robot1_1", x, y, z, 0, 1)
   time.sleep(2)
   # Отпускание пробы
   robot.move("Robot1_1", x, y, z, 0, 0)  # Деактивировать захват
   time.sleep(1)
   return True

Основная логика работы реализована в функции main. Она запускает цикл, который продолжается, пока не будет выполнено одно успешное испытание. У пользователя запрашивается номер пробы (образца). Проводится проверка корректности номера, после чего вызываются функции capture_sample, transport_to_camera и return_sample.

Если какая-то из функций возвращает False, выводится сообщение об ошибке и продолжается цикл. В случае, когда все операции выполнены успешно, счетчик увеличивает число успешных запусков и выводит сообщение об успешном завершении испытания. После успешного завершения всех операций выводится сообщение о завершении задания.

def main():
   robot = MCX()
   successful_runs = 0
   while successful_runs < 1:
       # Запрос номера пробы
       sample_number = int(input("Введите номер пробы (1-12): "))
       if sample_number < 1 or sample_number > 12:
           print("Некорректный номер пробы.                 Введите число от 1 до 12.")
           continue
       # Захват пробы
       if not capture_sample(robot, sample_number):
           print("Ошибка захвата пробы")
           continue
       # Транспортировка к камере
       if not transport_to_camera(robot):
           print("Ошибка транспортировки к камере")
           continue
       # Возврат пробы на место
       if not return_sample(robot, sample_number):
           print("Ошибка возврата пробы")
           continue
       # Успешное выполнение
       print(f"Испытание {successful_runs + 1} успешно завершено")
       successful_runs += 1
   print("Задание выполнено успешно 2 раза подряд.")
if __name__ == "__main__":
   main()

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

При решении этой подзадачи используются программы из предыдущей подзадачи.

Необходимо сократить функцию transport_to_camera и вынести в функцию capture_images процесс получения восемь изображений пробы под разными углами.

def transport_to_camera(robot):
   #Функция для транспортировки пробы к камере.
   camera_position = (670, -511, 311.5)   # Координаты камеры
   # Перемещение к камере
   robot.move("Robot4_1", *camera_position, 0, 1)
   while robot.getManipulatorStatus() == 1:  # Ожидание завершения движения
       time.sleep(0.1)
   return True
def capture_images(robot):
   #Функция для получения 8 изображений пробы под разными углами.
   angles = [0, 45, 90, 135, 180, -135, -90, -45]  # Углы для поворота (в пределах [-180°, 180°])
   images = []
   for angle in angles:
       # Поворот кисти манипулятора на заданный угол
       robot.move("Robot4_1", 670, -511, 311.5, angle, 1)
       while robot.getManipulatorStatus() == 1:  # Ожидание завершения движения
           time.sleep(0.1)
       # Получение изображения с камеры
       image_byte = robot.getCamera1Image()
       if image_byte:
           image_np = np.frombuffer(image_byte, np.uint8)
           image_np = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
           images.append(image_np)
           # Отображение изображения в отдельном окне
           window_name = f"Camera Image {angle}°"
           cv2.imshow(window_name, image_np)
           cv2.waitKey(500)  # Кратковременное отображение изображения
       else:
           print(f"Ошибка получения изображения под углом {angle}°")
           return False
   # Ожидание нажатия клавиши для закрытия окон
   print("Нажмите любую клавишу, чтобы закрыть окна с изображениями.")
   cv2.waitKey(0)
   cv2.destroyAllWindows()
   return True

Функция main теперь выглядит следующим образом.

def main():
   robot = MCX()
   successful_runs = 0
   while successful_runs < 1:  # Достаточно одного успешного выполнения
       # Запрос номера пробы
       sample_number = int(input("Введите номер пробы (1-12): "))
       if sample_number < 1 or sample_number > 12:
           print("Некорректный номер пробы. Введите число от 1 до 12.")
           continue
       # Захват пробы
       if not capture_sample(robot, sample_number):
           print("Ошибка захвата пробы")
           continue
       # Транспортировка к камере
       if not transport_to_camera(robot):
           print("Ошибка транспортировки к камере")
           continue
       # Осмотр пробы с разных ракурсов
       if not capture_images(robot):
           print("Ошибка получения изображений")
           continue
       # Возврат пробы на место
       if not return_sample(robot, sample_number):
           print("Ошибка возврата пробы")
           continue
       # Успешное выполнение
       print(f"Испытание успешно завершено")
       successful_runs += 1

Если в процессе выполнения задания появляются ошибки, и манипулятор не выполнит какую-то из промежуточных операций, то выполнение задания начинается заново.

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения подзадачи нужно использовать код из подзадачи ЛВА-2.

Дополнительно необходимо организовать функцию record_video для записи и сохранения видеофайла.

def record_video(robot, sample_number, output_folder):
# Функция для записи видео с поворотом пробы на 360°.
   # Создаем папку для сохранения видео, если ее нет
   if not os.path.exists(output_folder):
       os.makedirs(output_folder)
   # Настройки записи видео
   video_filename = os.path.join(output_folder,
     f"sample_{sample_number}.avi")
   fourcc = cv2.VideoWriter_fourcc(*'MJPG')
   fps = 10  # Частота кадров
   frame_size = (640, 480)
   # Размер кадра (зависит от разрешения камеры)
   # Инициализация VideoWriter
   out = cv2.VideoWriter(video_filename, fourcc, fps, frame_size)
   # Углы для поворота (360° с шагом 45°)
   angles = [-135, -90, -45, 0, 45, 90, 135, 180]
   for angle in angles:
       # Поворот кисти манипулятора на заданный угол
       robot.move("Robot4_1", 670, -511, 311.5, angle, 1)
       while robot.getManipulatorStatus() == 1:
           # Ожидание завершения движения
           time.sleep(1)
       # Получение изображения с камеры
       image_byte = robot.getCamera1Image()
       if image_byte:
           image_np = np.frombuffer(image_byte, np.uint8)
           image_np = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
           out.write(image_np)  # Запись кадра в видео
       else:
           print(f"Ошибка получения изображения под углом {angle}°")
           return False
   # Завершение записи видео
   out.release()
   print(f"Видео для пробы {sample_number} сохранено в
        {video_filename}")
   return True

Логику работы функции main необходимо скорректировать для получения видеороликов с обзором четырех проб.

def main():
   robot = MCX()
   output_folder = "videos"  # Папка для сохранения видео
   # Запрос 4 номеров проб
   sample_numbers = []
   for i in range(4):
       sample_number = int(input(f"Введите номер пробы
                                                 {i + 1} (1-12): "))
       if sample_number < 1 or sample_number > 12:
           print("Некорректный номер пробы. Введите число от 1 до 12.")
           return
       sample_numbers.append(sample_number)
   # Обработка каждой пробы
   for sample_number in sample_numbers:
       # Захват пробы
       if not capture_sample(robot, sample_number):
           print(f"Ошибка захвата пробы {sample_number}")
           continue
       # Транспортировка к камере
       if not transport_to_camera(robot):
           print(f"Ошибка транспортировки пробы
            {sample_number} к камере")
           continue

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения задачи нужно получить достаточное количество статистических наборов данных за БПА-5, БПА-6, ЛВА-3, ЛВА-4, БПЛА-4 и БПЛА-5, а также установить соответствие между пробами дорожного полотна в ЛВА и их классами. В статистических наборах данных содержатся схематичные изображения образцов дорожного полотна с обозначенными классами.

После того как соответствие установлено, следует собрать обучающий датасет для классификатора. Пробы дорожного полотна имеют цилиндрическую форму, поэтому для определения типа пробы необходимы их изображения со всех сторон. Для сбора датасета удобно использовать изображения проб с разных ракурсов, получаемые в ЛВА-2. Из каждого вырезается участок непосредственно с пробой, а затем из них создается новое изображение, которое передается на вход нейросети для классификации. Функция extract_and_concatenate его возвращает.

def extract_and_concatenate(image_list):
   """
   Извлекает из каждого изображения в списке область [150:500, 400:570]
   и склеивает их по горизонтали, возвращая результат как новое изображение.    """
   pieces = []
   for img in image_list:
       # Проверяем, что изображение имеет достаточную размерность
       if img.shape[0] < 500 or img.shape[1] < 570:
           raise ValueError("Размер изображения меньше требуемых для вырезания области [150:500, 400:570].")
       # Извлекаем область. С помощью slice: [150:500, 400:570]
       piece = img[150:500, 400:570].copy()
       # cv2.imshow("Piece", piece)
       # cv2.waitKey(0)
       pieces.append(piece)
   # Склейка по горизонтали, если pieces не пустой
   if pieces:
       concatenated_image = np.hstack(pieces)
   else:
       concatenated_image = None
   return concatenated_image

Ниже представлен код для формирования архитектуры классификатора, его обучения и сохранения обученной модели. Параметры обучения подбираются экспериментально.

import numpy as np
import cv2
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv2D, MaxPooling2D, Flatten, Dense, Dropout, BatchNormalization
from sklearn.model_selection import train_test_split
# Параметры исходного изображения
orig_height = 350
orig_width = 1360
channels = 3
# Множитель для уменьшения изображения втрое (по оси H и W)
scale = 1 / 3
# Вычисляем размеры изображения после масштабирования
resized_height = int(orig_height * scale)
resized_width = int(orig_width * scale)
# Количество классов
num_classes = 5
# Создание модели классификатора
input_shape = (resized_height, resized_width, channels)
model = Sequential(name="Image_Classifier")
model.add(Conv2D(32, (3, 3), activation='relu', padding="same",
    input_shape=input_shape))
model.add(BatchNormalization())
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(Dropout(0.25))
model.add(Conv2D(64, (3, 3), activation='relu', padding="same"))
model.add(BatchNormalization())
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(Dropout(0.25))
model.add(Conv2D(128, (3, 3), activation='relu', padding="same"))
model.add(BatchNormalization())
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(Dropout(0.25))
model.add(Flatten())
model.add(Dense(128, activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(num_classes, activation='softmax'))
# выход: 5 классов
model.compile(optimizer=tf.keras.optimizers.
Adam(learning_rate=1e-3),
loss='sparse_categorical_crossentropy', metrics=['accuracy'])
model.summary()
# Обучаем модель
epochs = 50
batch_size = 32
history = model.fit(X_train, y_train, epochs=epochs, batch_size=batch_size,
validation_data=(X_test, y_test))
# Сохранение модели
model.save("Class.h5")

Функция predict_sample_type использует список изображений, полученный в capture_images из ЛВА-2. Из списка формируется изображение для передачи на вход нейросетевому классификатору.

def predict_sample_type(images_list):
   model = load_model('Class.h5')
   image = extract_and_concatenate(images_list)
   resized_height = int(350 / 3)
   resized_width = int(1360 / 3)
   single_img = cv2.resize(image, (resized_width, resized_height), interpolation=cv2.INTER_LINEAR)
   single_img = single_img / 255.0
   # Добавляем измерение батча (получим форму (1, resized_height, resized_width, 3))
   single_img_batch = np.expand_dims(single_img, axis=0)
   # Получаем предсказание от модели (вероятности для каждого класса)
   prediction = model.predict(single_img_batch)
   # Определяем класс с максимальной вероятностью
   predicted_class = np.argmax(prediction, axis=1)[0]
   print("Тип пробы:", predicted_class)

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения задачи необходимо:

• безошибочно подключить соединительные провода к электронным модулям квадрокоптера;
• подключиться к бортовому компьютеру беспилотника через SSH;
• скопировать файлы базового программного кода на бортовой компьютер и запустить их для исполнения.

Эти действия выполняются по предоставленным инструкциям.

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения задачи требуется разработать два узла в системе ROS. Первый узел будет отвечать за управление перемещением квадрокоптера (далее — полетное задание), а второй — за распознавание и подсчет поврежденных участков дорожного покрытия.

Дрон будет перемещаться в системе координат ультразвуковой навигационной системы «Геоскан Локус 2», поэтому для составления маршрута полета необходимо задать определенные точки. Они должны быть расположены таким образом, чтобы фотографии, сделанные в этих точках, имели частичное перекрытие с предыдущими кадрами. Такой подход необходим для последующей фильтрации обнаруженных повреждений, которые могут встречаться на соседних кадрах.

Для определения необходимых координат можно использовать самый простой метод – измерение их вручную, перемещая квадрокоптер по полигону. Текущие координаты квадрокоптера можно получить, выполнив следующую команду в SSH-терминале:

rostopic echo /geoscan/navigation/local/position

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

Необходимо организовать связь между двумя узлами. Оптимальным способом связи является использование сервиса. Узел распознавания будет выступать в роли сервера сервиса, а узел полетного задания — клиентом. При достижении очередной точки маршрута узел полетного задания будет отправлять запрос на обработку кадра.

Следует разработать сервис для вывода статистики полета, включая количество поврежденных участков дорожного покрытия.

Код полетного задания.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import rospy
from rospy import ServiceProxy
from std_srvs.srv import Empty, Trigger
from gs_flight import FlightController, CallbackEvent
from gs_board import BoardManager
from gs_navigation import NavigationManager

rospy.init_node("flight_test_node")  # инициализируем узел

# получаем координаты стартовой точки
nav = NavigationManager()
HOME_POINT = nav.lps.position()
HOME_POINT.z += 1.0
HOME_POINT = [HOME_POINT.x, HOME_POINT.y, HOME_POINT.z]

coordinates = [ # массив координат точек
   HOME_POINT,
   ... # необходимо вставить замеренные точки в формате [x, y, z]
   HOME_POINT
]

run = True # переменная отвечающая за работу программы
position_number = 0 # счетчик пройденных точек

def callback(event): # функция обработки событий Автопилота
   global ap
   global run
   global coordinates
   global position_number
   global detect_client
   global stat_client

   event = event.data
   if event == CallbackEvent.ENGINES_STARTED: # блок обработки события запуска двигателя
       print("engine started")
       ap.takeoff() # отдаем команду взлета
   elif event == CallbackEvent.TAKEOFF_COMPLETE: # блок обработки события завершения взлета
       print("takeoff complete")
       position_number = 0
       ap.goToLocalPoint(coordinates[position_number][0], coordinates[position_number][1], coordinates[position_number][2]) # отдаем команду полета в точку
   elif event == CallbackEvent.POINT_REACHED: # блок обработки события достижения точки
       print(f"point {position_number} reached")
       detect_client()
       position_number += 1 # наращиваем счетчик точек
       if position_number < len(coordinates): # проверяем количество текущее количество точек с количеством точек в полетном задании
           ap.goToLocalPoint(coordinates[position_number][0], coordinates[position_number][1], coordinates[position_number][2]) # отдаем команду полета в точку
       else:
           ap.landing() # отдаем команду посадки
   elif event == CallbackEvent.COPTER_LANDED: # блок обработки события приземления
       print("finish programm")
       run = False # прекращем программу
       response = stat_client()
       print(f"Кол-во поврежденных участков {response.message}")

detect_client = ServiceProxy("/get_photo", Empty) # клиент сервиса распознавания
stat_client = ServiceProxy("/get_stat", Trigger)
board = BoardManager() # создаем объект бортового менеджера
ap = FlightController(callback) # создаем объект управления полета

once = False # переменная отвечающая за первое вхождение в начало программы

while not rospy.is_shutdown() and run:
   if board.runStatus() and not once: # проверка подлкючения RPi к Пионеру
       print("start programm")
       ap.preflight() # отдаем команду выполенения предстартовой подготовки
       once = True

Для определения поврежденных участков требуется обучить нейросетевой классификатор Yolo v4-tiny.

Алгоритм работы узла детекции следующий:

1. После вызова сервиса распознавания производится получение изображения с камеры.
2. Осуществляется конвертирование в формат, совместимый с классификатором.
3. Выполняется распознавание объектов.
4. Выбирается наиболее вероятный класс на изображении.

Из условия задачи следует, что на одном участке дороги может быть только одно повреждение, что позволяет сузить область поиска. После этого происходит сравнение с объектами, обнаруженными на предыдущем кадре.

Если объекты совпадают, алгоритм считает, что это тот же самый объект; в противном случае в массив объектов добавляется метка класса нового объекта. При вызове сервиса статистики производится подсчет количества ненулевых значений в массиве детекции, после чего возвращается итоговый результат.

Программный код узла распознавания приведен ниже.

import rospy
from sensor_msgs.msg import Image
from std_srvs.srv import Empty, EmptyResponse
from std_srvs.srv import Trigger, TriggerResponse
import cv2
from cv_bridge import CvBridge
from rospy import Service
import numpy as np
rospy.init_node("detect_node")  # инициализируем узел
bridge = CvBridge()  # создаем объект преобразования сообщений ROS в OpenCV
weights_path = "yolov4-tiny-obj_best.weights"
config_path = "yolov4-tiny-obj.cfg"
net = cv2.dnn.readNet(config_path, weights_path)
yolo_model = cv2.dnn.DetectionModel(net)
all_object = []
# Выдает самый вероятный объект на изображении
def get_most_probable_class(class_ids, confidences):
   if len(class_ids) == 0 or len(confidences) == 0:
       return None
   most_probable_index = np.argmax(confidences)
   return class_ids[most_probable_index]
# Считает количество элементов в списке, которые не равны None.
def count_not_none(all_objects):
   return len(list(filter(lambda x: x is not None, all_objects)))
def detect_handler(request):
   global bridge
   global yolo_model
   global all_object
   image_msg = rospy.wait_for_message("/pioneer_max_camera/image_raw", Image)
   try:
       cv_image = bridge.imgmsg_to_cv2(image_msg, "bgr8")
       class_ids, scores, boxes = yolo_model.detect(cv_image, 0.6, 0.4)
       most_class = get_most_probable_class(class_ids, scores)
       if (most_class is None) or (len(all_object) == 0):
           all_object.append(most_class)
       elif most_class == all_object[-1]:
           all_object.append(None)
       else:
           all_object.append(most_class)
   except:
       pass
   return EmptyResponse()
def stat_handler(request):
   global all_object
   response = TriggerResponse()
   response.success = True
   response.message = str(count_not_none(all_object))
   return response

detect_server = Service("/get_photo", Empty, detect_handler)  # создаем сервис для получения фотографии
stat_server = Service("/get_stat", Trigger, stat_handler)
rospy.spin()

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для решения подзадачи необходимо внести изменения в полетное задание из БПЛА-1, исключив вывод статистики поиска объектов, а также модифицировать узел обнаружения повреждений. В функции detect_handler узла обнаружения, после добавления обнаруженного объекта в массив всех найденных объектов, следует добавить следующий код для вывода распознанного повреждения.

if all_object[-1] is not None:
    print(f"Обнаруженный тип повреждения: {all_object[-1]}")

Необходимо удалить сервис статистики.

Код измененного полетного задания.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import rospy
from rospy import ServiceProxy
from std_srvs.srv import Empty
from gs_flight import FlightController, CallbackEvent
from gs_board import BoardManager
from gs_navigation import NavigationManager
rospy.init_node("flight_test_node")  # инициализируем узел
# получаем координаты стартовой точки
nav = NavigationManager()
HOME_POINT = nav.lps.position()
HOME_POINT.z += 1.0
HOME_POINT = [HOME_POINT.x, HOME_POINT.y, HOME_POINT.z]
coordinates = [ # массив координат точек
   HOME_POINT,
   ... # необходимо вставить измеренные точки в формате [x, y, z]
   HOME_POINT
]
run = True # переменная отвечающая за работу программы
position_number = 0 # счетчик пройденных точек
def callback(event): # функция обработки событий Автопилота
   global ap
   global run
   global coordinates
   global position_number
   global detect_client
   event = event.data
   if event == CallbackEvent.ENGINES_STARTED: # блок обработки события запуска двигателя
       print("engine started")
       ap.takeoff() # отдаем команду взлета
   elif event == CallbackEvent.TAKEOFF_COMPLETE: # блок обработки события завершения взлета
       print("takeoff complete")
       position_number = 0
       ap.goToLocalPoint(coordinates[position_number][0], coordinates[position_number][1], coordinates[position_number][2]) # отдаем команду полета в точку
   elif event == CallbackEvent.POINT_REACHED: # блок обработки события достижения точки
       print(f"point {position_number} reached")
       detect_client()
       position_number += 1 # наращиваем счетчик точек
       if position_number < len(coordinates): # проверяем количество текущее количество точек с количеством точек в полетном задании
           ap.goToLocalPoint(coordinates[position_number][0],
                             coordinates[position_number][1],
                             coordinates[position_number][2]) # отдаем команду полета в точку
       else:
           ap.landing() # отдаем команду посадки
   elif event == CallbackEvent.COPTER_LANDED: # блок обработки события приземления
       print("finish programm")
       run = False # прекращем программу
detect_client = ServiceProxy("/get_photo", Empty) # клиент сервиса распознавания
board = BoardManager() # создаем объект бортового менеджера
ap = FlightController(callback) # создаем объект управления полета
once = False # переменная отвечающая за первое вхождение в начало программы
while not rospy.is_shutdown() and run:
   if board.runStatus() and not once: # проверка подключения RPi к Пионеру
       print("start programm")
       ap.preflight() # отдаем команду выполнения предстартовой подготовки
       once = True
   pass

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Участникам было предоставлено изображение полигона \(1208 \times 889\) px (пикселей).

Красным прямоугольником выделена полетная зона квадрокоптера.

Для выполнения задачи необходимо выполнить привязку координат ультразвуковой навигационной системы к пикселям изображения. Размер зоны задается системой навигации «Геоскан Локус» в метрах, где 0 координат находится в нижнем левом углу, ось X направлена горизонтально, а ось Y — вертикально.

Полетная зона имеет размеры \(3 \times 6\) м. Следовательно, длина 1 px (пикселя) изображения составляет приблизительно 0,0067 м в системе координат ультразвуковой навигации. Эта величина является константой и должна использоваться для расчета местоположения повреждения.

Для точного указания координат объекта требуется модификация функции его поиска таким образом, чтобы она возвращала смещение центра объекта относительно центра камеры. Полученные данные необходимо перевести в метры и прибавить к текущим координатам квадрокоптера.

Модифицированная функция поиска объекта.

# Выдает самый вероятный объект на изображении и координаты центра объекта относительно центра изображения с камеры
def get_most_probable_class(class_ids, confidences, boxes, image_shape):
   if len(class_ids) == 0 or len(confidences) == 0 or len(boxes) == 0:
       return None, None, None  # Если объекты не найдены, возвращаем (None, None, None)
   # Находим индекс максимальной уверенности
   most_probable_index = np.argmax(confidences)
   # Получаем соответствующий class_id и box
   class_id = class_ids[most_probable_index]
   box = boxes[most_probable_index]
   # Рассчитываем центр рамки
   x, y, w, h = box  # Координаты рамки: (x, y, ширина, высота)
   box_center_x = x + w / 2
   box_center_y = y + h / 2
   # Рассчитываем центр изображения
   image_height, image_width = image_shape
   image_center_x = image_width / 2
   image_center_y = image_height / 2
   # Рассчитываем смещение центра рамки относительно центра изображения
   center_offset_x = box_center_x - image_center_x
   center_offset_y = box_center_y - image_center_y
   center_offset = (center_offset_x, center_offset_y)
   return class_id, center_offset

Необходимо разработать функцию для перевода смещения из пикселей в метры. Для этого следует использовать высоту квадрокоптера в момент съемки, а также угол обзора камеры. На квадрокоптере «Геоскан Пионер» в модификации НТО установлена камера Raspberry Pi Camera Module 2, которая имеет следующие углы обзора: вертикальный угол – 48,8°, горизонтальный угол – 62,2°.

Функция конвертации смещения из пикселей в метры будет использовать данные о высоте квадрокоптера и углах обзора камеры.

Функция конвертации смещения из пикселей в метры.

def convert_pixels_to_meters(pixel_offset_x, pixel_offset_y, camera_height, camera_fov_h, camera_fov_v):
   # Переводим углы обзора из градусов в радианы
   fov_h_rad = np.radians(camera_fov_h)
   fov_v_rad = np.radians(camera_fov_v)

   # Рассчитываем фокусные расстояния камеры в пикселях
   focal_length_h = camera_height / np.tan(fov_h_rad / 2)
   focal_length_v = camera_height / np.tan(fov_v_rad / 2)

   # Переводим смещение из пикселей в метры
   x_meters = (pixel_offset_x * camera_height) / focal_length_h
   y_meters = (pixel_offset_y * camera_height) / focal_length_v

   return x_meters, y_meters

Для создания карты недостаточно сохранять только метку класса объекта, необходимо фиксировать и его положение. Для хранения карты необходимо использовать сервер из БПА-4. Функция упаковки данных в словарь.

from gs_navigation import NavigationManager
PIXEL_CONST = 0.0067
# Инициализирует NavigationManager для получения координат
nav = NavigationManager()
# Создает словарь с классом и смещением
def create_dict(class_id, object_offset, number):
   global nav
   if class_id is None:
       return None
   else:
       position = nav.lps.position()
       object_offset_x, object_offset_y = convert_pixels_to_meters(
           *object_offset,
           position.z,
           62.2,
           48.8
       )
       return {
           "type": class_id,
           "number": number,
           "count" : 0,
           "x": (object_offset_x + position.x) / PIXEL_CONST,
           "y": (object_offset_y + position.y) / PIXEL_CONST
       }

Модифицированная функция детектирования объекта.

# Считает количество элементов в списке, которые не равны None.
def count_not_none(all_objects):
  return len(list(filter(lambda x: x is not None, all_objects)))
def detect_handler(request):
   global bridge
   global yolo_model
   global all_object

   image_msg = rospy.wait_for_message("/pioneer_max_camera/image_raw", Image)
   try:
       cv_image = bridge.imgmsg_to_cv2(image_msg, "bgr8")
       class_ids, scores, boxes = yolo_model.detect(cv_image, 0.6, 0.4)
       most_class, object_offset = get_most_probable_class(
           class_ids,
           scores,
           boxes,
           cv_image.shape
       )
       object_dict = create_dict(
           most_class,
           object_offset,
           count_not_none(all_object)
       )
       if (object_dict is None) or (len(all_object) == 0):
           all_object.append(object_dict)
       elif all_object[-1] is None:
           all_object.append(object_dict)
       elif object_dict["type"] == all_object[-1]["type"]:
           all_object.append(None)
       else:
           all_object.append(object_dict)
       if all_object[-1] is not None:
           print(f"Обнаруженный тип повреждения: {all_object[-1]}")
   except:
       pass
   return EmptyResponse()

Функция отправки данных на сервер:

import requests
SERVER_URL = "http://172.16.65.50:8000"
def send_points(points):
   ans = requests.post(f"{SERVER_URL}/points", json=points)
   if ans.status_code != 200:
       return False
   return ans.json()["status"] == "OK"

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

def send_points_handler(request):
   global all_object
   send_points(all_object)
   return EmptyResponse()
send_points_server = Service("/send_points", Empty, send_points_handler)  # создаем сервис для отправки точек

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

def callback(event): # функция обработки событий Автопилота
   global ap
   global run
   global coordinates
   global position_number
   global detect_client
   event = event.data
   if event == CallbackEvent.ENGINES_STARTED: # блок обработки события запуска двигателя
       print("engine started")
       ap.takeoff() # отдаем команду взлета
   elif event == CallbackEvent.TAKEOFF_COMPLETE: # блок обработки события завершения взлета
       print("takeoff complete")
       position_number = 0
       ap.goToLocalPoint(coordinates[position_number][0],
                         coordinates[position_number][1],
                         coordinates[position_number][2]) # отдаем команду полета в точку
   elif event == CallbackEvent.POINT_REACHED: # блок обработки события достижения точки
       print(f"point {position_number} reached")
       detect_client()
       position_number += 1 # наращиваем счетчик точек
       if position_number < len(coordinates): # проверяем количество текущее количество точек с количеством точек в полетном задании
           ap.goToLocalPoint(coordinates[position_number][0],
                             coordinates[position_number][1],
                             coordinates[position_number][2]) # отдаем команду полета в точку
       else:
           ap.landing() # отдаем команду посадки
   elif event == CallbackEvent.COPTER_LANDED: # блок обработки события приземления
       print("finish programm")
       send_points_client()
       run = False # прекращем программу
   send_points_client = ServiceProxy("/send_points", Empty) # клиент сервиса распознавания

Файлы с решениями подзадач находятся в папке: https://disk.yandex.ru/d/NJ2o0pOxDW162A.

Для выполнения подзадачи следует внести изменения в сервер, разработанный в рамках подзадачи БПА-4. Требуется добавить функцию постановки задачи квадрокоптеру и получения этой задачи, а также функцию добавления и сохранения изображений, необходимых для выполнения подзадачи.

class ImageRequest(BaseModel):
  original: str
  processed: str
last_images = None
@app.post("/copter_task")
async def post_copter_task(request: PointsRequest):
  global copter_task
  copter_task = request
  return {"status": "OK", "detail": None}
@app.get("/copter_task")
async def get_copter_task():
  global copter_task
  return {"status": "OK", "detail": copter_task}
@app.post("/images")
async def post_images(request: ImageRequest):
  global last_images
  last_images = request
  return {"status": "OK", "detail": None}
@app.get("/images")
async def get_images():
  global last_images
  return {"status": "OK", "detail": last_images}

Необходимо модифицировать код приложения карты, разработанного в рамках подзадачи БПА-5. Требуется добавить функции отправки данных о полетной задаче на сервер в класс Server.

def post_copter_task(self, task: dict) -> bool:
      ans = requests.post(f"{self.base_url}/copter_task", json=task)
      if ans.status_code != 200:
          return False
      return ans.json()["status"] == "OK"

Далее нужно изменить функцию mouse_callback, переназначив правую кнопку мыши на отправку задания квадрокоптеру.

if event == cv2.EVENT_RBUTTONDOWN:
       id = input("Введите id точки")
       map_point = server.get_points()
       task = None
       for point in map_point:
           if point["id"] == id:
               task = point
       if task is not None:
           server.post_copter_task(task)

В полетном задании перед запуском двигателя необходимо добавить ожидание полетной задачи. Изменения приведены ниже.

import requests
coordinates = [
   HOME_POINT
]
def callback(event): # функция обработки событий Автопилота
   global ap
   global run
   global coordinates
   global position_number
   global detect_client
   event = event.data
   if event == CallbackEvent.ENGINES_STARTED: # блок обработки события запуска двигателя
       print("engine started")
       ap.takeoff() # отдаем команду взлета
   elif event == CallbackEvent.TAKEOFF_COMPLETE: # блок обработки события завершения взлета
       print("takeoff complete")
       position_number = 0
       ap.goToLocalPoint(coordinates[position_number][0], coordinates[position_number][1], coordinates[position_number][2]) # отдаем команду полета в точку
   elif event == CallbackEvent.POINT_REACHED: # блок обработки события достижения точки
       print(f"point {position_number} reached")
       position_number += 1 # наращиваем счетчик точек
       if position_number < len(coordinates): # проверяем количество текущее количество точек с количеством точек в полетном задании
           ap.goToLocalPoint(coordinates[position_number][0],
                             coordinates[position_number][1],
                             coordinates[position_number][2]) # отдаем команду полета в точку
       elif position_number == len(coordinates):
           detect_client()
           ap.goToLocalPoint(*HOME_POINT)
       else:
           ap.landing() # отдаем команду посадки
   elif event == CallbackEvent.COPTER_LANDED: # блок обработки события приземления
       print("finish programm")
       send_points_client()
       run = False # прекращаем программу
SERVER_URL = "http://172.16.65.50:8000"
def get_copter_task():
   ans = requests.get(f"{SERVER_URL}/copter_task")
   if ans.status_code != 200:
       return None
   return ans.json()["detail"]
PIXEL_CONST = 0.0067
while not rospy.is_shutdown() and run:
   if board.runStatus() and not once: # проверка подключения RPi к Пионеру
       print("start programm")
       while True:
           task = get_copter_task()
           if task is not None:
               coordinates.append([task["x"] * PIXEL_CONST, task["y"] * PIXEL_CONST, HOME_POINT[2]])
               break
       ap.preflight() # отдаем команду выполнения предстартовой подготовки
       once = True
   pass

Необходимо внести изменения в узел обнаружения дефектов. Требуется удалить функциональность сохранения объектов и добавить запрос к MQTT-серверу для получения обработанного изображения, а также реализовать отправку изображения на общий сервер. Функция отправки двух изображений на общий сервер.

import base64
def send_image_to_server(original_image, processed_image):
   # Кодируем исходное изображение в JPEG
   _, buffer_orig = cv2.imencode('.jpg', original_image)
   # Кодируем обработанное изображение в JPEG
   _, buffer_proc = cv2.imencode('.jpg', processed_image)

   # Конвертируем в base64
   img_orig_base64 = base64.b64encode(buffer_orig).decode('utf-8')
   img_proc_base64 = base64.b64encode(buffer_proc).decode('utf-8')

   # Отправляем на сервер
   response = requests.post(
       f"{SERVER_URL}/images",
       json={
           "original": img_orig_base64,
           "processed": img_proc_base64
       }
   )
   return response.status_code == 200

Исправленная функция сервиса детекции.

def detect_handler(request):
   global bridge

   image_msg = rospy.wait_for_message("/pioneer_max_camera/image_raw", Image)
   try:
       cv_image = bridge.imgmsg_to_cv2(image_msg, "bgr8")
       client.send(cv_image)
       output = client.get_output()

       if output is not None:
           send_image_to_server(cv_image, output)

   except Exception as e:
       print(f"Ошибка в detect_handler: {str(e)}")
   return EmptyResponse()

Необходимо разработать скрипт, который будет запускаться на компьютере оператора и выводить изображения, полученные с квадрокоптера. Код программы приведен ниже.

import cv2
import numpy as np
import requests
import base64
import time
SERVER_URL = "http://172.16.65.50:8000"
def base64_to_image(base64_string):
   # Декодируем base64 строку в байты
   img_data = base64.b64decode(base64_string)
   # Преобразуем байты в numpy массив
   nparr = np.frombuffer(img_data, np.uint8)
   img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
   return img
def main():
   # URL сервера

   while True:
       try:
           # Получаем данные с сервера
           response = requests.get(f"{SERVER_URL}/images")
           data = response.json()

           if data["status"] == "OK" and data["detail"] is not None:
               # Получаем изображения
               original_img = base64_to_image(data["detail"]["original"])
               processed_img = base64_to_image(data["detail"]["processed"])

               # Создаем окно с двумя изображениями рядом
               combined = np.hstack((original_img, processed_img))

               # Показываем изображения
               cv2.imshow("Original and Processed Images", combined)

               # Ждем нажатия клавиши (1 мс)
               if cv2.waitKey(1) & 0xFF == ord('q'):
                   break

           # Небольшая задержка перед следующим запросом
           time.sleep(0.1)

       except Exception as e:
           print(f"Ошибка: {e}")
           time.sleep(1)  # Ждем секунду перед повторной попыткой

   cv2.destroyAllWindows()
if __name__ == "__main__":
   main()