material-tracking-iot

What this skill does

# Material Tracking with IoT

## Overview

This skill implements IoT-based material tracking systems for construction projects. Track materials from procurement to installation using sensors, RFID tags, GPS, and connected devices.

**Tracking Capabilities:**
- Delivery tracking (GPS)
- Inventory management (RFID)
- Storage conditions (temperature, humidity sensors)
- Usage monitoring (weight sensors, counters)
- Waste tracking

## Quick Start

```python
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Optional
from enum import Enum
import json

class MaterialStatus(Enum):
    ORDERED = "ordered"
    IN_TRANSIT = "in_transit"
    DELIVERED = "delivered"
    IN_STORAGE = "in_storage"
    IN_USE = "in_use"
    INSTALLED = "installed"
    WASTED = "wasted"

@dataclass
class MaterialItem:
    material_id: str
    name: str
    quantity: float
    unit: str
    rfid_tag: Optional[str] = None
    status: MaterialStatus = MaterialStatus.ORDERED
    location: Optional[str] = None
    last_updated: datetime = field(default_factory=datetime.now)

@dataclass
class SensorReading:
    sensor_id: str
    reading_type: str  # temperature, humidity, weight, gps
    value: float
    unit: str
    timestamp: datetime
    location: Optional[str] = None

# Quick inventory check
def check_inventory(materials: List[MaterialItem]) -> Dict:
    """Quick inventory status check"""
    inventory = {
        'total_items': len(materials),
        'by_status': {},
        'by_location': {}
    }

    for m in materials:
        status = m.status.value
        inventory['by_status'][status] = inventory['by_status'].get(status, 0) + 1

        if m.location:
            inventory['by_location'][m.location] = inventory['by_location'].get(m.location, 0) + 1

    return inventory

# Example
materials = [
    MaterialItem("MAT-001", "Rebar 12mm", 500, "kg", "RFID-001", MaterialStatus.IN_STORAGE, "Yard-A"),
    MaterialItem("MAT-002", "Concrete C30", 50, "m³", None, MaterialStatus.IN_TRANSIT),
    MaterialItem("MAT-003", "Steel Beam HEB200", 20, "pcs", "RFID-002", MaterialStatus.DELIVERED, "Yard-B")
]

print(check_inventory(materials))
```

## Comprehensive IoT Tracking System

### Material Tracking Engine

```python
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
from enum import Enum
import uuid
import json

class SensorType(Enum):
    RFID = "rfid"
    GPS = "gps"
    TEMPERATURE = "temperature"
    HUMIDITY = "humidity"
    WEIGHT = "weight"
    MOTION = "motion"
    CAMERA = "camera"

@dataclass
class Sensor:
    sensor_id: str
    sensor_type: SensorType
    location: str
    is_active: bool = True
    last_reading: Optional[datetime] = None
    battery_level: float = 100.0

@dataclass
class MaterialMovement:
    movement_id: str
    material_id: str
    from_location: Optional[str]
    to_location: str
    quantity: float
    timestamp: datetime
    recorded_by: str  # sensor_id or user_id
    movement_type: str  # delivery, transfer, usage, waste

@dataclass
class StorageCondition:
    location: str
    timestamp: datetime
    temperature: Optional[float] = None  # Celsius
    humidity: Optional[float] = None  # Percentage
    is_acceptable: bool = True
    alerts: List[str] = field(default_factory=list)

class MaterialTrackingSystem:
    """IoT-based material tracking system"""

    def __init__(self):
        self.materials: Dict[str, MaterialItem] = {}
        self.sensors: Dict[str, Sensor] = {}
        self.movements: List[MaterialMovement] = []
        self.readings: List[SensorReading] = []
        self.storage_conditions: Dict[str, StorageCondition] = {}
        self.alerts: List[Dict] = []

        # Material requirements (for condition monitoring)
        self.material_requirements = {
            'cement': {'max_humidity': 60, 'max_temp': 35, 'min_temp': 5},
            'steel': {'max_humidity': 70, 'max_temp': 50, 'min_temp': -20},
            'wood': {'max_humidity': 65, 'max_temp': 40, 'min_temp': 0},
            'paint': {'max_humidity': 80, 'max_temp': 30, 'min_temp': 5},
            'adhesive': {'max_humidity': 70, 'max_temp': 25, 'min_temp': 10}
        }

    def register_material(self, name: str, quantity: float, unit: str,
                         rfid_tag: Optional[str] = None,
                         material_type: str = 'general') -> MaterialItem:
        """Register new material in the system"""
        material_id = f"MAT-{uuid.uuid4().hex[:8].upper()}"

        material = MaterialItem(
            material_id=material_id,
            name=name,
            quantity=quantity,
            unit=unit,
            rfid_tag=rfid_tag,
            status=MaterialStatus.ORDERED
        )

        self.materials[material_id] = material
        return material

    def register_sensor(self, sensor_type: SensorType, location: str) -> Sensor:
        """Register new sensor"""
        sensor_id = f"SNS-{sensor_type.value.upper()}-{uuid.uuid4().hex[:6].upper()}"

        sensor = Sensor(
            sensor_id=sensor_id,
            sensor_type=sensor_type,
            location=location
        )

        self.sensors[sensor_id] = sensor
        return sensor

    def process_sensor_reading(self, sensor_id: str, value: float,
                               unit: str, metadata: Dict = None) -> SensorReading:
        """Process incoming sensor reading"""
        sensor = self.sensors.get(sensor_id)
        if not sensor:
            raise ValueError(f"Unknown sensor: {sensor_id}")

        reading = SensorReading(
            sensor_id=sensor_id,
            reading_type=sensor.sensor_type.value,
            value=value,
            unit=unit,
            timestamp=datetime.now(),
            location=sensor.location
        )

        self.readings.append(reading)
        sensor.last_reading = reading.timestamp

        # Process based on sensor type
        if sensor.sensor_type == SensorType.RFID:
            self._process_rfid_reading(reading, metadata)
        elif sensor.sensor_type == SensorType.GPS:
            self._process_gps_reading(reading, metadata)
        elif sensor.sensor_type in [SensorType.TEMPERATURE, SensorType.HUMIDITY]:
            self._process_environment_reading(reading)
        elif sensor.sensor_type == SensorType.WEIGHT:
            self._process_weight_reading(reading, metadata)

        return reading

    def _process_rfid_reading(self, reading: SensorReading, metadata: Dict):
        """Process RFID tag scan"""
        rfid_tag = metadata.get('rfid_tag') if metadata else None
        if not rfid_tag:
            return

        # Find material with this RFID
        material = None
        for m in self.materials.values():
            if m.rfid_tag == rfid_tag:
                material = m
                break

        if material:
            old_location = material.location
            new_location = reading.location

            # Record movement
            if old_location != new_location:
                movement = MaterialMovement(
                    movement_id=f"MOV-{uuid.uuid4().hex[:8]}",
                    material_id=material.material_id,
                    from_location=old_location,
                    to_location=new_location,
                    quantity=material.quantity,
                    timestamp=reading.timestamp,
                    recorded_by=reading.sensor_id,
                    movement_type='transfer'
                )
                self.movements.append(movement)

                material.location = new_location
                material.last_updated = reading.timestamp

                # Update status based on location
                if 'yard' in new_location.lower() or 'storage' in new_location.lower():
                    material.status = MaterialStatus.IN_STORAGE
                elif 'floor' in new_location.lower() or 'zone' in new_location.lower():