# IoT боты: управление умным домом и автоматизация устройств
Привет! Меня зовут Дмитрий, и я уже 4 года разрабатываю IoT-ботов для умных домов и промышленной автоматизации. За это время я создал более 60 IoT-ботов, подключил более 10,000 устройств и автоматизировал процессы в 200+ умных домах. В этой статье расскажу, как создать мощного IoT-бота, какие протоколы использовать, и как обеспечить безопасность подключенных устройств.
## Почему IoT-боты становятся необходимостью?
### Статистика IoT рынка
- **75 миллиардов** IoT устройств к 2025 году
- **$1.1 триллион** объем рынка IoT
- **67%** домовладельцев используют умные устройства
- **45%** экономии энергии при автоматизации
### Преимущества IoT-ботов
- **Централизованное управление** - все устройства в одном месте
- **Автоматизация сценариев** - умные алгоритмы
- **Голосовое управление** - удобство использования
- **Мониторинг и аналитика** - контроль состояния
## Архитектура IoT-бота
### 1. 🏠 Система управления умным домом
```python
import asyncio
import json
import aiohttp
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
import logging
class DeviceType(Enum):
LIGHT = "light"
THERMOSTAT = "thermostat"
CAMERA = "camera"
DOOR_LOCK = "door_lock"
SENSOR = "sensor"
SWITCH = "switch"
SPEAKER = "speaker"
SECURITY_SYSTEM = "security_system"
class DeviceStatus(Enum):
ONLINE = "online"
OFFLINE = "offline"
ERROR = "error"
MAINTENANCE = "maintenance"
class Protocol(Enum):
ZIGBEE = "zigbee"
Z_WAVE = "z_wave"
WIFI = "wifi"
BLUETOOTH = "bluetooth"
THREAD = "thread"
MATTER = "matter"
@dataclass
class IoTDevice:
id: str
name: str
type: DeviceType
protocol: Protocol
status: DeviceStatus
room: str
capabilities: List[str]
current_state: Dict[str, Any]
last_seen: datetime
ip_address: Optional[str] = None
mac_address: Optional[str] = None
@dataclass
class AutomationRule:
id: str
name: str
trigger_conditions: Dict[str, Any]
actions: List[Dict[str, Any]]
enabled: bool
created_at: datetime
last_triggered: Optional[datetime] = None
class IoTHomeBot:
def __init__(self, bot_token: str):
self.bot_token = bot_token
self.devices = {}
self.rooms = {}
self.automation_rules = {}
self.scenes = {}
self.schedules = {}
# Протоколы и интеграции
self.protocols = {
Protocol.ZIGBEE: ZigbeeProtocol(),
Protocol.Z_WAVE: ZWaveProtocol(),
Protocol.WIFI: WiFiProtocol(),
Protocol.BLUETOOTH: BluetoothProtocol(),
Protocol.MATTER: MatterProtocol()
}
# Системы безопасности
self.security_manager = SecurityManager()
self.energy_monitor = EnergyMonitor()
self.setup_handlers()
self.start_device_discovery()
def setup_handlers(self):
"""Настройка обработчиков команд"""
self.app.command('/start', self.handle_start)
self.app.command('/devices', self.handle_devices)
self.app.command('/rooms', self.handle_rooms)
self.app.command('/automation', self.handle_automation)
self.app.command('/security', self.handle_security)
self.app.command('/energy', self.handle_energy)
self.app.command('/scenes', self.handle_scenes)
# Обработчики сообщений
self.app.message_handler(filters.TEXT)(self.handle_message)
async def handle_start(self, update, context):
"""Обработка команды /start"""
user_id = update.effective_user.id
welcome_text = f"""
🏠 Добро пожаловать в умный дом, {update.effective_user.first_name}!
Я ваш персональный помощник по управлению умными устройствами.
🔧 Доступные команды:
/devices - Управление устройствами
/rooms - Управление комнатами
/automation - Автоматизация
/security - Безопасность
/energy - Мониторинг энергии
/scenes - Сцены освещения
💡 Примеры команд:
"Включи свет в гостиной"
"Установи температуру 22 градуса"
"Покажи камеры"
"Включи режим 'Дома'"
Готовы управлять вашим умным домом?
"""
await update.message.reply_text(welcome_text)
await self.show_main_menu(update, context)
async def show_main_menu(self, update, context):
"""Показ главного меню"""
keyboard = [
[InlineKeyboardButton("🏠 Устройства", callback_data="devices_menu")],
[InlineKeyboardButton("🚪 Комнаты", callback_data="rooms_menu")],
[InlineKeyboardButton("🤖 Автоматизация", callback_data="automation_menu")],
[InlineKeyboardButton("🔒 Безопасность", callback_data="security_menu")],
[InlineKeyboardButton("⚡ Энергия", callback_data="energy_menu")],
[InlineKeyboardButton("🎭 Сцены", callback_data="scenes_menu")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text("Выберите раздел:", reply_markup=reply_markup)
async def handle_devices(self, update, context):
"""Обработка команды /devices"""
if not self.devices:
await update.message.reply_text("Устройства не найдены. Запустите поиск устройств.")
return
devices_text = "🏠 Ваши устройства:\n\n"
for device_id, device in self.devices.items():
status_emoji = "🟢" if device.status == DeviceStatus.ONLINE else "🔴"
devices_text += f"{status_emoji} {device.name}\n"
devices_text += f"📍 Комната: {device.room}\n"
devices_text += f"🔧 Тип: {device.type.value}\n"
devices_text += f"📡 Протокол: {device.protocol.value}\n"
devices_text += f"⏰ Последний раз: {device.last_seen.strftime('%H:%M')}\n\n"
await update.message.reply_text(devices_text)
await self.show_device_controls(update, context)
async def show_device_controls(self, update, context):
"""Показ элементов управления устройствами"""
keyboard = []
# Группируем устройства по типам
device_groups = {}
for device in self.devices.values():
if device.type not in device_groups:
device_groups[device.type] = []
device_groups[device.type].append(device)
# Создаем кнопки для каждого типа устройств
for device_type, devices in device_groups.items():
if device_type == DeviceType.LIGHT:
keyboard.append([InlineKeyboardButton("💡 Управление светом", callback_data="lights_control")])
elif device_type == DeviceType.THERMOSTAT:
keyboard.append([InlineKeyboardButton("🌡️ Термостат", callback_data="thermostat_control")])
elif device_type == DeviceType.CAMERA:
keyboard.append([InlineKeyboardButton("📹 Камеры", callback_data="cameras_control")])
elif device_type == DeviceType.DOOR_LOCK:
keyboard.append([InlineKeyboardButton("🔐 Дверные замки", callback_data="locks_control")])
keyboard.append([InlineKeyboardButton("🔍 Поиск устройств", callback_data="discover_devices")])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text("Выберите тип устройств:", reply_markup=reply_markup)
async def handle_lights_control(self, update, context):
"""Управление освещением"""
lights = [device for device in self.devices.values() if device.type == DeviceType.LIGHT]
if not lights:
await update.message.reply_text("Осветительные приборы не найдены.")
return
lights_text = "💡 Управление освещением:\n\n"
for light in lights:
brightness = light.current_state.get('brightness', 0)
is_on = light.current_state.get('on', False)
status = "🟡 Включен" if is_on else "⚫ Выключен"
lights_text += f"{status} {light.name}\n"
lights_text += f"📍 {light.room} | Яркость: {brightness}%\n\n"
# Создаем кнопки управления
keyboard = []
# Кнопки для каждой лампы
for light in lights:
row = [
InlineKeyboardButton(f"💡 {light.name}", callback_data=f"light_toggle_{light.id}"),
InlineKeyboardButton("🔧", callback_data=f"light_settings_{light.id}")
]
keyboard.append(row)
# Общие команды
keyboard.append([
InlineKeyboardButton("🌅 Включить все", callback_data="lights_all_on"),
InlineKeyboardButton("🌙 Выключить все", callback_data="lights_all_off")
])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(lights_text, reply_markup=reply_markup)
async def toggle_light(self, device_id: str) -> bool:
"""Переключение состояния лампы"""
if device_id not in self.devices:
return False
device = self.devices[device_id]
if device.type != DeviceType.LIGHT:
return False
# Получаем текущее состояние
current_state = device.current_state.get('on', False)
new_state = not current_state
# Отправляем команду устройству
success = await self.send_device_command(device, 'set_state', {'on': new_state})
if success:
device.current_state['on'] = new_state
device.last_seen = datetime.now()
# Запускаем автоматизацию
await self.check_automation_triggers(device)
return True
return False
async def set_light_brightness(self, device_id: str, brightness: int) -> bool:
"""Установка яркости лампы"""
if device_id not in self.devices:
return False
device = self.devices[device_id]
if device.type != DeviceType.LIGHT:
return False
# Ограничиваем яркость
brightness = max(0, min(100, brightness))
# Отправляем команду устройству
success = await self.send_device_command(device, 'set_brightness', {'brightness': brightness})
if success:
device.current_state['brightness'] = brightness
device.current_state['on'] = brightness > 0
device.last_seen = datetime.now()
return True
return False
async def handle_thermostat_control(self, update, context):
"""Управление термостатом"""
thermostats = [device for device in self.devices.values() if device.type == DeviceType.THERMOSTAT]
if not thermostats:
await update.message.reply_text("Термостаты не найдены.")
return
thermostat_text = "🌡️ Управление термостатом:\n\n"
for thermostat in thermostats:
current_temp = thermostat.current_state.get('temperature', 0)
target_temp = thermostat.current_state.get('target_temperature', 0)
mode = thermostat.current_state.get('mode', 'auto')
thermostat_text += f"🏠 {thermostat.name}\n"
thermostat_text += f"📍 {thermostat.room}\n"
thermostat_text += f"🌡️ Текущая: {current_temp}°C\n"
thermostat_text += f"🎯 Целевая: {target_temp}°C\n"
thermostat_text += f"⚙️ Режим: {mode}\n\n"
# Создаем кнопки управления
keyboard = []
for thermostat in thermostats:
keyboard.append([
InlineKeyboardButton("🌡️ Настройки", callback_data=f"thermostat_settings_{thermostat.id}")
])
# Быстрые команды температуры
keyboard.append([
InlineKeyboardButton("❄️ 18°C", callback_data="temp_18"),
InlineKeyboardButton("🌡️ 22°C", callback_data="temp_22"),
InlineKeyboardButton("🔥 25°C", callback_data="temp_25")
])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(thermostat_text, reply_markup=reply_markup)
async def set_temperature(self, device_id: str, temperature: float) -> bool:
"""Установка температуры"""
if device_id not in self.devices:
return False
device = self.devices[device_id]
if device.type != DeviceType.THERMOSTAT:
return False
# Ограничиваем температуру
temperature = max(16, min(30, temperature))
# Отправляем команду устройству
success = await self.send_device_command(device, 'set_temperature', {'temperature': temperature})
if success:
device.current_state['target_temperature'] = temperature
device.last_seen = datetime.now()
return True
return False
async def handle_cameras_control(self, update, context):
"""Управление камерами"""
cameras = [device for device in self.devices.values() if device.type == DeviceType.CAMERA]
if not cameras:
await update.message.reply_text("Камеры не найдены.")
return
cameras_text = "📹 Управление камерами:\n\n"
for camera in cameras:
is_recording = camera.current_state.get('recording', False)
motion_detection = camera.current_state.get('motion_detection', False)
status = "🔴 Запись" if is_recording else "⏸️ Остановлена"
motion = "👁️ Детекция движения" if motion_detection else "👁️🗨️ Без детекции"
cameras_text += f"{status} {camera.name}\n"
cameras_text += f"📍 {camera.room} | {motion}\n\n"
# Создаем кнопки управления
keyboard = []
for camera in cameras:
keyboard.append([
InlineKeyboardButton(f"📹 {camera.name}", callback_data=f"camera_view_{camera.id}"),
InlineKeyboardButton("⚙️", callback_data=f"camera_settings_{camera.id}")
])
keyboard.append([
InlineKeyboardButton("🔴 Начать запись всех", callback_data="cameras_record_all"),
InlineKeyboardButton("⏸️ Остановить все", callback_data="cameras_stop_all")
])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(cameras_text, reply_markup=reply_markup)
async def get_camera_feed(self, device_id: str) -> Optional[str]:
"""Получение потока с камеры"""
if device_id not in self.devices:
return None
device = self.devices[device_id]
if device.type != DeviceType.CAMERA:
return None
# Получаем URL потока
stream_url = await self.send_device_command(device, 'get_stream_url', {})
return stream_url
async def send_device_command(self, device: IoTDevice, command: str, params: Dict[str, Any]) -> Any:
"""Отправка команды устройству"""
try:
protocol_handler = self.protocols.get(device.protocol)
if not protocol_handler:
return False
# Отправляем команду через соответствующий протокол
result = await protocol_handler.send_command(device, command, params)
return result
except Exception as e:
logging.error(f"Failed to send command to device {device.id}: {e}")
return False
def start_device_discovery(self):
"""Запуск поиска устройств"""
# Запускаем поиск устройств в фоновом режиме
asyncio.create_task(self.discover_devices())
async def discover_devices(self):
"""Поиск устройств в сети"""
while True:
try:
# Поиск устройств по каждому протоколу
for protocol, handler in self.protocols.items():
devices = await handler.discover_devices()
for device_data in devices:
device_id = device_data['id']
if device_id not in self.devices:
# Создаем новое устройство
device = IoTDevice(
id=device_id,
name=device_data.get('name', f'Device {device_id}'),
type=DeviceType(device_data.get('type', 'switch')),
protocol=protocol,
status=DeviceStatus.ONLINE,
room=device_data.get('room', 'Unknown'),
capabilities=device_data.get('capabilities', []),
current_state=device_data.get('state', {}),
last_seen=datetime.now(),
ip_address=device_data.get('ip_address'),
mac_address=device_data.get('mac_address')
)
self.devices[device_id] = device
# Уведомляем о новом устройстве
await self.notify_new_device(device)
# Ждем перед следующим поиском
await asyncio.sleep(30)
except Exception as e:
logging.error(f"Device discovery error: {e}")
await asyncio.sleep(60)
```
### 2. 🤖 Система автоматизации
```python
class AutomationEngine:
def __init__(self):
self.rules = {}
self.triggers = {}
self.conditions = {}
self.actions = {}
self.setup_default_rules()
def setup_default_rules(self):
"""Настройка правил автоматизации по умолчанию"""
# Правило "Дома"
home_rule = AutomationRule(
id="home_mode",
name="Режим 'Дома'",
trigger_conditions={
'type': 'presence',
'condition': 'user_arrives',
'time_range': {'start': '18:00', 'end': '23:00'}
},
actions=[
{'type': 'light', 'action': 'turn_on', 'room': 'living_room'},
{'type': 'thermostat', 'action': 'set_temperature', 'value': 22},
{'type': 'security', 'action': 'disable_alarm'}
],
enabled=True,
created_at=datetime.now()
)
# Правило "Спокойной ночи"
night_rule = AutomationRule(
id="good_night",
name="Режим 'Спокойной ночи'",
trigger_conditions={
'type': 'time',
'time': '23:00',
'days': ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']
},
actions=[
{'type': 'light', 'action': 'turn_off', 'room': 'all'},
{'type': 'thermostat', 'action': 'set_temperature', 'value': 18},
{'type': 'security', 'action': 'enable_alarm'},
{'type': 'camera', 'action': 'start_recording'}
],
enabled=True,
created_at=datetime.now()
)
# Правило "Экономия энергии"
energy_rule = AutomationRule(
id="energy_saving",
name="Экономия энергии",
trigger_conditions={
'type': 'absence',
'condition': 'no_motion',
'duration': 30 # минут
},
actions=[
{'type': 'light', 'action': 'turn_off', 'room': 'all'},
{'type': 'thermostat', 'action': 'set_temperature', 'value': 16}
],
enabled=True,
created_at=datetime.now()
)
self.rules = {
'home_mode': home_rule,
'good_night': night_rule,
'energy_saving': energy_rule
}
async def check_automation_triggers(self, device: IoTDevice):
"""Проверка триггеров автоматизации"""
for rule_id, rule in self.rules.items():
if not rule.enabled:
continue
# Проверяем условия триггера
if await self.evaluate_trigger_conditions(rule.trigger_conditions, device):
# Выполняем действия
await self.execute_rule_actions(rule.actions)
# Обновляем время последнего срабатывания
rule.last_triggered = datetime.now()
async def evaluate_trigger_conditions(self, conditions: Dict[str, Any], device: IoTDevice) -> bool:
"""Оценка условий триггера"""
trigger_type = conditions.get('type')
if trigger_type == 'presence':
return await self.check_presence_trigger(conditions)
elif trigger_type == 'time':
return await self.check_time_trigger(conditions)
elif trigger_type == 'device_state':
return await self.check_device_state_trigger(conditions, device)
elif trigger_type == 'absence':
return await self.check_absence_trigger(conditions)
elif trigger_type == 'motion':
return await self.check_motion_trigger(conditions, device)
return False
async def check_presence_trigger(self, conditions: Dict[str, Any]) -> bool:
"""Проверка триггера присутствия"""
# Здесь должна быть логика определения присутствия пользователя
# Упрощенная версия
return True
async def check_time_trigger(self, conditions: Dict[str, Any]) -> bool:
"""Проверка триггера времени"""
current_time = datetime.now().time()
trigger_time = datetime.strptime(conditions['time'], '%H:%M').time()
# Проверяем дни недели
current_day = datetime.now().strftime('%A').lower()
allowed_days = conditions.get('days', [])
if allowed_days and current_day not in allowed_days:
return False
# Проверяем время (с допуском в 5 минут)
time_diff = abs((current_time.hour * 60 + current_time.minute) -
(trigger_time.hour * 60 + trigger_time.minute))
return time_diff <= 5
async def check_device_state_trigger(self, conditions: Dict[str, Any], device: IoTDevice) -> bool:
"""Проверка триггера состояния устройства"""
device_type = conditions.get('device_type')
state_condition = conditions.get('state')
if device.type.value != device_type:
return False
# Проверяем состояние устройства
if state_condition == 'on' and device.current_state.get('on', False):
return True
elif state_condition == 'off' and not device.current_state.get('on', False):
return True
return False
async def check_absence_trigger(self, conditions: Dict[str, Any]) -> bool:
"""Проверка триггера отсутствия"""
# Здесь должна быть логика определения отсутствия пользователя
# Упрощенная версия
return False
async def check_motion_trigger(self, conditions: Dict[str, Any], device: IoTDevice) -> bool:
"""Проверка триггера движения"""
if device.type != DeviceType.SENSOR:
return False
# Проверяем детекцию движения
motion_detected = device.current_state.get('motion_detected', False)
return motion_detected
async def execute_rule_actions(self, actions: List[Dict[str, Any]]):
"""Выполнение действий правила"""
for action in actions:
action_type = action.get('type')
if action_type == 'light':
await self.execute_light_action(action)
elif action_type == 'thermostat':
await self.execute_thermostat_action(action)
elif action_type == 'security':
await self.execute_security_action(action)
elif action_type == 'camera':
await self.execute_camera_action(action)
async def execute_light_action(self, action: Dict[str, Any]):
"""Выполнение действия с освещением"""
action_type = action.get('action')
room = action.get('room')
if action_type == 'turn_on':
# Включаем свет в комнате
lights = [device for device in self.devices.values()
if device.type == DeviceType.LIGHT and device.room == room]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': True})
elif action_type == 'turn_off':
# Выключаем свет в комнате
lights = [device for device in self.devices.values()
if device.type == DeviceType.LIGHT and device.room == room]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': False})
async def execute_thermostat_action(self, action: Dict[str, Any]):
"""Выполнение действия с термостатом"""
action_type = action.get('action')
temperature = action.get('value')
if action_type == 'set_temperature':
# Устанавливаем температуру
thermostats = [device for device in self.devices.values()
if device.type == DeviceType.THERMOSTAT]
for thermostat in thermostats:
await self.send_device_command(thermostat, 'set_temperature', {'temperature': temperature})
async def execute_security_action(self, action: Dict[str, Any]):
"""Выполнение действия с системой безопасности"""
action_type = action.get('action')
if action_type == 'enable_alarm':
# Включаем сигнализацию
security_devices = [device for device in self.devices.values()
if device.type == DeviceType.SECURITY_SYSTEM]
for device in security_devices:
await self.send_device_command(device, 'enable_alarm', {})
elif action_type == 'disable_alarm':
# Выключаем сигнализацию
security_devices = [device for device in self.devices.values()
if device.type == DeviceType.SECURITY_SYSTEM]
for device in security_devices:
await self.send_device_command(device, 'disable_alarm', {})
async def execute_camera_action(self, action: Dict[str, Any]):
"""Выполнение действия с камерами"""
action_type = action.get('action')
if action_type == 'start_recording':
# Начинаем запись
cameras = [device for device in self.devices.values()
if device.type == DeviceType.CAMERA]
for camera in cameras:
await self.send_device_command(camera, 'start_recording', {})
elif action_type == 'stop_recording':
# Останавливаем запись
cameras = [device for device in self.devices.values()
if device.type == DeviceType.CAMERA]
for camera in cameras:
await self.send_device_command(camera, 'stop_recording', {})
async def create_custom_rule(self, rule_data: Dict[str, Any]) -> str:
"""Создание пользовательского правила"""
rule_id = f"custom_{len(self.rules) + 1}_{datetime.now().timestamp()}"
rule = AutomationRule(
id=rule_id,
name=rule_data.get('name', 'Custom Rule'),
trigger_conditions=rule_data.get('trigger_conditions', {}),
actions=rule_data.get('actions', []),
enabled=rule_data.get('enabled', True),
created_at=datetime.now()
)
self.rules[rule_id] = rule
return rule_id
```
### 3. 🔒 Система безопасности
```python
class SecurityManager:
def __init__(self):
self.security_devices = {}
self.alarm_status = False
self.motion_detectors = {}
self.door_sensors = {}
self.window_sensors = {}
self.security_cameras = {}
async def enable_security_system(self):
"""Включение системы безопасности"""
self.alarm_status = True
# Активируем все датчики движения
for sensor_id, sensor in self.motion_detectors.items():
await self.send_device_command(sensor, 'enable', {})
# Активируем датчики дверей и окон
for sensor_id, sensor in self.door_sensors.items():
await self.send_device_command(sensor, 'enable', {})
for sensor_id, sensor in self.window_sensors.items():
await self.send_device_command(sensor, 'enable', {})
# Включаем камеры безопасности
for camera_id, camera in self.security_cameras.items():
await self.send_device_command(camera, 'start_recording', {})
return True
async def disable_security_system(self):
"""Отключение системы безопасности"""
self.alarm_status = False
# Деактивируем все датчики
for sensor_id, sensor in self.motion_detectors.items():
await self.send_device_command(sensor, 'disable', {})
for sensor_id, sensor in self.door_sensors.items():
await self.send_device_command(sensor, 'disable', {})
for sensor_id, sensor in self.window_sensors.items():
await self.send_device_command(sensor, 'disable', {})
# Останавливаем запись камер
for camera_id, camera in self.security_cameras.items():
await self.send_device_command(camera, 'stop_recording', {})
return True
async def handle_security_alert(self, device: IoTDevice, alert_type: str):
"""Обработка сигнала безопасности"""
if not self.alarm_status:
return
alert_message = f"🚨 СИГНАЛ БЕЗОПАСНОСТИ!\n\n"
alert_message += f"📍 Устройство: {device.name}\n"
alert_message += f"🏠 Комната: {device.room}\n"
alert_message += f"⚠️ Тип сигнала: {alert_type}\n"
alert_message += f"⏰ Время: {datetime.now().strftime('%H:%M:%S')}\n"
# Отправляем уведомление пользователю
await self.send_security_notification(alert_message)
# Активируем дополнительные меры безопасности
await self.activate_security_measures(device, alert_type)
async def activate_security_measures(self, device: IoTDevice, alert_type: str):
"""Активация мер безопасности"""
if alert_type == 'motion_detected':
# Включаем все камеры
for camera_id, camera in self.security_cameras.items():
await self.send_device_command(camera, 'start_recording', {})
# Включаем освещение
lights = [d for d in self.devices.values() if d.type == DeviceType.LIGHT]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': True})
elif alert_type == 'door_opened':
# Проверяем, разрешено ли открытие двери
if not await self.is_authorized_access(device):
# Несанкционированный доступ
await self.trigger_alarm()
elif alert_type == 'window_opened':
# Окно открыто - потенциальная угроза
await self.trigger_alarm()
async def trigger_alarm(self):
"""Активация сигнализации"""
# Включаем звуковую сигнализацию
speakers = [device for device in self.devices.values() if device.type == DeviceType.SPEAKER]
for speaker in speakers:
await self.send_device_command(speaker, 'play_alarm', {})
# Включаем световую сигнализацию
lights = [device for device in self.devices.values() if device.type == DeviceType.LIGHT]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': True, 'brightness': 100})
# Отправляем уведомления
await self.send_emergency_notifications()
async def send_security_notification(self, message: str):
"""Отправка уведомления о безопасности"""
# Здесь должна быть логика отправки уведомлений
# Упрощенная версия
print(f"Security Alert: {message}")
async def send_emergency_notifications(self):
"""Отправка экстренных уведомлений"""
emergency_message = """
🚨 ЭКСТРЕННАЯ СИТУАЦИЯ! 🚨
В вашем доме сработала сигнализация!
Возможно несанкционированное проникновение.
Рекомендуется:
• Немедленно связаться с полицией
• Проверить камеры безопасности
• Не входить в дом до прибытия полиции
Время: {datetime.now().strftime('%H:%M:%S')}
"""
# Отправляем уведомления всем контактам
await self.send_security_notification(emergency_message)
async def is_authorized_access(self, device: IoTDevice) -> bool:
"""Проверка авторизованного доступа"""
# Здесь должна быть логика проверки авторизации
# Упрощенная версия
return False
```
### 4. ⚡ Мониторинг энергии
```python
class EnergyMonitor:
def __init__(self):
self.energy_consumption = {}
self.power_sensors = {}
self.smart_plugs = {}
self.energy_reports = {}
async def monitor_energy_consumption(self):
"""Мониторинг потребления энергии"""
while True:
try:
# Собираем данные о потреблении энергии
for device_id, device in self.devices.items():
if device.type in [DeviceType.LIGHT, DeviceType.THERMOSTAT, DeviceType.SWITCH]:
power_consumption = await self.get_device_power_consumption(device)
if device_id not in self.energy_consumption:
self.energy_consumption[device_id] = []
self.energy_consumption[device_id].append({
'timestamp': datetime.now(),
'power': power_consumption,
'device_name': device.name,
'room': device.room
})
# Генерируем отчеты
await self.generate_energy_reports()
# Ждем перед следующим измерением
await asyncio.sleep(300) # 5 минут
except Exception as e:
logging.error(f"Energy monitoring error: {e}")
await asyncio.sleep(600)
async def get_device_power_consumption(self, device: IoTDevice) -> float:
"""Получение потребления энергии устройством"""
# Здесь должна быть логика получения данных о потреблении
# Упрощенная версия
base_consumption = {
DeviceType.LIGHT: 10.0, # Вт
DeviceType.THERMOSTAT: 50.0, # Вт
DeviceType.SWITCH: 5.0, # Вт
DeviceType.CAMERA: 15.0, # Вт
DeviceType.SPEAKER: 20.0 # Вт
}
consumption = base_consumption.get(device.type, 0.0)
# Корректируем в зависимости от состояния
if device.type == DeviceType.LIGHT:
brightness = device.current_state.get('brightness', 0)
consumption *= (brightness / 100.0)
return consumption
async def generate_energy_reports(self):
"""Генерация отчетов по энергии"""
current_hour = datetime.now().hour
# Генерируем отчет за час
hourly_report = await self.calculate_hourly_consumption()
# Генерируем отчет за день
daily_report = await self.calculate_daily_consumption()
# Генерируем отчет за месяц
monthly_report = await self.calculate_monthly_consumption()
self.energy_reports = {
'hourly': hourly_report,
'daily': daily_report,
'monthly': monthly_report
}
async def calculate_hourly_consumption(self) -> Dict[str, Any]:
"""Расчет потребления за час"""
current_hour = datetime.now().hour
hourly_consumption = 0.0
for device_id, consumption_data in self.energy_consumption.items():
# Берем данные за последний час
recent_data = [data for data in consumption_data
if data['timestamp'].hour == current_hour]
if recent_data:
avg_power = sum(data['power'] for data in recent_data) / len(recent_data)
hourly_consumption += avg_power
return {
'total_consumption': hourly_consumption,
'cost': hourly_consumption * 0.05, # 5 копеек за кВт⋅ч
'hour': current_hour
}
async def calculate_daily_consumption(self) -> Dict[str, Any]:
"""Расчет потребления за день"""
today = datetime.now().date()
daily_consumption = 0.0
for device_id, consumption_data in self.energy_consumption.items():
# Берем данные за сегодня
today_data = [data for data in consumption_data
if data['timestamp'].date() == today]
if today_data:
avg_power = sum(data['power'] for data in today_data) / len(today_data)
daily_consumption += avg_power
return {
'total_consumption': daily_consumption,
'cost': daily_consumption * 0.05,
'date': today
}
async def calculate_monthly_consumption(self) -> Dict[str, Any]:
"""Расчет потребления за месяц"""
current_month = datetime.now().month
monthly_consumption = 0.0
for device_id, consumption_data in self.energy_consumption.items():
# Берем данные за текущий месяц
month_data = [data for data in consumption_data
if data['timestamp'].month == current_month]
if month_data:
avg_power = sum(data['power'] for data in month_data) / len(month_data)
monthly_consumption += avg_power
return {
'total_consumption': monthly_consumption,
'cost': monthly_consumption * 0.05,
'month': current_month
}
async def get_energy_savings_tips(self) -> List[str]:
"""Получение советов по экономии энергии"""
tips = []
# Анализируем потребление и даем советы
if self.energy_reports:
daily_consumption = self.energy_reports.get('daily', {}).get('total_consumption', 0)
if daily_consumption > 1000: # Высокое потребление
tips.append("💡 Высокое потребление энергии! Проверьте настройки термостата.")
tips.append("🌡️ Установите температуру на 1-2 градуса ниже для экономии.")
if daily_consumption > 500:
tips.append("💡 Отключайте неиспользуемые устройства.")
tips.append("🔌 Используйте умные розетки для автоматического отключения.")
return tips
```
## Заключение
IoT-боты - это не просто управление устройствами, это создание интеллектуальной экосистемы умного дома. За 4 года работы я понял, что:
### 🎯 **Главные принципы IoT-ботов:**
1. **Централизованное управление** - все устройства в одном месте
2. **Автоматизация сценариев** - умные алгоритмы
3. **Безопасность** - защита от несанкционированного доступа
4. **Энергоэффективность** - мониторинг и оптимизация
### 💡 **Что работает лучше всего:**
- **Автоматические сценарии** - "Дома", "Спокойной ночи", "Экономия"
- **Голосовое управление** - удобство использования
- **Мониторинг безопасности** - защита дома
- **Аналитика энергии** - оптимизация потребления
### ⚠️ **Типичные ошибки:**
- Игнорирование безопасности устройств
- Отсутствие резервных систем
- Слабая автоматизация
- Недооценка важности мониторинга
### 🚀 **Советы для успеха:**
1. **Начните с простых устройств** - свет, термостат
2. **Создайте автоматические сценарии** - повысьте удобство
3. **Обеспечьте безопасность** - защитите от взлома
4. **Мониторьте энергию** - оптимизируйте потребление
Помните: лучший IoT-бот - это тот, который работает незаметно, но значительно повышает комфорт и безопасность дома!
---
*Готовы создать умный дом? Обращайтесь к нам за хостингом, консультациями и технической поддержкой!*
Привет! Меня зовут Дмитрий, и я уже 4 года разрабатываю IoT-ботов для умных домов и промышленной автоматизации. За это время я создал более 60 IoT-ботов, подключил более 10,000 устройств и автоматизировал процессы в 200+ умных домах. В этой статье расскажу, как создать мощного IoT-бота, какие протоколы использовать, и как обеспечить безопасность подключенных устройств.
## Почему IoT-боты становятся необходимостью?
### Статистика IoT рынка
- **75 миллиардов** IoT устройств к 2025 году
- **$1.1 триллион** объем рынка IoT
- **67%** домовладельцев используют умные устройства
- **45%** экономии энергии при автоматизации
### Преимущества IoT-ботов
- **Централизованное управление** - все устройства в одном месте
- **Автоматизация сценариев** - умные алгоритмы
- **Голосовое управление** - удобство использования
- **Мониторинг и аналитика** - контроль состояния
## Архитектура IoT-бота
### 1. 🏠 Система управления умным домом
```python
import asyncio
import json
import aiohttp
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
import logging
class DeviceType(Enum):
LIGHT = "light"
THERMOSTAT = "thermostat"
CAMERA = "camera"
DOOR_LOCK = "door_lock"
SENSOR = "sensor"
SWITCH = "switch"
SPEAKER = "speaker"
SECURITY_SYSTEM = "security_system"
class DeviceStatus(Enum):
ONLINE = "online"
OFFLINE = "offline"
ERROR = "error"
MAINTENANCE = "maintenance"
class Protocol(Enum):
ZIGBEE = "zigbee"
Z_WAVE = "z_wave"
WIFI = "wifi"
BLUETOOTH = "bluetooth"
THREAD = "thread"
MATTER = "matter"
@dataclass
class IoTDevice:
id: str
name: str
type: DeviceType
protocol: Protocol
status: DeviceStatus
room: str
capabilities: List[str]
current_state: Dict[str, Any]
last_seen: datetime
ip_address: Optional[str] = None
mac_address: Optional[str] = None
@dataclass
class AutomationRule:
id: str
name: str
trigger_conditions: Dict[str, Any]
actions: List[Dict[str, Any]]
enabled: bool
created_at: datetime
last_triggered: Optional[datetime] = None
class IoTHomeBot:
def __init__(self, bot_token: str):
self.bot_token = bot_token
self.devices = {}
self.rooms = {}
self.automation_rules = {}
self.scenes = {}
self.schedules = {}
# Протоколы и интеграции
self.protocols = {
Protocol.ZIGBEE: ZigbeeProtocol(),
Protocol.Z_WAVE: ZWaveProtocol(),
Protocol.WIFI: WiFiProtocol(),
Protocol.BLUETOOTH: BluetoothProtocol(),
Protocol.MATTER: MatterProtocol()
}
# Системы безопасности
self.security_manager = SecurityManager()
self.energy_monitor = EnergyMonitor()
self.setup_handlers()
self.start_device_discovery()
def setup_handlers(self):
"""Настройка обработчиков команд"""
self.app.command('/start', self.handle_start)
self.app.command('/devices', self.handle_devices)
self.app.command('/rooms', self.handle_rooms)
self.app.command('/automation', self.handle_automation)
self.app.command('/security', self.handle_security)
self.app.command('/energy', self.handle_energy)
self.app.command('/scenes', self.handle_scenes)
# Обработчики сообщений
self.app.message_handler(filters.TEXT)(self.handle_message)
async def handle_start(self, update, context):
"""Обработка команды /start"""
user_id = update.effective_user.id
welcome_text = f"""
🏠 Добро пожаловать в умный дом, {update.effective_user.first_name}!
Я ваш персональный помощник по управлению умными устройствами.
🔧 Доступные команды:
/devices - Управление устройствами
/rooms - Управление комнатами
/automation - Автоматизация
/security - Безопасность
/energy - Мониторинг энергии
/scenes - Сцены освещения
💡 Примеры команд:
"Включи свет в гостиной"
"Установи температуру 22 градуса"
"Покажи камеры"
"Включи режим 'Дома'"
Готовы управлять вашим умным домом?
"""
await update.message.reply_text(welcome_text)
await self.show_main_menu(update, context)
async def show_main_menu(self, update, context):
"""Показ главного меню"""
keyboard = [
[InlineKeyboardButton("🏠 Устройства", callback_data="devices_menu")],
[InlineKeyboardButton("🚪 Комнаты", callback_data="rooms_menu")],
[InlineKeyboardButton("🤖 Автоматизация", callback_data="automation_menu")],
[InlineKeyboardButton("🔒 Безопасность", callback_data="security_menu")],
[InlineKeyboardButton("⚡ Энергия", callback_data="energy_menu")],
[InlineKeyboardButton("🎭 Сцены", callback_data="scenes_menu")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text("Выберите раздел:", reply_markup=reply_markup)
async def handle_devices(self, update, context):
"""Обработка команды /devices"""
if not self.devices:
await update.message.reply_text("Устройства не найдены. Запустите поиск устройств.")
return
devices_text = "🏠 Ваши устройства:\n\n"
for device_id, device in self.devices.items():
status_emoji = "🟢" if device.status == DeviceStatus.ONLINE else "🔴"
devices_text += f"{status_emoji} {device.name}\n"
devices_text += f"📍 Комната: {device.room}\n"
devices_text += f"🔧 Тип: {device.type.value}\n"
devices_text += f"📡 Протокол: {device.protocol.value}\n"
devices_text += f"⏰ Последний раз: {device.last_seen.strftime('%H:%M')}\n\n"
await update.message.reply_text(devices_text)
await self.show_device_controls(update, context)
async def show_device_controls(self, update, context):
"""Показ элементов управления устройствами"""
keyboard = []
# Группируем устройства по типам
device_groups = {}
for device in self.devices.values():
if device.type not in device_groups:
device_groups[device.type] = []
device_groups[device.type].append(device)
# Создаем кнопки для каждого типа устройств
for device_type, devices in device_groups.items():
if device_type == DeviceType.LIGHT:
keyboard.append([InlineKeyboardButton("💡 Управление светом", callback_data="lights_control")])
elif device_type == DeviceType.THERMOSTAT:
keyboard.append([InlineKeyboardButton("🌡️ Термостат", callback_data="thermostat_control")])
elif device_type == DeviceType.CAMERA:
keyboard.append([InlineKeyboardButton("📹 Камеры", callback_data="cameras_control")])
elif device_type == DeviceType.DOOR_LOCK:
keyboard.append([InlineKeyboardButton("🔐 Дверные замки", callback_data="locks_control")])
keyboard.append([InlineKeyboardButton("🔍 Поиск устройств", callback_data="discover_devices")])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text("Выберите тип устройств:", reply_markup=reply_markup)
async def handle_lights_control(self, update, context):
"""Управление освещением"""
lights = [device for device in self.devices.values() if device.type == DeviceType.LIGHT]
if not lights:
await update.message.reply_text("Осветительные приборы не найдены.")
return
lights_text = "💡 Управление освещением:\n\n"
for light in lights:
brightness = light.current_state.get('brightness', 0)
is_on = light.current_state.get('on', False)
status = "🟡 Включен" if is_on else "⚫ Выключен"
lights_text += f"{status} {light.name}\n"
lights_text += f"📍 {light.room} | Яркость: {brightness}%\n\n"
# Создаем кнопки управления
keyboard = []
# Кнопки для каждой лампы
for light in lights:
row = [
InlineKeyboardButton(f"💡 {light.name}", callback_data=f"light_toggle_{light.id}"),
InlineKeyboardButton("🔧", callback_data=f"light_settings_{light.id}")
]
keyboard.append(row)
# Общие команды
keyboard.append([
InlineKeyboardButton("🌅 Включить все", callback_data="lights_all_on"),
InlineKeyboardButton("🌙 Выключить все", callback_data="lights_all_off")
])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(lights_text, reply_markup=reply_markup)
async def toggle_light(self, device_id: str) -> bool:
"""Переключение состояния лампы"""
if device_id not in self.devices:
return False
device = self.devices[device_id]
if device.type != DeviceType.LIGHT:
return False
# Получаем текущее состояние
current_state = device.current_state.get('on', False)
new_state = not current_state
# Отправляем команду устройству
success = await self.send_device_command(device, 'set_state', {'on': new_state})
if success:
device.current_state['on'] = new_state
device.last_seen = datetime.now()
# Запускаем автоматизацию
await self.check_automation_triggers(device)
return True
return False
async def set_light_brightness(self, device_id: str, brightness: int) -> bool:
"""Установка яркости лампы"""
if device_id not in self.devices:
return False
device = self.devices[device_id]
if device.type != DeviceType.LIGHT:
return False
# Ограничиваем яркость
brightness = max(0, min(100, brightness))
# Отправляем команду устройству
success = await self.send_device_command(device, 'set_brightness', {'brightness': brightness})
if success:
device.current_state['brightness'] = brightness
device.current_state['on'] = brightness > 0
device.last_seen = datetime.now()
return True
return False
async def handle_thermostat_control(self, update, context):
"""Управление термостатом"""
thermostats = [device for device in self.devices.values() if device.type == DeviceType.THERMOSTAT]
if not thermostats:
await update.message.reply_text("Термостаты не найдены.")
return
thermostat_text = "🌡️ Управление термостатом:\n\n"
for thermostat in thermostats:
current_temp = thermostat.current_state.get('temperature', 0)
target_temp = thermostat.current_state.get('target_temperature', 0)
mode = thermostat.current_state.get('mode', 'auto')
thermostat_text += f"🏠 {thermostat.name}\n"
thermostat_text += f"📍 {thermostat.room}\n"
thermostat_text += f"🌡️ Текущая: {current_temp}°C\n"
thermostat_text += f"🎯 Целевая: {target_temp}°C\n"
thermostat_text += f"⚙️ Режим: {mode}\n\n"
# Создаем кнопки управления
keyboard = []
for thermostat in thermostats:
keyboard.append([
InlineKeyboardButton("🌡️ Настройки", callback_data=f"thermostat_settings_{thermostat.id}")
])
# Быстрые команды температуры
keyboard.append([
InlineKeyboardButton("❄️ 18°C", callback_data="temp_18"),
InlineKeyboardButton("🌡️ 22°C", callback_data="temp_22"),
InlineKeyboardButton("🔥 25°C", callback_data="temp_25")
])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(thermostat_text, reply_markup=reply_markup)
async def set_temperature(self, device_id: str, temperature: float) -> bool:
"""Установка температуры"""
if device_id not in self.devices:
return False
device = self.devices[device_id]
if device.type != DeviceType.THERMOSTAT:
return False
# Ограничиваем температуру
temperature = max(16, min(30, temperature))
# Отправляем команду устройству
success = await self.send_device_command(device, 'set_temperature', {'temperature': temperature})
if success:
device.current_state['target_temperature'] = temperature
device.last_seen = datetime.now()
return True
return False
async def handle_cameras_control(self, update, context):
"""Управление камерами"""
cameras = [device for device in self.devices.values() if device.type == DeviceType.CAMERA]
if not cameras:
await update.message.reply_text("Камеры не найдены.")
return
cameras_text = "📹 Управление камерами:\n\n"
for camera in cameras:
is_recording = camera.current_state.get('recording', False)
motion_detection = camera.current_state.get('motion_detection', False)
status = "🔴 Запись" if is_recording else "⏸️ Остановлена"
motion = "👁️ Детекция движения" if motion_detection else "👁️🗨️ Без детекции"
cameras_text += f"{status} {camera.name}\n"
cameras_text += f"📍 {camera.room} | {motion}\n\n"
# Создаем кнопки управления
keyboard = []
for camera in cameras:
keyboard.append([
InlineKeyboardButton(f"📹 {camera.name}", callback_data=f"camera_view_{camera.id}"),
InlineKeyboardButton("⚙️", callback_data=f"camera_settings_{camera.id}")
])
keyboard.append([
InlineKeyboardButton("🔴 Начать запись всех", callback_data="cameras_record_all"),
InlineKeyboardButton("⏸️ Остановить все", callback_data="cameras_stop_all")
])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(cameras_text, reply_markup=reply_markup)
async def get_camera_feed(self, device_id: str) -> Optional[str]:
"""Получение потока с камеры"""
if device_id not in self.devices:
return None
device = self.devices[device_id]
if device.type != DeviceType.CAMERA:
return None
# Получаем URL потока
stream_url = await self.send_device_command(device, 'get_stream_url', {})
return stream_url
async def send_device_command(self, device: IoTDevice, command: str, params: Dict[str, Any]) -> Any:
"""Отправка команды устройству"""
try:
protocol_handler = self.protocols.get(device.protocol)
if not protocol_handler:
return False
# Отправляем команду через соответствующий протокол
result = await protocol_handler.send_command(device, command, params)
return result
except Exception as e:
logging.error(f"Failed to send command to device {device.id}: {e}")
return False
def start_device_discovery(self):
"""Запуск поиска устройств"""
# Запускаем поиск устройств в фоновом режиме
asyncio.create_task(self.discover_devices())
async def discover_devices(self):
"""Поиск устройств в сети"""
while True:
try:
# Поиск устройств по каждому протоколу
for protocol, handler in self.protocols.items():
devices = await handler.discover_devices()
for device_data in devices:
device_id = device_data['id']
if device_id not in self.devices:
# Создаем новое устройство
device = IoTDevice(
id=device_id,
name=device_data.get('name', f'Device {device_id}'),
type=DeviceType(device_data.get('type', 'switch')),
protocol=protocol,
status=DeviceStatus.ONLINE,
room=device_data.get('room', 'Unknown'),
capabilities=device_data.get('capabilities', []),
current_state=device_data.get('state', {}),
last_seen=datetime.now(),
ip_address=device_data.get('ip_address'),
mac_address=device_data.get('mac_address')
)
self.devices[device_id] = device
# Уведомляем о новом устройстве
await self.notify_new_device(device)
# Ждем перед следующим поиском
await asyncio.sleep(30)
except Exception as e:
logging.error(f"Device discovery error: {e}")
await asyncio.sleep(60)
```
### 2. 🤖 Система автоматизации
```python
class AutomationEngine:
def __init__(self):
self.rules = {}
self.triggers = {}
self.conditions = {}
self.actions = {}
self.setup_default_rules()
def setup_default_rules(self):
"""Настройка правил автоматизации по умолчанию"""
# Правило "Дома"
home_rule = AutomationRule(
id="home_mode",
name="Режим 'Дома'",
trigger_conditions={
'type': 'presence',
'condition': 'user_arrives',
'time_range': {'start': '18:00', 'end': '23:00'}
},
actions=[
{'type': 'light', 'action': 'turn_on', 'room': 'living_room'},
{'type': 'thermostat', 'action': 'set_temperature', 'value': 22},
{'type': 'security', 'action': 'disable_alarm'}
],
enabled=True,
created_at=datetime.now()
)
# Правило "Спокойной ночи"
night_rule = AutomationRule(
id="good_night",
name="Режим 'Спокойной ночи'",
trigger_conditions={
'type': 'time',
'time': '23:00',
'days': ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']
},
actions=[
{'type': 'light', 'action': 'turn_off', 'room': 'all'},
{'type': 'thermostat', 'action': 'set_temperature', 'value': 18},
{'type': 'security', 'action': 'enable_alarm'},
{'type': 'camera', 'action': 'start_recording'}
],
enabled=True,
created_at=datetime.now()
)
# Правило "Экономия энергии"
energy_rule = AutomationRule(
id="energy_saving",
name="Экономия энергии",
trigger_conditions={
'type': 'absence',
'condition': 'no_motion',
'duration': 30 # минут
},
actions=[
{'type': 'light', 'action': 'turn_off', 'room': 'all'},
{'type': 'thermostat', 'action': 'set_temperature', 'value': 16}
],
enabled=True,
created_at=datetime.now()
)
self.rules = {
'home_mode': home_rule,
'good_night': night_rule,
'energy_saving': energy_rule
}
async def check_automation_triggers(self, device: IoTDevice):
"""Проверка триггеров автоматизации"""
for rule_id, rule in self.rules.items():
if not rule.enabled:
continue
# Проверяем условия триггера
if await self.evaluate_trigger_conditions(rule.trigger_conditions, device):
# Выполняем действия
await self.execute_rule_actions(rule.actions)
# Обновляем время последнего срабатывания
rule.last_triggered = datetime.now()
async def evaluate_trigger_conditions(self, conditions: Dict[str, Any], device: IoTDevice) -> bool:
"""Оценка условий триггера"""
trigger_type = conditions.get('type')
if trigger_type == 'presence':
return await self.check_presence_trigger(conditions)
elif trigger_type == 'time':
return await self.check_time_trigger(conditions)
elif trigger_type == 'device_state':
return await self.check_device_state_trigger(conditions, device)
elif trigger_type == 'absence':
return await self.check_absence_trigger(conditions)
elif trigger_type == 'motion':
return await self.check_motion_trigger(conditions, device)
return False
async def check_presence_trigger(self, conditions: Dict[str, Any]) -> bool:
"""Проверка триггера присутствия"""
# Здесь должна быть логика определения присутствия пользователя
# Упрощенная версия
return True
async def check_time_trigger(self, conditions: Dict[str, Any]) -> bool:
"""Проверка триггера времени"""
current_time = datetime.now().time()
trigger_time = datetime.strptime(conditions['time'], '%H:%M').time()
# Проверяем дни недели
current_day = datetime.now().strftime('%A').lower()
allowed_days = conditions.get('days', [])
if allowed_days and current_day not in allowed_days:
return False
# Проверяем время (с допуском в 5 минут)
time_diff = abs((current_time.hour * 60 + current_time.minute) -
(trigger_time.hour * 60 + trigger_time.minute))
return time_diff <= 5
async def check_device_state_trigger(self, conditions: Dict[str, Any], device: IoTDevice) -> bool:
"""Проверка триггера состояния устройства"""
device_type = conditions.get('device_type')
state_condition = conditions.get('state')
if device.type.value != device_type:
return False
# Проверяем состояние устройства
if state_condition == 'on' and device.current_state.get('on', False):
return True
elif state_condition == 'off' and not device.current_state.get('on', False):
return True
return False
async def check_absence_trigger(self, conditions: Dict[str, Any]) -> bool:
"""Проверка триггера отсутствия"""
# Здесь должна быть логика определения отсутствия пользователя
# Упрощенная версия
return False
async def check_motion_trigger(self, conditions: Dict[str, Any], device: IoTDevice) -> bool:
"""Проверка триггера движения"""
if device.type != DeviceType.SENSOR:
return False
# Проверяем детекцию движения
motion_detected = device.current_state.get('motion_detected', False)
return motion_detected
async def execute_rule_actions(self, actions: List[Dict[str, Any]]):
"""Выполнение действий правила"""
for action in actions:
action_type = action.get('type')
if action_type == 'light':
await self.execute_light_action(action)
elif action_type == 'thermostat':
await self.execute_thermostat_action(action)
elif action_type == 'security':
await self.execute_security_action(action)
elif action_type == 'camera':
await self.execute_camera_action(action)
async def execute_light_action(self, action: Dict[str, Any]):
"""Выполнение действия с освещением"""
action_type = action.get('action')
room = action.get('room')
if action_type == 'turn_on':
# Включаем свет в комнате
lights = [device for device in self.devices.values()
if device.type == DeviceType.LIGHT and device.room == room]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': True})
elif action_type == 'turn_off':
# Выключаем свет в комнате
lights = [device for device in self.devices.values()
if device.type == DeviceType.LIGHT and device.room == room]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': False})
async def execute_thermostat_action(self, action: Dict[str, Any]):
"""Выполнение действия с термостатом"""
action_type = action.get('action')
temperature = action.get('value')
if action_type == 'set_temperature':
# Устанавливаем температуру
thermostats = [device for device in self.devices.values()
if device.type == DeviceType.THERMOSTAT]
for thermostat in thermostats:
await self.send_device_command(thermostat, 'set_temperature', {'temperature': temperature})
async def execute_security_action(self, action: Dict[str, Any]):
"""Выполнение действия с системой безопасности"""
action_type = action.get('action')
if action_type == 'enable_alarm':
# Включаем сигнализацию
security_devices = [device for device in self.devices.values()
if device.type == DeviceType.SECURITY_SYSTEM]
for device in security_devices:
await self.send_device_command(device, 'enable_alarm', {})
elif action_type == 'disable_alarm':
# Выключаем сигнализацию
security_devices = [device for device in self.devices.values()
if device.type == DeviceType.SECURITY_SYSTEM]
for device in security_devices:
await self.send_device_command(device, 'disable_alarm', {})
async def execute_camera_action(self, action: Dict[str, Any]):
"""Выполнение действия с камерами"""
action_type = action.get('action')
if action_type == 'start_recording':
# Начинаем запись
cameras = [device for device in self.devices.values()
if device.type == DeviceType.CAMERA]
for camera in cameras:
await self.send_device_command(camera, 'start_recording', {})
elif action_type == 'stop_recording':
# Останавливаем запись
cameras = [device for device in self.devices.values()
if device.type == DeviceType.CAMERA]
for camera in cameras:
await self.send_device_command(camera, 'stop_recording', {})
async def create_custom_rule(self, rule_data: Dict[str, Any]) -> str:
"""Создание пользовательского правила"""
rule_id = f"custom_{len(self.rules) + 1}_{datetime.now().timestamp()}"
rule = AutomationRule(
id=rule_id,
name=rule_data.get('name', 'Custom Rule'),
trigger_conditions=rule_data.get('trigger_conditions', {}),
actions=rule_data.get('actions', []),
enabled=rule_data.get('enabled', True),
created_at=datetime.now()
)
self.rules[rule_id] = rule
return rule_id
```
### 3. 🔒 Система безопасности
```python
class SecurityManager:
def __init__(self):
self.security_devices = {}
self.alarm_status = False
self.motion_detectors = {}
self.door_sensors = {}
self.window_sensors = {}
self.security_cameras = {}
async def enable_security_system(self):
"""Включение системы безопасности"""
self.alarm_status = True
# Активируем все датчики движения
for sensor_id, sensor in self.motion_detectors.items():
await self.send_device_command(sensor, 'enable', {})
# Активируем датчики дверей и окон
for sensor_id, sensor in self.door_sensors.items():
await self.send_device_command(sensor, 'enable', {})
for sensor_id, sensor in self.window_sensors.items():
await self.send_device_command(sensor, 'enable', {})
# Включаем камеры безопасности
for camera_id, camera in self.security_cameras.items():
await self.send_device_command(camera, 'start_recording', {})
return True
async def disable_security_system(self):
"""Отключение системы безопасности"""
self.alarm_status = False
# Деактивируем все датчики
for sensor_id, sensor in self.motion_detectors.items():
await self.send_device_command(sensor, 'disable', {})
for sensor_id, sensor in self.door_sensors.items():
await self.send_device_command(sensor, 'disable', {})
for sensor_id, sensor in self.window_sensors.items():
await self.send_device_command(sensor, 'disable', {})
# Останавливаем запись камер
for camera_id, camera in self.security_cameras.items():
await self.send_device_command(camera, 'stop_recording', {})
return True
async def handle_security_alert(self, device: IoTDevice, alert_type: str):
"""Обработка сигнала безопасности"""
if not self.alarm_status:
return
alert_message = f"🚨 СИГНАЛ БЕЗОПАСНОСТИ!\n\n"
alert_message += f"📍 Устройство: {device.name}\n"
alert_message += f"🏠 Комната: {device.room}\n"
alert_message += f"⚠️ Тип сигнала: {alert_type}\n"
alert_message += f"⏰ Время: {datetime.now().strftime('%H:%M:%S')}\n"
# Отправляем уведомление пользователю
await self.send_security_notification(alert_message)
# Активируем дополнительные меры безопасности
await self.activate_security_measures(device, alert_type)
async def activate_security_measures(self, device: IoTDevice, alert_type: str):
"""Активация мер безопасности"""
if alert_type == 'motion_detected':
# Включаем все камеры
for camera_id, camera in self.security_cameras.items():
await self.send_device_command(camera, 'start_recording', {})
# Включаем освещение
lights = [d for d in self.devices.values() if d.type == DeviceType.LIGHT]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': True})
elif alert_type == 'door_opened':
# Проверяем, разрешено ли открытие двери
if not await self.is_authorized_access(device):
# Несанкционированный доступ
await self.trigger_alarm()
elif alert_type == 'window_opened':
# Окно открыто - потенциальная угроза
await self.trigger_alarm()
async def trigger_alarm(self):
"""Активация сигнализации"""
# Включаем звуковую сигнализацию
speakers = [device for device in self.devices.values() if device.type == DeviceType.SPEAKER]
for speaker in speakers:
await self.send_device_command(speaker, 'play_alarm', {})
# Включаем световую сигнализацию
lights = [device for device in self.devices.values() if device.type == DeviceType.LIGHT]
for light in lights:
await self.send_device_command(light, 'set_state', {'on': True, 'brightness': 100})
# Отправляем уведомления
await self.send_emergency_notifications()
async def send_security_notification(self, message: str):
"""Отправка уведомления о безопасности"""
# Здесь должна быть логика отправки уведомлений
# Упрощенная версия
print(f"Security Alert: {message}")
async def send_emergency_notifications(self):
"""Отправка экстренных уведомлений"""
emergency_message = """
🚨 ЭКСТРЕННАЯ СИТУАЦИЯ! 🚨
В вашем доме сработала сигнализация!
Возможно несанкционированное проникновение.
Рекомендуется:
• Немедленно связаться с полицией
• Проверить камеры безопасности
• Не входить в дом до прибытия полиции
Время: {datetime.now().strftime('%H:%M:%S')}
"""
# Отправляем уведомления всем контактам
await self.send_security_notification(emergency_message)
async def is_authorized_access(self, device: IoTDevice) -> bool:
"""Проверка авторизованного доступа"""
# Здесь должна быть логика проверки авторизации
# Упрощенная версия
return False
```
### 4. ⚡ Мониторинг энергии
```python
class EnergyMonitor:
def __init__(self):
self.energy_consumption = {}
self.power_sensors = {}
self.smart_plugs = {}
self.energy_reports = {}
async def monitor_energy_consumption(self):
"""Мониторинг потребления энергии"""
while True:
try:
# Собираем данные о потреблении энергии
for device_id, device in self.devices.items():
if device.type in [DeviceType.LIGHT, DeviceType.THERMOSTAT, DeviceType.SWITCH]:
power_consumption = await self.get_device_power_consumption(device)
if device_id not in self.energy_consumption:
self.energy_consumption[device_id] = []
self.energy_consumption[device_id].append({
'timestamp': datetime.now(),
'power': power_consumption,
'device_name': device.name,
'room': device.room
})
# Генерируем отчеты
await self.generate_energy_reports()
# Ждем перед следующим измерением
await asyncio.sleep(300) # 5 минут
except Exception as e:
logging.error(f"Energy monitoring error: {e}")
await asyncio.sleep(600)
async def get_device_power_consumption(self, device: IoTDevice) -> float:
"""Получение потребления энергии устройством"""
# Здесь должна быть логика получения данных о потреблении
# Упрощенная версия
base_consumption = {
DeviceType.LIGHT: 10.0, # Вт
DeviceType.THERMOSTAT: 50.0, # Вт
DeviceType.SWITCH: 5.0, # Вт
DeviceType.CAMERA: 15.0, # Вт
DeviceType.SPEAKER: 20.0 # Вт
}
consumption = base_consumption.get(device.type, 0.0)
# Корректируем в зависимости от состояния
if device.type == DeviceType.LIGHT:
brightness = device.current_state.get('brightness', 0)
consumption *= (brightness / 100.0)
return consumption
async def generate_energy_reports(self):
"""Генерация отчетов по энергии"""
current_hour = datetime.now().hour
# Генерируем отчет за час
hourly_report = await self.calculate_hourly_consumption()
# Генерируем отчет за день
daily_report = await self.calculate_daily_consumption()
# Генерируем отчет за месяц
monthly_report = await self.calculate_monthly_consumption()
self.energy_reports = {
'hourly': hourly_report,
'daily': daily_report,
'monthly': monthly_report
}
async def calculate_hourly_consumption(self) -> Dict[str, Any]:
"""Расчет потребления за час"""
current_hour = datetime.now().hour
hourly_consumption = 0.0
for device_id, consumption_data in self.energy_consumption.items():
# Берем данные за последний час
recent_data = [data for data in consumption_data
if data['timestamp'].hour == current_hour]
if recent_data:
avg_power = sum(data['power'] for data in recent_data) / len(recent_data)
hourly_consumption += avg_power
return {
'total_consumption': hourly_consumption,
'cost': hourly_consumption * 0.05, # 5 копеек за кВт⋅ч
'hour': current_hour
}
async def calculate_daily_consumption(self) -> Dict[str, Any]:
"""Расчет потребления за день"""
today = datetime.now().date()
daily_consumption = 0.0
for device_id, consumption_data in self.energy_consumption.items():
# Берем данные за сегодня
today_data = [data for data in consumption_data
if data['timestamp'].date() == today]
if today_data:
avg_power = sum(data['power'] for data in today_data) / len(today_data)
daily_consumption += avg_power
return {
'total_consumption': daily_consumption,
'cost': daily_consumption * 0.05,
'date': today
}
async def calculate_monthly_consumption(self) -> Dict[str, Any]:
"""Расчет потребления за месяц"""
current_month = datetime.now().month
monthly_consumption = 0.0
for device_id, consumption_data in self.energy_consumption.items():
# Берем данные за текущий месяц
month_data = [data for data in consumption_data
if data['timestamp'].month == current_month]
if month_data:
avg_power = sum(data['power'] for data in month_data) / len(month_data)
monthly_consumption += avg_power
return {
'total_consumption': monthly_consumption,
'cost': monthly_consumption * 0.05,
'month': current_month
}
async def get_energy_savings_tips(self) -> List[str]:
"""Получение советов по экономии энергии"""
tips = []
# Анализируем потребление и даем советы
if self.energy_reports:
daily_consumption = self.energy_reports.get('daily', {}).get('total_consumption', 0)
if daily_consumption > 1000: # Высокое потребление
tips.append("💡 Высокое потребление энергии! Проверьте настройки термостата.")
tips.append("🌡️ Установите температуру на 1-2 градуса ниже для экономии.")
if daily_consumption > 500:
tips.append("💡 Отключайте неиспользуемые устройства.")
tips.append("🔌 Используйте умные розетки для автоматического отключения.")
return tips
```
## Заключение
IoT-боты - это не просто управление устройствами, это создание интеллектуальной экосистемы умного дома. За 4 года работы я понял, что:
### 🎯 **Главные принципы IoT-ботов:**
1. **Централизованное управление** - все устройства в одном месте
2. **Автоматизация сценариев** - умные алгоритмы
3. **Безопасность** - защита от несанкционированного доступа
4. **Энергоэффективность** - мониторинг и оптимизация
### 💡 **Что работает лучше всего:**
- **Автоматические сценарии** - "Дома", "Спокойной ночи", "Экономия"
- **Голосовое управление** - удобство использования
- **Мониторинг безопасности** - защита дома
- **Аналитика энергии** - оптимизация потребления
### ⚠️ **Типичные ошибки:**
- Игнорирование безопасности устройств
- Отсутствие резервных систем
- Слабая автоматизация
- Недооценка важности мониторинга
### 🚀 **Советы для успеха:**
1. **Начните с простых устройств** - свет, термостат
2. **Создайте автоматические сценарии** - повысьте удобство
3. **Обеспечьте безопасность** - защитите от взлома
4. **Мониторьте энергию** - оптимизируйте потребление
Помните: лучший IoT-бот - это тот, который работает незаметно, но значительно повышает комфорт и безопасность дома!
---
*Готовы создать умный дом? Обращайтесь к нам за хостингом, консультациями и технической поддержкой!*
93 просмотров
0 лайков
0 комментариев
Комментарии (0)
Пока нет комментариев. Будьте первым!