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

@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

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 = 'destination'):
    """Add geofence for location monitoring"""
    self.geofences[name] = {
        'center': (center_lat, center_lon),
        'radius': radius_meters,
        'type': fence_type,
        'vehicles_inside': []
    }

def _calculate_eta(self, vehicle: DeliveryVehicle) -> datetime:
    """Calculate estimated time of arrival"""
    if not vehicle.current_position or not vehicle.destination:
        return None

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

    # Use current speed or assume 40 km/h average
    speed = vehicle.current_position.speed if vehicle.current_position.speed > 0 else 40

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

def _haversine_distance(self, lat1: float, lon1: float,
                       lat2: float, lon2: float) -> float:
    """Calculate distance between two points in km"""
    R = 6371  # Earth radius in km

    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/2)**2 +
         math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(delta_lon/2)**2)
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))

    return R * c

def _check_geofences(self, vehicle_id: str, position: GPSPosition):
    """Check if vehicle entered/exited geofences"""
    for fence_name, fence in self.geofences.items():
        distance = self._haversine_distance(
            position.latitude,
            position.longitude,
            fence['center'][0],
            fence['center'][1]
        ) * 1000  # Convert to meters

        is_inside = distance <= fence['radius']
        was_inside = vehicle_id in fence['vehicles_inside']

        if is_inside and not was_inside:
            # Entered geofence
            fence['vehicles_inside'].append(vehicle_id)
            self._on_geofence_enter(vehicle_id, fence_name, fence['type'])

        elif not is_inside and was_inside:
            # Exited geofence
            fence['vehicles_inside'].remove(vehicle_id)
            self._on_geofence_exit(vehicle_id, fence_name, fence['type'])

def _on_geofence_enter(self, vehicle_id: str, fence_name: str, fence_type: str):
    """Handle geofence entry event"""
    print(f"Vehicle {vehicle_id} entered {fence_name} ({fence_type})")
    # Could trigger webhook, update status, send notification

def _on_geofence_exit(self, vehicle_id: str, fence_name: str, fence_type: str):
    """Handle geofence exit event"""
    print(f"Vehicle {vehicle_id} exited {fence_name} ({fence_type})")

def get_fleet_status(self) -> Dict:
    """Get current fleet status"""
    status = {
        'timestamp': datetime.now().isoformat(),
        'total_vehicles': len(self.vehicles),
        'vehicles': []
    }

    for vehicle in self.vehicles.values():
        v_status = {
            'id': vehicle.vehicle_id,
            'plate': vehicle.plate_number,
            'driver': vehicle.driver,
            'materials': vehicle.material_ids,
            'position': None,
            'eta': None
        }

        if vehicle.current_position:
            v_status['position'] = {
                'lat': vehicle.current_position.latitude,
                'lon': vehicle.current_position.longitude,
                'speed': vehicle.current_position.speed,
                'updated': vehicle.current_position.timestamp.isoformat()
            }

        if vehicle.estimated_arrival:
            v_status['eta'] = vehicle.estimated_arrival.isoformat()

        status['vehicles'].append(v_status)

    return 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 in self.topic_handlers.items():
            if self._topic_matches(topic, pattern):
                handler(topic, data)
                break

    except json.JSONDecodeError:
        print(f"Invalid JSON in message: {topic}")
    except Exception as e:
        print(f"Error processing message: {e}")

def _topic_matches(self, topic: str, pattern: str) -> bool:
    """Check if topic matches pattern (+ is wildcard)"""
    topic_parts = topic.split('/')
    pattern_parts = pattern.split('/')

    if len(topic_parts) != len(pattern_parts):
        return False

    for t, p in zip(topic_parts, pattern_parts):
        if p != '+' and t != p:
            return False

    return True

def _handle_rfid(self, topic: str, data: Dict):
    """Handle RFID scan message"""
    sensor_id = topic.split('/')[-1]

    self.tracking.process_sensor_reading(
        sensor_id=sensor_id,
        value=1,  # Scan detected
        unit='scan',
        metadata={
            'rfid_tag': data.get('tag_id'),
            'signal_strength': data.get('rssi')
        }
    )

def _handle_gps(self, topic: str, data: Dict):
    """Handle GPS position message"""
    sensor_id = topic.split('/')[-1]

    self.tracking.process_sensor_reading(
        sensor_id=sensor_id,
        value=data.get('latitude', 0),
        unit='degrees',
        metadata={
            'longitude': data.get('longitude'),
            'speed': data.get('speed'),
            'vehicle_id': data.get('vehicle_id'),
            'material_ids': data.get('materials', [])
        }
    )

def _handle_temperature(self, topic: str, data: Dict):
    """Handle temperature sensor message"""
    sensor_id = topic.split('/')[-1]

    self.tracking.process_sensor_reading(
        sensor_id=sensor_id,
        value=data.get('value', 0),
        unit='celsius',
        metadata={}
    )

def _handle_humidity(self, topic: str, data: Dict):
    """Handle humidity sensor message"""
    sensor_id = topic.split('/')[-1]

    self.tracking.process_sensor_reading(
        sensor_id=sensor_id,
        value=data.get('value', 0),
        unit='percent',
        metadata={}
    )

def _handle_weight(self, topic: str, data: Dict):
    """Handle weight sensor message"""
    sensor_id = topic.split('/')[-1]

    self.tracking.process_sensor_reading(
        sensor_id=sensor_id,
        value=data.get('value', 0),
        unit='kg',
        metadata={
            'material_id': data.get('material_id'),
            'previous_weight': data.get('previous'),
            'reorder_level': data.get('reorder_level')
        }
    )

Example MQTT message formats

EXAMPLE_MESSAGES = { 'rfid_scan': { 'topic': 'sensors/rfid/SNS-RFID-001', 'payload': { 'tag_id': 'RFID-MAT-001', 'rssi': -45, 'timestamp': '2024-01-15T10:30:00Z' } }, 'gps_update': { 'topic': 'sensors/gps/VEH-ABC123', 'payload': { 'latitude': 55.7558, 'longitude': 37.6173, 'speed': 45, 'heading': 90, 'vehicle_id': 'VEH-ABC123', 'materials': ['MAT-001', 'MAT-002'] } }, 'temperature': { 'topic': 'sensors/temperature/SNS-TEMP-001', 'payload': { 'value': 28.5, 'battery': 85 } } }

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['alerts']),
            'active_deliveries': len([v for v in self.fleet.vehicles.values()
                                     if v.material_ids])
        },
        'inventory': inventory,
        'fleet': fleet_status,
        'alerts': inventory['alerts'][:10],  # Top 10
        'usage_trend': daily_usage,
        'storage_conditions': {
            loc: {
                'temperature': cond.temperature,
                'humidity': cond.humidity,
                'ok': cond.is_acceptable
            }
            for loc, cond in self.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