---
name: "material-tracking-iot"
description: "IoT-based material tracking for construction sites. Monitor material delivery, storage conditions, usage, and inventory with sensors, RFID, GPS, and real-time dashboards."
homepage: "https://datadrivenconstruction.io"
metadata: {"openclaw": {"emoji": "🚀", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}
---
# 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():
                    material.status = MaterialStatus.IN_USE

    def _process_gps_reading(self, reading: SensorReading, metadata: Dict):
        """Process GPS location update"""
        vehicle_id = metadata.get('vehicle_id') if metadata else None
        material_ids = metadata.get('material_ids', []) if metadata else []

        lat, lon = reading.value, metadata.get('longitude', 0) if metadata else 0

        for material_id in material_ids:
            material = self.materials.get(material_id)
            if material:
                material.location = f"GPS: {lat:.6f}, {lon:.6f}"
                material.status = MaterialStatus.IN_TRANSIT
                material.last_updated = reading.timestamp

    def _process_environment_reading(self, reading: SensorReading):
        """Process temperature/humidity reading"""
        location = reading.location

        if location not in self.storage_conditions:
            self.storage_conditions[location] = StorageCondition(
                location=location,
                timestamp=reading.timestamp
            )

        condition = self.storage_conditions[location]
        condition.timestamp = reading.timestamp

        if reading.reading_type == 'temperature':
            condition.temperature = reading.value
        elif reading.reading_type == 'humidity':
            condition.humidity = reading.value

        # Check conditions for materials in this location
        self._check_storage_conditions(location)

    def _check_storage_conditions(self, location: str):
        """Check if storage conditions are acceptable for materials"""
        condition = self.storage_conditions.get(location)
        if not condition:
            return

        condition.alerts = []
        condition.is_acceptable = True

        # Find materials in this location
        materials_here = [m for m in self.materials.values() if m.location == location]

        for material in materials_here:
            # Determine material type from name (simplified)
            material_type = None
            for mtype in self.material_requirements.keys():
                if mtype in material.name.lower():
                    material_type = mtype
                    break

            if not material_type:
                continue

            reqs = self.material_requirements[material_type]

            if condition.temperature is not None:
                if condition.temperature > reqs.get('max_temp', 100):
                    alert = f"Temperature too high for {material.name}: {condition.temperature}°C > {reqs['max_temp']}°C"
                    condition.alerts.append(alert)
                    condition.is_acceptable = False
                    self._create_alert('temperature_high', location, material.material_id, alert)

                if condition.temperature < reqs.get('min_temp', -100):
                    alert = f"Temperature too low for {material.name}: {condition.temperature}°C < {reqs['min_temp']}°C"
                    condition.alerts.append(alert)
                    condition.is_acceptable = False
                    self._create_alert('temperature_low', location, material.material_id, alert)

            if condition.humidity is not None:
                if condition.humidity > reqs.get('max_humidity', 100):
                    alert = f"Humidity too high for {material.name}: {condition.humidity}% > {reqs['max_humidity']}%"
                    condition.alerts.append(alert)
                    condition.is_acceptable = False
                    self._create_alert('humidity_high', location, material.material_id, alert)

    def _process_weight_reading(self, reading: SensorReading, metadata: Dict):
        """Process weight sensor reading for usage tracking"""
        material_id = metadata.get('material_id') if metadata else None
        if not material_id:
            return

        material = self.materials.get(material_id)
        if not material:
            return

        previous_weight = metadata.get('previous_weight', material.quantity)
        current_weight = reading.value

        if current_weight < previous_weight:
            # Material was used
            used_quantity = previous_weight - current_weight

            movement = MaterialMovement(
                movement_id=f"MOV-{uuid.uuid4().hex[:8]}",
                material_id=material_id,
                from_location=reading.location,
                to_location="installed",
                quantity=used_quantity,
                timestamp=reading.timestamp,
                recorded_by=reading.sensor_id,
                movement_type='usage'
            )
            self.movements.append(movement)

            material.quantity = current_weight
            material.last_updated = reading.timestamp

            # Check for low stock
            if material.quantity < metadata.get('reorder_level', 0):
                self._create_alert(
                    'low_stock',
                    reading.location,
                    material_id,
                    f"Low stock: {material.name} at {material.quantity} {material.unit}"
                )

    def _create_alert(self, alert_type: str, location: str,
                      material_id: str, message: str):
        """Create system alert"""
        self.alerts.append({
            'alert_id': f"ALT-{uuid.uuid4().hex[:8]}",
            'type': alert_type,
            'location': location,
            'material_id': material_id,
            'message': message,
            'timestamp': datetime.now().isoformat(),
            'acknowledged': False
        })

    def record_delivery(self, material_id: str, quantity: float,
                        location: str) -> MaterialMovement:
        """Record material delivery"""
        material = self.materials.get(material_id)
        if not material:
            raise ValueError(f"Unknown material: {material_id}")

        movement = MaterialMovement(
            movement_id=f"MOV-{uuid.uuid4().hex[:8]}",
            material_id=material_id,
            from_location="supplier",
            to_location=location,
            quantity=quantity,
            timestamp=datetime.now(),
            recorded_by="manual_entry",
            movement_type='delivery'
        )

        self.movements.append(movement)

        material.quantity += quantity
        material.location = location
        material.status = MaterialStatus.DELIVERED
        material.last_updated = datetime.now()

        return movement

    def get_material_history(self, material_id: str) -> List[Dict]:
        """Get movement history for a material"""
        history = []

        for movement in self.movements:
            if movement.material_id == material_id:
                history.append({
                    'movement_id': movement.movement_id,
                    'type': movement.movement_type,
                    'from': movement.from_location,
                    'to': movement.to_location,
                    'quantity': movement.quantity,
                    'timestamp': movement.timestamp.isoformat()
                })

        return sorted(history, key=lambda x: x['timestamp'])

    def get_inventory_report(self) -> Dict:
        """Generate inventory report"""
        report = {
            'generated_at': datetime.now().isoformat(),
            'total_materials': len(self.materials),
            'by_status': {},
            'by_location': {},
            'alerts': [a for a in self.alerts if not a['acknowledged']],
            'materials': []
        }

        for material in self.materials.values():
            status = material.status.value
            report['by_status'][status] = report['by_status'].get(status, 0) + 1

            if material.location:
                loc = material.location
                if loc not in report['by_location']:
                    report['by_location'][loc] = []
                report['by_location'][loc].append({
                    'id': material.material_id,
                    'name': material.name,
                    'quantity': material.quantity,
                    'unit': material.unit
                })

            report['materials'].append({
                'id': material.material_id,
                'name': material.name,
                'quantity': material.quantity,
                'unit': material.unit,
                'status': status,
                'location': material.location,
                'rfid': material.rfid_tag,
                'last_updated': material.last_updated.isoformat()
            })

        return report
```

### GPS Fleet Tracking

```python
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
import math

@dataclass
class GPSPosition:
    latitude: float
    longitude: float
    timestamp: datetime
    speed: float = 0  # km/h
    heading: float = 0  # degrees

@dataclass
class DeliveryVehicle:
    vehicle_id: str
    plate_number: str
    driver: str
    current_position: Optional[GPSPosition] = None
    material_ids: List[str] = None
    destination: Optional[Tuple[float, float]] = None
    estimated_arrival: Optional[datetime] = None

class FleetTracker:
    """GPS-based fleet tracking for material deliveries"""

    def __init__(self):
        self.vehicles: Dict[str, DeliveryVehicle] = {}
        self.position_history: Dict[str, List[GPSPosition]] = {}
        self.geofences: Dict[str, Dict] = {}

    def register_vehicle(self, plate_number: str, driver: str) -> DeliveryVehicle:
        """Register delivery vehicle"""
        vehicle_id = f"VEH-{plate_number.replace(' ', '').upper()}"

        vehicle = DeliveryVehicle(
            vehicle_id=vehicle_id,
            plate_number=plate_number,
            driver=driver,
            material_ids=[]
        )

        self.vehicles[vehicle_id] = vehicle
        self.position_history[vehicle_id] = []

        return vehicle

    def update_position(self, vehicle_id: str, lat: float, lon: float,
                       speed: float = 0, heading: float = 0) -> GPSPosition:
        """Update vehicle position"""
        vehicle = self.vehicles.get(vehicle_id)
        if not vehicle:
            raise ValueError(f"Unknown vehicle: {vehicle_id}")

        position = GPSPosition(
            latitude=lat,
            longitude=lon,
            timestamp=datetime.now(),
            speed=speed,
            heading=heading
        )

        vehicle.current_position = position
        self.position_history[vehicle_id].append(position)

        # Update ETA if destination set
        if vehicle.destination:
            vehicle.estimated_arrival = self._calculate_eta(vehicle)

        # Check geofences
        self._check_geofences(vehicle_id, position)

        return position

    def set_destination(self, vehicle_id: str, lat: float, lon: float):
        """Set delivery destination"""
        vehicle = self.vehicles.get(vehicle_id)
        if vehicle:
            vehicle.destination = (lat, lon)
            if vehicle.current_position:
                vehicle.estimated_arrival = self._calculate_eta(vehicle)

    def assign_materials(self, vehicle_id: str, material_ids: List[str]):
        """Assign materials to vehicle for delivery"""
        vehicle = self.vehicles.get(vehicle_id)
        if vehicle:
            vehicle.material_ids = material_ids

    def add_geofence(self, name: str, center_lat: float, center_lon: float,
                    radius_meters: float, fence_type: str = 'de