use-wifi-to-control-mijia-camera

use-wifi-to-control-mijia-camera.py

import subprocess
import requests
import logging
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Home Assistant 配置
HA_URL = ""  # 替换为你的 Home Assistant IP 和端口
HA_TOKEN = ""  # 替换为你的长期访问令牌
DEVICE = ""  # 替换为你的实体标识符

# 要检查的 MAC 地址列表
MAC_ADDRESSES = [
    "MAC1",
    "MAC2",
]

# Home Assistant API 请求头
HEADERS = {
    "Authorization": f"Bearer {HA_TOKEN}",
    "Content-Type": "application/json"
}

def run_arp_command():
    """执行 arp -a 命令并返回输出"""
    try:
        result = subprocess.run(["arp-scan", "--localnet"], capture_output=True, text=True, check=True)
        return result.stdout.lower()  # 转换为小写以便匹配
    except subprocess.CalledProcessError as e:
        logger.info(f"Error running arp command: {e}")
        return ""

def get_switch_state(entity_id):
    """获取指定 switch 的状态"""
    url = f"{HA_URL}/api/states/{entity_id}"
    try:
        response = requests.get(url, headers=HEADERS)
        if response.status_code == 200:
            state = response.json().get("state")
            logger.info(f"Current state of {entity_id}: {state}")
            return state
        else:
            logger.info(f"Failed to get state for {entity_id}: {response.status_code} - {response.text}")
            return None
    except requests.RequestException as e:
        logger.info(f"Error getting switch state: {e}")
        return None

def call_ha_api(entity_id, service):
    """调用 Home Assistant API 执行服务（turn_on 或 turn_off）"""
    url = f"{HA_URL}/api/services/switch/{service}"
    data = {"entity_id": entity_id}
    try:
        response = requests.post(url, headers=HEADERS, json=data)
        if response.status_code in [200, 201]:
            logger.info(f"Successfully called {service} for {entity_id}")
        else:
            logger.info(f"Failed to call {service} for {entity_id}: {response.status_code} - {response.text}")
    except requests.RequestException as e:
        logger.info(f"Error calling Home Assistant API: {e}")

def main():
    # 获取 ARP 表
    arp_output = run_arp_command()

    state = get_switch_state(DEVICE)
    
    # 检查是否包含任一 MAC 地址
    mac_found = any(mac.lower() in arp_output for mac in MAC_ADDRESSES)
    if mac_found and state == "on":
        logger.info("Found matching MAC address. Turning off.")
        call_ha_api(DEVICE, "turn_off")
    elif not mac_found and state == "off":
        logger.info("No matching MAC address found. Turning on.")
        call_ha_api(DEVICE, "turn_on")
    else:
        logger.info("do noting")

if __name__ == "__main__":
    main()