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():