feat(module3): add simple competition workflow

This commit is contained in:
2026-09-15 21:14:44 +03:00
parent 4ba36d8b18
commit 6c5e23c86c
6 changed files with 309 additions and 0 deletions
+1
View File
@@ -7,3 +7,4 @@ reports/
!module3/models/latest.pt !module3/models/latest.pt
module3/runs/ module3/runs/
module3/predictions/ module3/predictions/
module3/training_new/
+4
View File
@@ -1,5 +1,9 @@
# Модуль В: датасет и YOLO # Модуль В: датасет и YOLO
Максимально простой сценарий на один день находится в [SIMPLE.md](SIMPLE.md).
Он не заменяет полный `module3.py`: там отдельно собраны нарезка видео,
локальная разметка, обучение через CLI, OpenCV-демо и проверка захвата.
Камера Webots: `/RMC1/arm95/camera_gripper/image_color`. На физическом RMC1 новая Камера Webots: `/RMC1/arm95/camera_gripper/image_color`. На физическом RMC1 новая
документация указывает `/RMC1/arm95/svcam/right/image/compressed` с типом документация указывает `/RMC1/arm95/svcam/right/image/compressed` с типом
`sensor_msgs/msg/Image` — несмотря на суффикс `compressed`. `sensor_msgs/msg/Image` — несмотря на суффикс `compressed`.
+112
View File
@@ -0,0 +1,112 @@
# Модуль В: простой вариант
## 1. Нарезать видео на кадры
Из корня репозитория:
```bash
mkdir -p module3/dataset/raw
ffmpeg -i video.mp4 -vf fps=2 module3/dataset/raw/frame_%05d.jpg
```
Двух кадров в секунду обычно достаточно. Удали смазанные и почти одинаковые
кадры. Оставь хотя бы 30–60 нормальных изображений.
## 2. Разметить локально
Один раз установить X-AnyLabeling в отдельное окружение:
```bash
python3 -m venv ~/.venvs/xanylabeling
~/.venvs/xanylabeling/bin/pip install -U uv
~/.venvs/xanylabeling/bin/uv pip install "x-anylabeling-cvhub[cpu]"
```
Открыть кадры:
```bash
~/.venvs/xanylabeling/bin/xanylabeling module3/dataset/raw \
--labels module3/dataset/classes.txt --autosave
```
Классы уже лежат в `dataset/classes.txt`:
```text
hammer
pliers
wrench
```
Нажми `R`, обведи каждый видимый инструмент прямоугольником, выбери класс,
затем `D` для следующего кадра. После разметки выбери `File -> Export
Annotations -> Export YOLO Annotations -> Detection`, укажи
`dataset/classes.txt`. В `dataset/raw/labels` должны появиться TXT-файлы.
Подготовить train/valid/test:
```bash
python3 module3/simple_prepare_dataset.py
```
Скрипт создаст `module3/training_new` и при повторном запуске пересоздаст только
эту папку.
## 3. Обучить через Ultralytics CLI
```bash
cd module3
source .venv/bin/activate
yolo detect train model=yolo11n.pt data=training_new/data.yaml epochs=30 patience=7 imgsz=512 batch=8 device=0 workers=4 project=training_new/runs name=tools
cp training_new/runs/tools/weights/best.pt models/latest.pt
```
Если не хватает видеопамяти, поменяй только `batch=8` на `batch=4`.
## 4. Показать работу модели в OpenCV
На тестовых фотографиях:
```bash
python simple_module_b.py training_new/test/images
```
На исходном видео:
```bash
python simple_module_b.py /путь/к/video.mp4
```
На обычной USB-камере:
```bash
python simple_module_b.py 0
```
На изображении будут bounding box, класс и confidence. Результаты также пишутся
в `logs/simple_module_b.log`. `Q` или `Esc` закрывает окно.
## 5. Открыть и закрыть захват
После запуска симулятора и контроллеров:
```bash
source /opt/ros/jazzy/setup.bash
source ~/ros2_ws/install/setup.bash
python3 module3/simple_gripper.py
```
Скрипт сначала открывает захват, по `Enter` закрывает и ещё по `Enter` снова
открывает. Команды сохраняются в `module3/logs/simple_gripper.log`.
Если нужно отдельно показать движение всей руки, используй выданный пример:
```bash
ros2 launch ar_webots_fms_ros2 moveit_api_example.launch.py
```
## Что сказать эксперту
«Я нарезал видео на кадры, разметил три класса bounding box-ами, разделил
датасет на train/valid/test и обучил YOLO11n через Ultralytics CLI. В OpenCV
показываю класс, confidence и bounding box. Управление захватом отправляет ROS 2
команду контроллеру RMC1, а события выводятся в терминал и сохраняются в лог».
+72
View File
@@ -0,0 +1,72 @@
#!/usr/bin/env python3
from datetime import datetime
from pathlib import Path
import time
import rclpy
from builtin_interfaces.msg import Duration
from control_msgs.action import GripperCommand
from rclpy.action import ActionClient
from rclpy.node import Node
from trajectory_msgs.msg import JointTrajectory, JointTrajectoryPoint
ACTION = "/RMC1/arm95/gripper_controller/gripper_cmd"
TOPIC = "/RMC1/arm95/gripper_trajectory_controller/joint_trajectory"
HERE = Path(__file__).resolve().parent
class Gripper(Node):
def __init__(self):
super().__init__("simple_gripper")
self.client = ActionClient(self, GripperCommand, ACTION)
self.publisher = self.create_publisher(JointTrajectory, TOPIC, 10)
def set_state(self, opened):
state = "open" if opened else "closed"
if self.client.wait_for_server(timeout_sec=2.0):
goal = GripperCommand.Goal()
goal.command.position = 0.04 if opened else 0.0
goal.command.max_effort = 20.0
future = self.client.send_goal_async(goal)
rclpy.spin_until_future_complete(self, future)
handle = future.result()
if not handle.accepted:
raise RuntimeError("Контроллер отклонил команду")
result = handle.get_result_async()
rclpy.spin_until_future_complete(self, result)
else:
message = JointTrajectory()
message.joint_names = ["left_joint", "right_joint"]
point = JointTrajectoryPoint()
value = 0.04 if opened else 0.0213
point.positions = [value, value]
point.time_from_start = Duration(sec=1)
message.points = [point]
for _ in range(5):
self.publisher.publish(message)
rclpy.spin_once(self, timeout_sec=0.1)
line = f"{datetime.now().isoformat(timespec='seconds')} | gripper={state}"
print(line)
with (HERE / "logs/simple_gripper.log").open("a", encoding="utf-8") as log:
log.write(line + "\n")
time.sleep(1)
def main():
rclpy.init()
node = Gripper()
try:
node.set_state(True)
input("Захват открыт. Enter — закрыть: ")
node.set_state(False)
input("Захват закрыт. Enter — снова открыть: ")
node.set_state(True)
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == "__main__":
main()
+61
View File
@@ -0,0 +1,61 @@
#!/usr/bin/env python3
import argparse
from datetime import datetime
from pathlib import Path
import cv2
from ultralytics import YOLO
HERE = Path(__file__).resolve().parent
def main():
parser = argparse.ArgumentParser()
parser.add_argument("source", help="Фото, видео, папка с фото или 0 для веб-камеры")
parser.add_argument("--weights", default=str(HERE / "models/latest.pt"))
parser.add_argument("--conf", type=float, default=0.25)
args = parser.parse_args()
source = int(args.source) if args.source.isdigit() else args.source
model = YOLO(args.weights)
log_path = HERE / "logs/simple_module_b.log"
log_path.parent.mkdir(exist_ok=True)
still_images = False
if isinstance(source, str):
path = Path(source)
still_images = path.is_dir() or path.suffix.lower() in {
".jpg",
".jpeg",
".png",
".bmp",
".webp",
}
with log_path.open("a", encoding="utf-8") as log:
results = model.predict(source=source, conf=args.conf, stream=True, verbose=False)
for result in results:
now = datetime.now().isoformat(timespec="seconds")
detections = []
if result.boxes is not None:
for box in result.boxes:
name = result.names[int(box.cls.item())]
confidence = float(box.conf.item())
detections.append(f"{name} {confidence:.2f}")
line = f"{now} | {', '.join(detections) if detections else 'ничего не найдено'}"
print(line)
log.write(line + "\n")
log.flush()
cv2.imshow("YOLO: q или Esc для выхода", result.plot())
key = cv2.waitKey(0 if still_images else 1) & 0xFF
if key in (ord("q"), 27):
break
cv2.destroyAllWindows()
if __name__ == "__main__":
main()
+59
View File
@@ -0,0 +1,59 @@
#!/usr/bin/env python3
import random
import shutil
from pathlib import Path
HERE = Path(__file__).resolve().parent
SOURCE = HERE / "dataset/raw"
LABELS = SOURCE / "labels"
OUTPUT = HERE / "training_new"
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
def main():
images = sorted(path for path in SOURCE.iterdir() if path.suffix.lower() in IMAGE_EXTENSIONS)
if not images:
raise SystemExit(f"Нет кадров в {SOURCE}")
if not LABELS.is_dir():
raise SystemExit(f"Сначала экспортируйте YOLO-разметку в {LABELS}")
random.Random(42).shuffle(images)
test_count = max(1, round(len(images) * 0.1)) if len(images) >= 3 else 0
valid_count = max(1, round(len(images) * 0.2)) if len(images) >= 3 else 0
parts = {
"test": images[:test_count],
"valid": images[test_count : test_count + valid_count],
"train": images[test_count + valid_count :],
}
if OUTPUT.exists():
shutil.rmtree(OUTPUT)
for part, part_images in parts.items():
image_dir = OUTPUT / part / "images"
label_dir = OUTPUT / part / "labels"
image_dir.mkdir(parents=True)
label_dir.mkdir(parents=True)
for image in part_images:
shutil.copy2(image, image_dir / image.name)
label = LABELS / f"{image.stem}.txt"
destination = label_dir / f"{image.stem}.txt"
if label.exists():
shutil.copy2(label, destination)
else:
destination.touch()
classes = (HERE / "dataset/classes.txt").read_text(encoding="utf-8").splitlines()
names = "\n".join(f" {number}: {name}" for number, name in enumerate(classes))
(OUTPUT / "data.yaml").write_text(
f"path: {OUTPUT}\ntrain: train/images\nval: valid/images\ntest: test/images\n\nnames:\n{names}\n",
encoding="utf-8",
)
print(f"Готово: train={len(parts['train'])}, valid={len(parts['valid'])}, test={len(parts['test'])}")
print(OUTPUT / "data.yaml")
if __name__ == "__main__":
main()