material-tracking-iot

IoT による資材追跡

概要

このスキルは、建設プロジェクト向けに IoT ベースの資材追跡システムを実装します。センサー、RFID タグ、GPS、接続デバイスを使用して、調達から設置まで資材を追跡します。

追跡機能：

• 配送追跡（GPS）
• 在庫管理（RFID）
• 保管環境（温度・湿度センサー）
• 使用状況の監視（重量センサー、カウンター）
• 廃棄物の追跡

クイックスタート

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

包括的な IoT 追跡システム

資材追跡エンジン

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 車両追跡

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 統合

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
        }
    }
}

ダッシュボードデータ生成ツール

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

クイックリファレンス

センサーの種類	用途	データ形式	更新頻度
RFID	資材の識別	タグ ID、信号強度	スキャン時
GPS	配送追跡	緯度／経度、速度、進行方向	10～60 秒
温度	保管状況の監視	摂氏	5～15 分
湿度	保管状況の監視	パーセンテージ	5～15 分
重量	使用状況の追跡	キログラム	変化時

MQTT トピックの構成

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}

リソース

• MQTT プロトコル：https://mqtt.org
• IoT プラットフォーム：AWS IoT、Azure IoT Hub、ThingsBoard
• DDC ウェブサイト：https://datadrivenconstruction.io

次のステップ

• IoT イベントの自動化については、n8n-workflow-automation を参照してください
• 追跡ダッシュボードについては、data-visualization を参照してください
• 資材数量の統合については、qto-report を参照してください