Розробка AI-системи автоматичної інвентаризації складу через камери дрони

Розроблення AI для інвентаризації складу (камери та дрони)

Ручна інвентаризація складу займає 2–5 днів простою з 1–3% помилок підрахунку. AI-системи на основі нерухомих камер або дронів скорочують час до кількох годин із точністю підрахунку 98–99%. Дрони особливо ефективні для високих стелажів (10+ метрів), недоступних ручним сканерам. Завдання: підрахунок піддонів, контроль заповнення комірок, ідентифікація артикулів за штрихкодами та зоровими ознаками.

Система підрахунку піддонів через камери

import numpy as np
import cv2
from ultralytics import YOLO
import sqlite3
from datetime import datetime

class WarehouseInventorySystem:
    def __init__(self, detector_path: str,
                  db_path: str = 'warehouse.db',
                  device: str = 'cuda'):
        self.detector = YOLO(detector_path)
        self.device = device
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self._init_db()

    def _init_db(self):
        self.db.execute('''
            CREATE TABLE IF NOT EXISTS inventory_snapshots (
                id INTEGER PRIMARY KEY,
                camera_id TEXT,
                zone_id TEXT,
                timestamp TIMESTAMP,
                pallet_count INTEGER,
                occupancy_pct REAL,
                detections TEXT
            )
        ''')
        self.db.commit()

    def count_pallets(self, image: np.ndarray,
                       camera_id: str,
                       zone_id: str,
                       total_slots: int = None) -> dict:
        """
        Підрахунок піддонів у зоні сховища.
        total_slots: відоме загальне число комірок для % заповнення.
        """
        results = self.detector(image, conf=0.45)
        detections = []

        for box in results[0].boxes:
            cls_name = self.detector.model.names[int(box.cls)]
            detections.append({
                'type': cls_name,
                'bbox': box.xyxy[0].tolist(),
                'conf': float(box.conf)
            })

        pallet_count = sum(
            1 for d in detections
            if d['type'] in ['pallet', 'loaded_pallet', 'empty_pallet']
        )
        loaded = sum(1 for d in detections if d['type'] == 'loaded_pallet')
        empty = sum(1 for d in detections if d['type'] == 'empty_pallet')

        occupancy = (pallet_count / total_slots * 100) if total_slots else None

        result = {
            'camera_id': camera_id,
            'zone_id': zone_id,
            'pallet_count': pallet_count,
            'loaded_pallets': loaded,
            'empty_pallets': empty,
            'occupancy_pct': occupancy,
            'timestamp': datetime.now().isoformat()
        }

        # Збереження в базу даних
        self.db.execute(
            'INSERT INTO inventory_snapshots VALUES (NULL, ?, ?, ?, ?, ?, ?)',
            (camera_id, zone_id, result['timestamp'],
             pallet_count, occupancy, str(detections))
        )
        self.db.commit()

        return result

    def get_zone_history(self, zone_id: str,
                          hours: int = 24) -> list[dict]:
        """Історія змін заповнення зони"""
        rows = self.db.execute('''
            SELECT timestamp, pallet_count, occupancy_pct
            FROM inventory_snapshots
            WHERE zone_id = ?
            AND timestamp > datetime('now', ?)
            ORDER BY timestamp DESC
        ''', (zone_id, f'-{hours} hours')).fetchall()
        return [
            {'timestamp': r[0], 'pallet_count': r[1], 'occupancy': r[2]}
            for r in rows
        ]

Дронова система для високих стелажів

class DroneInventorySystem:
    """
    Автономний об'їзд стелажів із камерою:
    1. Планування маршруту за відомою картою складу
    2. Знімання кожної комірки
    3. OCR штрихкодів та QR-кодів
    4. Перевірка з WMS (Warehouse Management System)
    """
    def __init__(self, barcode_reader, wms_client,
                  rack_detector: YOLO):
        self.barcode_reader = barcode_reader
        self.wms = wms_client
        self.rack_detector = rack_detector

    def inventory_rack(self, rack_images: list[dict]) -> dict:
        """
        rack_images: [{'image': np.ndarray, 'rack_id': str, 'shelf': int, 'slot': int}]
        Аналіз усіх фото конкретного стелажа.
        """
        rack_inventory = []
        discrepancies = []

        for item in rack_images:
            image = item['image']
            location = f"{item['rack_id']}-{item['shelf']}-{item['slot']}"

            # Детекція заповнення комірки
            results = self.rack_detector(image, conf=0.5)
            is_occupied = len(results[0].boxes) > 0

            # OCR штрихкода/QR-кода
            barcodes = self._read_barcodes(image)

            slot_data = {
                'location': location,
                'occupied': is_occupied,
                'barcodes': barcodes,
                'sku_ids': [self._barcode_to_sku(b) for b in barcodes]
            }

            rack_inventory.append(slot_data)

            # Перевірка з WMS
            if barcodes:
                for sku_id in slot_data['sku_ids']:
                    wms_expected = self.wms.get_expected_sku(location)
                    if sku_id != wms_expected:
                        discrepancies.append({
                            'location': location,
                            'found_sku': sku_id,
                            'expected_sku': wms_expected,
                            'discrepancy_type': 'wrong_sku' if wms_expected else 'unexpected_item'
                        })

        return {
            'total_slots_checked': len(rack_inventory),
            'occupied_slots': sum(1 for s in rack_inventory if s['occupied']),
            'empty_slots': sum(1 for s in rack_inventory if not s['occupied']),
            'barcodes_read': sum(len(s['barcodes']) for s in rack_inventory),
            'discrepancies': discrepancies,
            'discrepancy_rate': len(discrepancies) / len(rack_inventory) if rack_inventory else 0,
            'inventory': rack_inventory
        }

    def _read_barcodes(self, image: np.ndarray) -> list[str]:
        """Читання штрихкодів через pyzbar + QR через cv2"""
        from pyzbar import pyzbar
        decoded = pyzbar.decode(image)
        barcodes = [d.data.decode('utf-8') for d in decoded]

        # Спроба QR через OpenCV
        qr = cv2.QRCodeDetector()
        data, _, _ = qr.detectAndDecode(image)
        if data:
            barcodes.append(data)

        return barcodes

    def _barcode_to_sku(self, barcode: str) -> str:
        """Маппінг штрихкода → SKU через WMS"""
        return self.wms.lookup_barcode(barcode) or barcode

Генерування звіту інвентаризації

def generate_inventory_report(inventory_data: list[dict],
                                wms_data: dict,
                                output_format: str = 'json') -> dict:
    """Зведений звіт із розбіжностями для бухгалтерії/логістики"""
    total_items = sum(s.get('quantity', 1) for s in inventory_data)
    occupied = sum(1 for s in inventory_data if s.get('occupied'))

    all_discrepancies = []
    for slot in inventory_data:
        for disc in slot.get('discrepancies', []):
            all_discrepancies.append({**disc, 'location': slot['location']})

    return {
        'report_date': datetime.now().isoformat(),
        'summary': {
            'total_slots_scanned': len(inventory_data),
            'occupied': occupied,
            'empty': len(inventory_data) - occupied,
            'occupancy_rate_pct': occupied / len(inventory_data) * 100,
            'total_items': total_items,
            'discrepancy_count': len(all_discrepancies)
        },
        'discrepancies': all_discrepancies,
        'accuracy_rate': 1 - len(all_discrepancies) / max(len(inventory_data), 1)
    }