IoT Industry Applications — Smart Home, Healthcare & Manufacturing

# configuration.yaml
# Automate lights based on motion + time of day
automation:
  - alias: "Kitchen Lights on Motion"
    trigger:
      - platform: state
        entity_id: binary_sensor.kitchen_motion
        to: 'on'
    condition:
      - condition: sun
        after: sunset
      - condition: template
        value_template: "{{ states('sensor.kitchen_illuminance') | int < 100 }}"
    action:
      - service: light.turn_on
        target:
          entity_id: light.kitchen_lights
        data:
          brightness_pct: 80
          color_temp: 400
  
  - alias: "Leave Home — Turn Everything Off"
    trigger:
      - platform: state
        entity_id: binary_sensor.front_door
        to: 'on'
      - platform: state
        entity_id: person.john_doe
        to: 'away'
    condition:
      - condition: state
        entity_id: person.john_doe
        state: 'away'
    action:
      - service: light.turn_off
        data: {}
        target:
          area_id: all
      - service: climate.turn_off
        target:
          entity_id: climate.thermostat

# Python Matter Server integration
from matter_server.client import MatterClient
import asyncio

async def control_matter_device():
    client = MatterClient("ws://localhost:5580/ws")
    await client.connect()
    
    # Discover devices
    nodes = await client.get_nodes()
    for node in nodes:
        print(f"Node: {node.node_id}, Vendor: {node.vendor_name}")
        
        # Control a light
        if "Light" in node.attribute_tree:
            await client.send_command(
                node.node_id,
                "onoff",
                "on",
                True
            )
    
    await client.disconnect()

asyncio.run(control_matter_device())

import asyncio
import json
from bleak import BleakClient
import aiohttp

# Wearable device (BLE heart rate monitor)
HR_CHARACTERISTIC = "00002a37-0000-1000-8000-00805f9b34fb"

class WearableMonitor:
    def __init__(self, device_address, patient_id):
        self.address = device_address
        self.patient_id = patient_id
        self.alert_thresholds = {
            'hr_max': 120,
            'hr_min': 40,
            'spo2_min': 90
        }
    
    async def monitor(self):
        async with BleakClient(self.address) as client:
            print(f"Connected to {self.address}")
            
            def notification_handler(sender, data):
                # Parse BLE heart rate measurement
                hr_value = data[1]  # byte 1 = HR value (uint8)
                spo2_value = data[2] if len(data) > 2 else 98
                
                reading = {
                    'patient_id': self.patient_id,
                    'heart_rate': hr_value,
                    'spo2': spo2_value,
                    'timestamp': time.time()
                }
                
                # Check thresholds
                if hr_value > self.alert_thresholds['hr_max']:
                    self.send_alert('CRITICAL', f'Tachycardia: {hr_value} bpm')
                elif hr_value < self.alert_thresholds['hr_min']:
                    self.send_alert('CRITICAL', f'Bradycardia: {hr_value} bpm')
                
                # Send to cloud
                asyncio.create_task(self.send_to_cloud(reading))
            
            await client.start_notify(HR_CHARACTERISTIC, notification_handler)
            await asyncio.Event().wait()  # Run indefinitely
    
    def send_alert(self, severity, message):
        print(f"[{severity}] {self.patient_id}: {message}")
        # POST to hospital alert system
    
    async def send_to_cloud(self, reading):
        async with aiohttp.ClientSession() as session:
            await session.post(
                'https://api.hospital.com/vitals',
                json=reading,
                headers={'Authorization': 'Bearer TOKEN'}
            )

import numpy as np
from scipy import fft, signal
import json

class PredictiveMaintenance:
    """
    Predict machine failure using vibration analysis.
    Monitors bearing wear, imbalance, and misalignment.
    """
    
    def __init__(self, machine_id):
        self.machine_id = machine_id
        self.vibration_buffer = []
        self.baseline_fft = None
        
    def analyze_vibration(self, time_domain_samples, sample_rate=1000):
        """
        Analyze vibration FFT for fault frequencies.
        
        Bearing fault frequencies:
        - BPFI (Ball Pass Frequency, Inner): ~5x RPM
        - BPFO (Ball Pass Frequency, Outer): ~3x RPM  
        - BSF (Ball Spin Frequency): ~2x RPM
        """
        # FFT analysis
        n = len(time_domain_samples)
        freqs = np.fft.rfftfreq(n, d=1/sample_rate)
        fft_values = np.abs(np.fft.rfft(time_domain_samples - 
                                         np.mean(time_domain_samples)))
        
        # Find dominant frequencies
        peak_indices = signal.find_peaks(fft_values, height=np.std(fft_values)*3)[0]
        dominant_freqs = freqs[peak_indices]
        
        # Check for bearing fault frequencies
        rpm = 1800  # Motor RPM
        fault_freqs = {
            'BPFI': rpm / 60 * 5.43,
            'BPFO': rpm / 60 * 3.21,
            'BSF': rpm / 60 * 2.17
        }
        
        findings = []
        for fault_name, expected_freq in fault_freqs.items():
            # Check if dominant frequency matches fault
            match = any(abs(f - expected_freq) < 2 for f in dominant_freqs)
            if match:
                findings.append({
                    'fault_type': fault_name,
                    'severity': 'high' if match else 'none',
                    'recommendation': f'Schedule maintenance for {fault_name}'
                })
        
        # Overall health score
        rms = np.sqrt(np.mean(time_domain_samples ** 2))
        crest_factor = np.max(np.abs(time_domain_samples)) / rms
        
        health_score = max(0, 100 - (rms * 10 + crest_factor * 5))
        
        return {
            'machine_id': self.machine_id,
            'health_score': min(health_score, 100),
            'rms_vibration': rms,
            'crest_factor': crest_factor,
            'faults': findings,
            'maintenance_required': len(findings) > 0 or health_score < 60
        }

# Simulate vibration data
np.random.seed(42)
normal_vibration = np.sin(2 * np.pi * 30 * np.linspace(0, 1, 1000)) * 0.5
fault_vibration = normal_vibration + np.sin(2 * np.pi * 162 * np.linspace(0, 1, 1000)) * 1.5

analyzer = PredictiveMaintenance("motor-pump-07")
result = analyzer.analyze_vibration(fault_vibration)
print(json.dumps(result, indent=2))

{
  "machine_id": "motor-pump-07",
  "health_score": 42,
  "rms_vibration": 0.85,
  "crest_factor": 3.2,
  "faults": [{"fault_type": "BPFI", "severity": "high", "recommendation": "Schedule maintenance for BPFI"}],
  "maintenance_required": true
}

import asyncio
import json
import random

class IoTSimulator:
    """Simulate devices across smart home, healthcare, and IIoT."""
    
    def __init__(self):
        self.devices = {
            'smart_home': [
                {'id': 'living_room_light', 'state': 'off'},
                {'id': 'thermostat', 'temperature': 22.0, 'target': 23.0},
                {'id': 'front_door_sensor', 'state': 'closed'},
            ],
            'healthcare': [
                {'id': 'patient_hr_monitor', 'heart_rate': 72},
                {'id': 'insulin_pump', 'battery': 85, 'reservoir': 60},
            ],
            'industrial': [
                {'id': 'conveyor_motor_01', 'vibration': 0.5, 'temp': 45},
                {'id': 'robot_arm_03', 'current_draw': 12.5, 'cycles': 15234},
            ]
        }
    
    async def simulate(self):
        while True:
            await asyncio.gather(
                self.simulate_smart_home(),
                self.simulate_healthcare(),
                self.simulate_industrial()
            )
            await asyncio.sleep(5)
    
    async def simulate_smart_home(self):
        self.devices['smart_home'][0]['state'] = random.choice(['on', 'off'])
        self.devices['smart_home'][1]['temperature'] += random.gauss(0, 0.2)
        print(f"Home: Light={self.devices['smart_home'][0]['state']}, "
              f"Temp={self.devices['smart_home'][1]['temperature']:.1f}°C")
    
    async def simulate_healthcare(self):
        hr = self.devices['healthcare'][0]
        hr['heart_rate'] += random.randint(-5, 5)
        hr['heart_rate'] = max(40, min(180, hr['heart_rate']))
        if hr['heart_rate'] > 100:
            print(f"[ALERT] Patient tachycardia: {hr['heart_rate']} bpm")
    
    async def simulate_industrial(self):
        motor = self.devices['industrial'][0]
        motor['vibration'] += random.gauss(0, 0.1)
        motor['vibration'] = max(0, motor['vibration'])
        if motor['vibration'] > 2.0:
            print(f"[ALERT] Motor vibration high: {motor['vibration']:.2f}")

sim = IoTSimulator()
asyncio.run(sim.simulate())