File size: 4,371 Bytes
8387086
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"""
Layout Agent
Analyses spatial distribution of detected objects and suggests
layout optimisations to reduce travel time and improve throughput.
"""

from loguru import logger
from dataclasses import dataclass


@dataclass
class LayoutSuggestion:
    category: str
    description: str
    priority: str  # "high", "medium", "low"
    estimated_saving_pct: float  # % improvement estimate


class LayoutAgent:
    def __init__(self):
        self.rules = self._load_rules()
        logger.info("LayoutAgent ready")

    def _load_rules(self) -> list:
        """
        Rule-based layout heuristics.
        Can be replaced by an ML model or LLM reasoning later.
        """
        return [
            {
                "trigger": "forklift_near_worker",
                "check": lambda d: self._forklifts_near_workers(d),
                "suggestion": LayoutSuggestion(
                    category="Safety Zone",
                    description="Forklifts detected in worker zones. Recommend separating pedestrian lanes.",
                    priority="high",
                    estimated_saving_pct=15.0,
                ),
            },
            {
                "trigger": "pallet_blocking",
                "check": lambda d: self._pallets_in_pathways(d),
                "suggestion": LayoutSuggestion(
                    category="Pathway Clearance",
                    description="Pallets detected in likely pathway zones. Move to designated storage.",
                    priority="high",
                    estimated_saving_pct=10.0,
                ),
            },
            {
                "trigger": "cluster_density",
                "check": lambda d: self._high_density_cluster(d),
                "suggestion": LayoutSuggestion(
                    category="Zone Density",
                    description="High object density detected in one zone. Redistribute to balance load.",
                    priority="medium",
                    estimated_saving_pct=8.0,
                ),
            },
        ]

    def suggest(self, detections: list) -> list[LayoutSuggestion]:
        """
        Generate layout suggestions based on detections.

        Args:
            detections: List of Detection objects from VisionAgent

        Returns:
            List of LayoutSuggestion objects
        """
        suggestions = []
        for rule in self.rules:
            if rule["check"](detections):
                suggestions.append(rule["suggestion"])
                logger.debug(f"Layout rule triggered: {rule['trigger']}")
        return suggestions

    # --- Heuristic checks ---

    def _forklifts_near_workers(self, detections: list) -> bool:
        """Check if forklifts and workers share overlapping bounding box zones."""
        forklifts = [d for d in detections if d.label == "forklift"]
        workers = [d for d in detections if d.label == "worker"]
        for f in forklifts:
            for w in workers:
                if self._overlap_or_close(f.bbox, w.bbox, threshold=100):
                    return True
        return False

    def _pallets_in_pathways(self, detections: list) -> bool:
        """Simple check: pallets detected near centre of image (likely pathway)."""
        pallets = [d for d in detections if d.label == "pallet"]
        for p in pallets:
            x1, y1, x2, y2 = p.bbox
            cx = (x1 + x2) / 2
            if 300 < cx < 700:  # rough centre-zone threshold
                return True
        return False

    def _high_density_cluster(self, detections: list) -> bool:
        """Check if more than 5 objects are packed in a small region."""
        if len(detections) < 5:
            return False
        bboxes = [d.bbox for d in detections]
        xs = [(b[0] + b[2]) / 2 for b in bboxes]
        ys = [(b[1] + b[3]) / 2 for b in bboxes]
        x_range = max(xs) - min(xs)
        y_range = max(ys) - min(ys)
        return x_range < 300 and y_range < 300

    def _overlap_or_close(self, bbox_a: list, bbox_b: list, threshold: float = 50) -> bool:
        """Check if two bounding boxes are within a pixel threshold."""
        ax1, ay1, ax2, ay2 = bbox_a
        bx1, by1, bx2, by2 = bbox_b
        horiz = abs((ax1 + ax2) / 2 - (bx1 + bx2) / 2)
        vert = abs((ay1 + ay2) / 2 - (by1 + by2) / 2)
        return horiz < threshold and vert < threshold