Разработка системы автоматической инвентаризации через камеры

Разработка системы автоматической инвентаризации через камеры

Автоматическая инвентаризация через камеры — задача непрерывного или периодического подсчёта товарных единиц без физического сканирования. Традиционная инвентаризация раз в квартал занимает целый день при закрытом магазине. Camera-based инвентаризация работает ежедневно или в реальном времени, не мешая работе.

Архитектура системы

Стационарные камеры (полки) → Periodic Analysis (каждые N часов)
    +
Мобильные/дроновые камеры → On-demand инвентаризация
    +
Point-of-sale данные → Reconciliation с физическим подсчётом
    ↓
Inventory Management System (ERP/WMS)

class AutoInventorySystem:
    def __init__(self, detector_path: str, inventory_db_path: str):
        self.detector = YOLO(detector_path)
        self.db = InventoryDatabase(inventory_db_path)

    def run_inventory_cycle(self, shelf_images: dict) -> InventoryReport:
        """
        shelf_images: {shelf_id: image} - фото всех полок
        """
        report = InventoryReport()

        for shelf_id, image in shelf_images.items():
            shelf_counts = self._count_shelf(image, shelf_id)
            report.add_shelf(shelf_id, shelf_counts)

        # Сравнение с ожидаемым остатком
        expected = self.db.get_expected_quantities()
        report.discrepancies = self._find_discrepancies(
            report.actual_counts, expected
        )

        # Автоматическое обновление в ERP
        self.db.update_inventory(report.actual_counts)

        return report

    def _count_shelf(self, image: np.ndarray,
                      shelf_id: str) -> dict:
        """Подсчёт товаров на одной полке"""
        detections = self.detector(image, conf=0.45)

        counts = {}
        for box in detections[0].boxes:
            sku = self.detector.model.names[int(box.cls)]
            counts[sku] = counts.get(sku, 0) + 1

        return counts

Обработка перспективных искажений

Для shelf-камер с широким углом — коррекция искажений и создание плоского вида полки:

def create_shelf_rectified_view(image: np.ndarray,
                                 shelf_corners: list,
                                 output_size: tuple = (2000, 400)) -> np.ndarray:
    """
    Плоское (top-down) представление полки для удобного анализа
    shelf_corners: 4 угла полки в изображении
    """
    pts_src = np.array(shelf_corners, dtype='float32')
    w, h = output_size
    pts_dst = np.array([
        [0, 0], [w - 1, 0],
        [w - 1, h - 1], [0, h - 1]
    ], dtype='float32')

    M = cv2.getPerspectiveTransform(pts_src, pts_dst)
    rectified = cv2.warpPerspective(image, M, (w, h))
    return rectified

Инвентаризация через дроны

Для складов с высокими стеллажами — дрон с камерой:

class DroneInventoryController:
    def __init__(self, drone_api, inventory_system):
        self.drone = drone_api
        self.inventory = inventory_system
        self.waypoints = []  # заранее запрограммированные точки съёмки

    async def run_inventory_mission(self) -> InventoryReport:
        images = {}
        await self.drone.takeoff()

        for waypoint in self.waypoints:
            await self.drone.fly_to(waypoint)
            await self.drone.stabilize(seconds=1.0)

            # Фото с нескольких ракурсов для лучшего покрытия
            for angle_offset in [0, -15, 15]:
                await self.drone.rotate(angle_offset)
                image = await self.drone.capture_image()
                images[f"{waypoint['shelf_id']}_{angle_offset}"] = image

        await self.drone.land()

        return self.inventory.run_inventory_cycle(images)

Интеграция с ERP/WMS системами

Обновление остатков через API:

import requests

class ERPIntegration:
    def __init__(self, erp_url: str, api_key: str):
        self.base_url = erp_url
        self.headers = {'Authorization': f'Bearer {api_key}'}

    def update_stock_levels(self, inventory: dict,
                             location_id: str) -> dict:
        """Обновление остатков в ERP системе"""
        stock_updates = [
            {
                'sku': sku,
                'quantity': qty,
                'location_id': location_id,
                'source': 'camera_inventory',
                'timestamp': get_iso_timestamp()
            }
            for sku, qty in inventory.items()
        ]

        response = requests.post(
            f'{self.base_url}/api/inventory/bulk-update',
            json={'updates': stock_updates},
            headers=self.headers
        )
        return response.json()

Reconciliation: сверка с системой учёта

def reconcile(camera_counts: dict, system_counts: dict,
               tolerance_percent: float = 5.0) -> list[dict]:
    """Нахождение расхождений между физическим и системным учётом"""
    discrepancies = []

    all_skus = set(camera_counts) | set(system_counts)
    for sku in all_skus:
        camera_qty = camera_counts.get(sku, 0)
        system_qty = system_counts.get(sku, 0)

        if system_qty > 0:
            diff_pct = abs(camera_qty - system_qty) / system_qty * 100
        else:
            diff_pct = 100 if camera_qty > 0 else 0

        if diff_pct > tolerance_percent:
            discrepancies.append({
                'sku': sku,
                'camera': camera_qty,
                'system': system_qty,
                'diff_percent': round(diff_pct, 1),
                'severity': 'high' if diff_pct > 20 else 'medium'
            })

    return sorted(discrepancies, key=lambda x: x['diff_percent'], reverse=True)