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

AI система розпізнавання товарів на полицях магазину

Розпізнавання товарів на полиці вирішує завдання ширше планограми: інвентаризація в реальному часі, оцінка Out-of-Stock (порожні полиці), аналіз цін конкурентів, моніторинг промо-виставок, підтримка покупця (що це за товар?). Особливість завдання — Fine-Grained Visual Recognition: розрізнити 10,000+ SKU по упаковці, де товари одного бренду візуально подібні.

Fine-Grained SKU розпізнавання

import torch
import torch.nn as nn
import timm
import numpy as np
import faiss
from pathlib import Path

class SKURecognitionEngine:
    """
    Підхід: навчання embedding space (metric learning),
    а не класифікатор на 10,000 класів.
    Перевага: додавання нових SKU без переосмислення.
    """
    def __init__(self, embedding_dim: int = 512,
                  device: str = 'cuda'):
        self.device = device
        self.embedding_dim = embedding_dim

        # EfficientNet-B4: гарний баланс точність/швидкість для FGVR
        self.encoder = timm.create_model(
            'efficientnet_b4',
            pretrained=True,
            num_classes=0  # видаляємо класифікатор
        )
        # Projection head для embedding
        self.projector = nn.Sequential(
            nn.Linear(self.encoder.num_features, 1024),
            nn.BatchNorm1d(1024),
            nn.GELU(),
            nn.Linear(1024, embedding_dim),
            nn.BatchNorm1d(embedding_dim)
        )

        self.encoder.to(device)
        self.projector.to(device)

        # FAISS index
        self.index = faiss.IndexFlatIP(embedding_dim)
        self.sku_registry = []  # список SKU метаданих

    @torch.no_grad()
    def get_embedding(self, image: np.ndarray) -> np.ndarray:
        from torchvision import transforms
        from PIL import Image

        transform = transforms.Compose([
            transforms.Resize((380, 380)),
            transforms.CenterCrop(350),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406],
                                  [0.229, 0.224, 0.225])
        ])
        pil_img = Image.fromarray(image[:, :, ::-1] if image.shape[2] == 3 else image)
        tensor = transform(pil_img).unsqueeze(0).to(self.device)

        self.encoder.eval()
        self.projector.eval()

        features = self.encoder(tensor)
        embedding = self.projector(features).cpu().numpy()
        faiss.normalize_L2(embedding)
        return embedding

    def register_sku(self, images: list[np.ndarray],
                      sku_id: str,
                      sku_metadata: dict) -> None:
        """
        Реєстрація нового SKU: кілька фото → середнє embedding.
        Не потребує переосмислення моделі.
        """
        embeddings = np.array([self.get_embedding(img)[0] for img in images])
        # Середнє нормалізоване embedding
        mean_emb = embeddings.mean(axis=0, keepdims=True)
        faiss.normalize_L2(mean_emb)

        self.index.add(mean_emb)
        self.sku_registry.append({
            'sku_id': sku_id,
            **sku_metadata
        })

    def recognize(self, crop_image: np.ndarray,
                   top_k: int = 3,
                   min_similarity: float = 0.7) -> list[dict]:
        """
        Розпізнавання одного культури товару.
        Повертає top_k кандидатів з впевненістю.
        """
        if self.index.ntotal == 0:
            return []

        query = self.get_embedding(crop_image)
        k = min(top_k, self.index.ntotal)
        similarities, indices = self.index.search(query, k)

        results = []
        for sim, idx in zip(similarities[0], indices[0]):
            if float(sim) >= min_similarity:
                results.append({
                    **self.sku_registry[idx],
                    'similarity': float(sim),
                    'recognized': True
                })

        if not results:
            results.append({'recognized': False, 'similarity': 0})

        return results

Виявлення Out-of-Stock та порожніх полиць

class OutOfStockDetector:
    """
    Виявлення порожніх просторів на полиці.
    Порожня полиця = немає товарів у даній зоні полочного простору.
    """
    def __init__(self, detector: YOLO):
        self.detector = detector

    def detect_oos(self, shelf_image: np.ndarray,
                    shelf_zones: list[dict] = None) -> dict:
        """
        shelf_zones: опціональні зони полиці (якщо відома структура)
        Без зон — аналіз щільності виявлень по горизонталі.
        """
        # Виявлення всіх видимих товарів
        results = self.detector(shelf_image, conf=0.35)
        boxes = []
        if results[0].boxes:
            boxes = results[0].boxes.xyxy.cpu().numpy()

        h, w = shelf_image.shape[:2]

        if shelf_zones:
            return self._check_zones(boxes, shelf_zones, w, h)
        else:
            return self._density_analysis(boxes, w, h)

    def _density_analysis(self, boxes: np.ndarray,
                           img_width: int,
                           img_height: int,
                           n_columns: int = 10) -> dict:
        """
        Розбиває полицю на колонки, перевіряє покриття.
        Колонка без товарів = потенційний OOS.
        """
        column_width = img_width / n_columns
        column_coverage = [False] * n_columns

        for box in boxes:
            x1, y1, x2, y2 = box
            center_x = (x1 + x2) / 2
            col = int(center_x / column_width)
            if 0 <= col < n_columns:
                column_coverage[col] = True

        empty_columns = [i for i, covered in enumerate(column_coverage) if not covered]
        oos_percentage = len(empty_columns) / n_columns * 100

        return {
            'out_of_stock': len(empty_columns) > 0,
            'oos_percentage': oos_percentage,
            'empty_positions': empty_columns,
            'total_products_detected': len(boxes),
            'severity': 'critical' if oos_percentage > 30 else
                       'warning' if oos_percentage > 10 else 'ok'
        }

Автоматичний екстрактор ціннику

class PriceTagExtractor:
    """
    Виявлення та OCR цінників на полиці.
    Застосування: аудит цін, відповідність промо-акціям.
    """
    def __init__(self, tag_detector: YOLO, ocr_engine):
        self.detector = tag_detector
        self.ocr = ocr_engine

    def extract_prices(self, shelf_image: np.ndarray) -> list[dict]:
        results = self.detector(shelf_image, conf=0.5)
        price_data = []

        for box in results[0].boxes:
            x1, y1, x2, y2 = map(int, box.xyxy[0])
            tag_crop = shelf_image[y1:y2, x1:x2]

            # OCR ціннику
            ocr_result = self.ocr.ocr(tag_crop)
            price = self._parse_price(ocr_result)

            price_data.append({
                'bbox': [x1, y1, x2, y2],
                'raw_text': ocr_result,
                'price': price,
                'position_x': (x1 + x2) / 2
            })

        return sorted(price_data, key=lambda p: p['position_x'])

    def _parse_price(self, ocr_result) -> float | None:
        import re
        if not ocr_result or not ocr_result[0]:
            return None
        text = ' '.join(line[1][0] for line in ocr_result[0])
        prices = re.findall(r'\d+[.,]\d{2}', text)
        return float(prices[0].replace(',', '.')) if prices else None