material-tracking-iot

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":"[Truncated]"}}}

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

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

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


 :
    location: 
    timestamp: datetime
    temperature: [] =   
    humidity: [] =   
    is_acceptable:  = 
    alerts: [] = field(default_factory=)

 :
    

     ():
        .materials: [, MaterialItem] = {}
        .sensors: [, Sensor] = {}
        .movements: [MaterialMovement] = []
        .readings: [SensorReading] = []
        .storage_conditions: [, StorageCondition] = {}
        .alerts: [] = []

        
        .material_requirements = {
            : {: , : , : },
            : {: , : , : -},
            : {: , : , : },
            : {: , : , : },
            : {: , : , : }
        }

     () -> MaterialItem:
        
        material_id = 

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

        .materials[material_id] = material
         material

     () -> Sensor:
        
        sensor_id = 

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

        .sensors[sensor_id] = sensor
         sensor

     () -> SensorReading:
        
        sensor = .sensors.get(sensor_id)
          sensor:
             ValueError()

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

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

        
         sensor.sensor_type == SensorType.RFID:
            ._process_rfid_reading(reading, metadata)
         sensor.sensor_type == SensorType.GPS:
            ._process_gps_reading(reading, metadata)
         sensor.sensor_type  [SensorType.TEMPERATURE, SensorType.HUMIDITY]:
            ._process_environment_reading(reading)
         sensor.sensor_type == SensorType.WEIGHT:
            ._process_weight_reading(reading, metadata)

         reading

     ():
        
        rfid_tag = metadata.get()  metadata  
          rfid_tag:
            

        
        material = 
         m  .materials.values():
             m.rfid_tag == rfid_tag:
                material = m
                

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

            
             old_location != new_location:
                movement = MaterialMovement(
                    movement_id=,
                    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=
                )
                .movements.append(movement)

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

                
                   new_location.lower()    new_location.lower():
                    material.status = MaterialStatus.IN_STORAGE
                   new_location.lower()    new_location.lower():
                    material.status = MaterialStatus.IN_USE

     ():
        
        vehicle_id = metadata.get()  metadata  
        material_ids = metadata.get(, [])  metadata  []

        lat, lon = reading.value, metadata.get(, )  metadata  

         material_id  material_ids:
            material = .materials.get(material_id)
             material:
                material.location = 
                material.status = MaterialStatus.IN_TRANSIT
                material.last_updated = reading.timestamp

     ():
        
        location = reading.location

         location   .storage_conditions:
            .storage_conditions[location] = StorageCondition(
                location=location,
                timestamp=reading.timestamp
            )

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

         reading.reading_type == :
            condition.temperature = reading.value
         reading.reading_type == :
            condition.humidity = reading.value

        
        ._check_storage_conditions(location)

     ():
        
        condition = .storage_conditions.get(location)
          condition:
            

        condition.alerts = []
        condition.is_acceptable = 

        
        materials_here = [m  m  .materials.values()  m.location == location]

         material  materials_here:
            
            material_type = 
             mtype  .material_requirements.keys():
                 mtype  material.name.lower():
                    material_type = mtype
                    

              material_type:
                

            reqs = .material_requirements[material_type]

             condition.temperature   :
                 condition.temperature > reqs.get(, ):
                    alert = 
                    condition.alerts.append(alert)
                    condition.is_acceptable = 
                    ._create_alert(, location, material.material_id, alert)

                 condition.temperature < reqs.get(, -):
                    alert = 
                    condition.alerts.append(alert)
                    condition.is_acceptable = 
                    ._create_alert(, location, material.material_id, alert)

             condition.humidity   :
                 condition.humidity > reqs.get(, ):
                    alert = 
                    condition.alerts.append(alert)
                    condition.is_acceptable = 
                    ._create_alert(, location, material.material_id, alert)

     ():
        
        material_id = metadata.get()  metadata  
          material_id:
            

        material = .materials.get(material_id)
          material:
            

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

         current_weight < previous_weight:
            
            used_quantity = previous_weight - current_weight

            movement = MaterialMovement(
                movement_id=,
                material_id=material_id,
                from_location=reading.location,
                to_location=,
                quantity=used_quantity,
                timestamp=reading.timestamp,
                recorded_by=reading.sensor_id,
                movement_type=
            )
            .movements.append(movement)

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

            
             material.quantity < metadata.get(, ):
                ._create_alert(
                    ,
                    reading.location,
                    material_id,
                    
                )

     ():
        
        .alerts.append({
            : ,
            : alert_type,
            : location,
            : material_id,
            : message,
            : datetime.now().isoformat(),
            : 
        })

     () -> MaterialMovement:
        
        material = .materials.get(material_id)
          material:
             ValueError()

        movement = MaterialMovement(
            movement_id=,
            material_id=material_id,
            from_location=,
            to_location=location,
            quantity=quantity,
            timestamp=datetime.now(),
            recorded_by=,
            movement_type=
        )

        .movements.append(movement)

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

         movement

     () -> []:
        
        history = []

         movement  .movements:
             movement.material_id == material_id:
                history.append({
                    : movement.movement_id,
                    : movement.movement_type,
                    : movement.from_location,
                    : movement.to_location,
                    : movement.quantity,
                    : movement.timestamp.isoformat()
                })

         (history, key= x: x[])

     () -> :
        
        report = {
            : datetime.now().isoformat(),
            : (.materials),
            : {},
            : {},
            : [a  a  .alerts   a[]],
            : []
        }

         material  .materials.values():
            status = material.status.value
            report[][status] = report[].get(status, ) + 

             material.location:
                loc = material.location
                 loc   report[]:
                    report[][loc] = []
                report[][loc].append({
                    : material.material_id,
                    : material.name,
                    : material.quantity,
                    : material.unit
                })

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

         report

GPS Fleet Tracking

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: [, [GPSPosition]] = {}
        .geofences: [, ] = {}

     () -> DeliveryVehicle:
        
        vehicle_id = 

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

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

         vehicle

     () -> GPSPosition:
        
        vehicle = .vehicles.get(vehicle_id)
          vehicle:
             ValueError()

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

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

        
         vehicle.destination:
            vehicle.estimated_arrival = ._calculate_eta(vehicle)

        
        ._check_geofences(vehicle_id, position)

         position

     ():
        
        vehicle = .vehicles.get(vehicle_id)
         vehicle:
            vehicle.destination = (lat, lon)
             vehicle.current_position:
                vehicle.estimated_arrival = ._calculate_eta(vehicle)

     ():
        
        vehicle = .vehicles.get(vehicle_id)
         vehicle:
            vehicle.material_ids = material_ids

     ():
        
        .geofences[name] = {
            : (center_lat, center_lon),
            : radius_meters,
            : fence_type,
            : []
        }

     () -> datetime:
        
          vehicle.current_position   vehicle.destination:
             

        distance = ._haversine_distance(
            vehicle.current_position.latitude,
            vehicle.current_position.longitude,
            vehicle.destination[],
            vehicle.destination[]
        )

        
        speed = vehicle.current_position.speed  vehicle.current_position.speed >   

        hours = distance / speed
         datetime.now() + timedelta(hours=hours)

     () -> :
        
        R =   

        lat1_rad = math.radians(lat1)
        lat2_rad = math.radians(lat2)
        delta_lat = math.radians(lat2 - lat1)
        delta_lon = math.radians(lon2 - lon1)

        a = (math.sin(delta_lat/)** +
             math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(delta_lon/)**)
        c =  * math.atan2(math.sqrt(a), math.sqrt(-a))

         R * c

     ():
        
         fence_name, fence  .geofences.items():
            distance = ._haversine_distance(
                position.latitude,
                position.longitude,
                fence[][],
                fence[][]
            ) *   

            is_inside = distance <= fence[]
            was_inside = vehicle_id  fence[]

             is_inside   was_inside:
                
                fence[].append(vehicle_id)
                ._on_geofence_enter(vehicle_id, fence_name, fence[])

              is_inside  was_inside:
                
                fence[].remove(vehicle_id)
                ._on_geofence_exit(vehicle_id, fence_name, fence[])

     ():
        
        ()
        

     ():
        
        ()

     () -> :
        
        status = {
            : datetime.now().isoformat(),
            : (.vehicles),
            : []
        }

         vehicle  .vehicles.values():
            v_status = {
                : vehicle.vehicle_id,
                : vehicle.plate_number,
                : vehicle.driver,
                : vehicle.material_ids,
                : ,
                : 
            }

             vehicle.current_position:
                v_status[] = {
                    : vehicle.current_position.latitude,
                    : vehicle.current_position.longitude,
                    : vehicle.current_position.speed,
                    : vehicle.current_position.timestamp.isoformat()
                }

             vehicle.estimated_arrival:
                v_status[] = vehicle.estimated_arrival.isoformat()

            status[].append(v_status)

         status

MQTT Integration for Sensors

import json
from datetime import datetime
from typing import Callable, Dict

class IoTMessageHandler:
    """Handle MQTT messages from IoT sensors"""

    def __init__(self, tracking_system: MaterialTrackingSystem):
        self.tracking = tracking_system
        self.topic_handlers: Dict[str, Callable] = {}
        self._setup_handlers()

    def _setup_handlers(self):
        """Setup topic handlers"""
        self.topic_handlers = {
            'sensors/rfid/+': self._handle_rfid,
            'sensors/gps/+': self._handle_gps,
            'sensors/temperature/+': self._handle_temperature,
            'sensors/humidity/+': self._handle_humidity,
            'sensors/weight/+': self._handle_weight
        }

    def process_message(self, topic: str, payload: bytes):
        """Process incoming MQTT message"""
        try:
            data = json.loads(payload.decode('utf-8'))

            # Find matching handler
            for pattern, handler  .topic_handlers.items():
                 ._topic_matches(topic, pattern):
                    handler(topic, data)
                    

         json.JSONDecodeError:
            ()
         Exception  e:
            ()

     () -> :
        
        topic_parts = topic.split()
        pattern_parts = pattern.split()

         (topic_parts) != (pattern_parts):
             

         t, p  (topic_parts, pattern_parts):
             p !=   t != p:
                 

         

     ():
        
        sensor_id = topic.split()[-]

        .tracking.process_sensor_reading(
            sensor_id=sensor_id,
            value=,  
            unit=,
            metadata={
                : data.get(),
                : data.get()
            }
        )

     ():
        
        sensor_id = topic.split()[-]

        .tracking.process_sensor_reading(
            sensor_id=sensor_id,
            value=data.get(, ),
            unit=,
            metadata={
                : data.get(),
                : data.get(),
                : data.get(),
                : data.get(, [])
            }
        )

     ():
        
        sensor_id = topic.split()[-]

        .tracking.process_sensor_reading(
            sensor_id=sensor_id,
            value=data.get(, ),
            unit=,
            metadata={}
        )

     ():
        
        sensor_id = topic.split()[-]

        .tracking.process_sensor_reading(
            sensor_id=sensor_id,
            value=data.get(, ),
            unit=,
            metadata={}
        )

     ():
        
        sensor_id = topic.split()[-]

        .tracking.process_sensor_reading(
            sensor_id=sensor_id,
            value=data.get(, ),
            unit=,
            metadata={
                : data.get(),
                : data.get(),
                : data.get()
            }
        )



EXAMPLE_MESSAGES = {
    : {
        : ,
        : {
            : ,
            : -,
            : 
        }
    },
    : {
        : ,
        : {
            : ,
            : ,
            : ,
            : ,
            : ,
            : [, ]
        }
    },
    : {
        : ,
        : {
            : ,
            : 
        }
    }
}

Dashboard Data Generator

class MaterialDashboard:
    """Generate data for material tracking dashboard"""

    def __init__(self, tracking: MaterialTrackingSystem,
                 fleet: FleetTracker):
        self.tracking = tracking
        self.fleet = fleet

    def get_dashboard_data(self) -> Dict:
        """Get comprehensive dashboard data"""
        inventory = self.tracking.get_inventory_report()
        fleet_status = self.fleet.get_fleet_status()

        # Calculate usage trends (last 7 days)
        usage_movements = [
            m for m in self.tracking.movements
            if m.movement_type == 'usage' and
            (datetime.now() - m.timestamp).days <= 7
        ]

        daily_usage = {}
        for m in usage_movements:
            day = m.timestamp.strftime('%Y-%m-%d')
            daily_usage[day] = daily_usage.get(day, 0) + m.quantity

        return {
            'summary': {
                'total_materials': inventory['total_materials'],
                'in_transit': inventory['by_status'].get('in_transit', 0),
                'in_storage': inventory['by_status'].get('in_storage', 0),
                'active_alerts': len(inventory[]),
                : ([v  v  .fleet.vehicles.values()
                                          v.material_ids])
            },
            : inventory,
            : fleet_status,
            : inventory[][:],  
            : daily_usage,
            : {
                loc: {
                    : cond.temperature,
                    : cond.humidity,
                    : cond.is_acceptable
                }
                 loc, cond  .tracking.storage_conditions.items()
            }
        }

Quick Reference

Sensor Type	Use Case	Data Format	Update Frequency
RFID	Material identification	Tag ID, signal strength	On scan
GPS	Delivery tracking	Lat/Lon, speed, heading	10-60 sec
Temperature	Storage monitoring	Celsius	5-15 min
Humidity	Storage monitoring	Percentage	5-15 min
Weight	Usage tracking	Kilograms	On change

MQTT Topics Structure

sensors/
├── rfid/{sensor_id}
├── gps/{vehicle_id}
├── temperature/{sensor_id}
├── humidity/{sensor_id}
├── weight/{sensor_id}
└── motion/{sensor_id}

alerts/
├── low_stock/{material_id}
├── temperature/{location}
├── delivery/{vehicle_id}
└── geofence/{fence_name}

Resources

• MQTT Protocol: https://mqtt.org
• IoT Platforms: AWS IoT, Azure IoT Hub, ThingsBoard
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See n8n-workflow-automation for IoT event automation
• See data-visualization for tracking dashboards
• See qto-report for material quantity integration