别再被“AI培训”收割了：从零开始，让 OpenClaw 自己动手构建你的智能管家

#!/usr/bin/env python3"""Home Assistant Control CLI控制智能家居设备：灯光、窗帘、空调、加湿器等支持区域（Areas）查询和按区域控制"""
import osimport sysimport jsonimport argparseimport urllib.requestimport urllib.errorfrom typing import Optional, Dict, Any, List

def load_env(env_file: str = ".env") -> Dict[str, str]:    """手动解析.env文件"""    env_vars = {}
    # 尝试多个可能的 .env 位置    possible_paths = [        env_file,        os.path.expanduser("~/.openclaw/workspace/.env"),        os.path.join(os.path.dirname(__file__), ".env"),    ]
    for path in possible_paths:        if os.path.exists(path):            with open(path, 'r') as f:                for line in f:                    line = line.strip()                    if line and not line.startswith('#') and '=' in line:                        key, value = line.split('=', 1)                        env_vars[key.strip()] = value.strip()            break
    # 优先使用环境变量    if 'HA_URL' in os.environ:        env_vars['HA_URL'] = os.environ['HA_URL']    if 'HA_TOKEN' in os.environ:        env_vars['HA_TOKEN'] = os.environ['HA_TOKEN']
    return env_vars

class HomeAssistantClient:    """Home Assistant API 客户端"""
    def __init__(self, url: str, token: str):        self.url = url.rstrip('/')        self.token = token        self.headers = {            "Authorization": f"Bearer {token}",            "Content-Type": "application/json",        }        self._areas_cache = None        self._area_entities_cache = None
    def _request(self, endpoint: str, method: str = "GET", data: Optional[Dict] = None) -> Any:        """发送 HTTP 请求"""        url = f"{self.url}/api/{endpoint}"
        if data:            body = json.dumps(data).encode('utf-8')        else:            body = None
        req = urllib.request.Request(url, headers=self.headers, method=method, data=body)
        try:            with urllib.request.urlopen(req, timeout=10) as response:                if response.status == 200:                    return json.loads(response.read().decode())                else:                    return {"error": f"HTTP {response.status}"}        except urllib.error.URLError as e:            return {"error": str(e)}        except Exception as e:            return {"error": str(e)}
    def _render_template(self, template: str) -> Any:        """使用模板 API 渲染模板"""        data = json.dumps({"template": template}).encode('utf-8')        req = urllib.request.Request(            f"{self.url}/api/template",            headers=self.headers,            data=data,            method="POST"        )        try:            with urllib.request.urlopen(req, timeout=10) as response:                return response.read().decode()        except Exception as e:            return {"error": str(e)}
    def get_areas(self) -> Dict[str, str]:        """获取所有区域 {area_id: area_name}"""        if self._areas_cache:            return self._areas_cache
        template = '''{% for area_id in areas() %}{{ area_id }}|{{ area_name(area_id) }}{% endfor %}'''        result = self._render_template(template)
        if isinstance(result, dict) and 'error' in result:            return result
        areas = {}        for line in result.strip().split('\n'):            if '|' in line:                area_id, area_name = line.split('|', 1)                areas[area_id] = area_name
        self._areas_cache = areas        return areas
    def get_area_entities(self, area_id: str) -> List[str]:        """获取区域内的所有实体"""        template = f'{{{{ area_entities("{area_id}") }}}}'        result = self._render_template(template)
        if isinstance(result, dict) and 'error' in result:            return result
        try:            # 解析列表格式            entities = json.loads(result.replace("'", '"'))            return entities        except:            return []
    def get_area_entities_by_type(self, area_id: str, domain: str) -> List[str]:        """获取区域内特定类型的实体"""        entities = self.get_area_entities(area_id)        if isinstance(entities, dict) and 'error' in entities:            return entities        return [e for e in entities if e.startswith(f"{domain}.")]
    def find_area_by_name(self, name: str) -> Optional[str]:        """根据名称模糊匹配区域ID"""        areas = self.get_areas()        if isinstance(areas, dict) and 'error' in areas:            return None
        name_lower = name.lower()
        # 精确匹配        for area_id, area_name in areas.items():            if name_lower == area_name.lower() or name_lower == area_id.lower():                return area_id
        # 模糊匹配        for area_id, area_name in areas.items():            if name_lower in area_name.lower() or name_lower in area_id.lower():                return area_id
        # 中文别名映射        aliases = {            "主卧": "zhu_wo",            "次卧": "ci_wo",            "客厅": "ke_ting",            "餐厅": "can_ting",            "书房": "shu_fang",        }        if name_lower in aliases:            return aliases[name_lower]
        return None
    def get_states(self) -> List[Dict]:        """获取所有实体状态"""        return self._request("states")
    def get_state(self, entity_id: str) -> Dict:        """获取特定实体状态"""        return self._request(f"states/{entity_id}")
    def call_service(self, domain: str, service: str, entity_id: str,                      service_data: Optional[Dict] = None) -> Dict:        """调用服务"""        data = {"entity_id": entity_id}        if service_data:            data.update(service_data)        return self._request(f"services/{domain}/{service}", method="POST", data=data)
    def call_service_batch(self, domain: str, service: str, entity_ids: List[str],                          service_data: Optional[Dict] = None) -> List[Dict]:        """批量调用服务"""        results = []        for entity_id in entity_ids:            result = self.call_service(domain, service, entity_id, service_data)            results.append({"entity_id": entity_id, "result": result})        return results
    def list_entities(self, domain: Optional[str] = None) -> List[Dict]:        """列出实体，可选按域名过滤"""        states = self.get_states()        if isinstance(states, dict) and 'error' in states:            return states        if domain:            return [s for s in states if s.get('entity_id', '').startswith(f"{domain}.")]        return states
    def search_entities(self, keyword: str) -> List[Dict]:        """按关键词搜索实体"""        states = self.get_states()        if isinstance(states, dict) and 'error' in states:            return states        keyword_lower = keyword.lower()        results = []        for s in states:            entity_id = s.get('entity_id', '').lower()            friendly_name = s.get('attributes', {}).get('friendly_name', '').lower()            if keyword_lower in entity_id or keyword_lower in friendly_name:                results.append(s)        return results

def format_state(state: Dict, verbose: bool = False) -> str:    """格式化实体状态输出"""    entity_id = state.get('entity_id', 'unknown')    current_state = state.get('state', 'unknown')    friendly_name = state.get('attributes', {}).get('friendly_name', '')
    if verbose:        attrs = state.get('attributes', {})        attr_str = json.dumps(attrs, ensure_ascii=False, indent=2)        return f"{entity_id}\n  名称: {friendly_name}\n  状态: {current_state}\n  属性:\n{attr_str}"    else:        return f"{entity_id:<50} | {current_state:<20} | {friendly_name}"

def main():    parser = argparse.ArgumentParser(        description="Home Assistant 智能家居控制工具",        formatter_class=argparse.RawDescriptionHelpFormatter,        epilog="""示例:  %(prog)s status                          # 查看全屋状态  %(prog)s areas                           # 列出所有区域  %(prog)s area 主卧                        # 查看主卧设备  %(prog)s area 主卧 cover                  # 查看主卧窗帘  %(prog)s list light                      # 列出所有灯光  %(prog)s search 窗帘                      # 搜索窗帘相关设备  %(prog)s get sensor.xiaomi_p800_8453_temperature  # 查询特定实体  %(prog)s call light.living_room turn_on  # 打开灯光  %(prog)s call-area 主卧 cover open_cover # 打开主卧所有窗帘  %(prog)s call-area 主卧 cover set_cover_position '{"position": 50}'  # 设置窗帘位置        """    )
    subparsers = parser.add_subparsers(dest="command", help="命令")
    # status 命令    subparsers.add_parser("status", help="查看所有设备状态")
    # areas 命令    areas_parser = subparsers.add_parser("areas", help="列出所有区域")
    # area 命令    area_parser = subparsers.add_parser("area", help="查看区域设备")    area_parser.add_argument("name", help="区域名称（如：主卧、客厅）")    area_parser.add_argument("domain", nargs="?", help="可选：实体类型 (light, cover, climate)")
    # list 命令    list_parser = subparsers.add_parser("list", help="列出指定类型的实体")    list_parser.add_argument("domain", nargs="?", help="实体类型 (light, switch, cover, climate, humidifier, sensor)")
    # search 命令    search_parser = subparsers.add_parser("search", help="搜索实体")    search_parser.add_argument("keyword", help="搜索关键词")
    # get 命令    get_parser = subparsers.add_parser("get", help="查询实体状态")    get_parser.add_argument("entity_id", help="实体ID")    get_parser.add_argument("-v", "--verbose", action="store_true", help="详细输出")
    # call 命令    call_parser = subparsers.add_parser("call", help="调用服务")    call_parser.add_argument("entity_id", help="实体ID (如 light.living_room)")    call_parser.add_argument("service", help="服务名 (如 turn_on, turn_off)")    call_parser.add_argument("service_data", nargs="?", help="服务参数 JSON (如 '{\"brightness\": 200}')")
    # call-area 命令    call_area_parser = subparsers.add_parser("call-area", help="按区域调用服务")    call_area_parser.add_argument("area_name", help="区域名称（如：主卧、客厅）")    call_area_parser.add_argument("domain", help="实体类型 (light, cover, climate)")    call_area_parser.add_argument("service", help="服务名 (如 turn_on, open_cover)")    call_area_parser.add_argument("service_data", nargs="?", help="服务参数 JSON")
    args = parser.parse_args()
    # 加载配置    env = load_env()    ha_url = env.get("HA_URL")    ha_token = env.get("HA_TOKEN")
    if not ha_url or not ha_token:        print("错误: 未找到 HA_URL 或 HA_TOKEN，请检查 .env 文件或环境变量")        sys.exit(1)
    client = HomeAssistantClient(ha_url, ha_token)
    # 执行命令    if args.command == "status" or args.command is None:        states = client.get_states()        if isinstance(states, dict) and 'error' in states:            print(f"错误: {states['error']}")            sys.exit(1)
        print(f"{'Entity ID':<50} | {'State':<20} | {'Friendly Name'}")        print("-" * 100)        for state in sorted(states, key=lambda x: x.get('entity_id', '')):            print(format_state(state))
    elif args.command == "areas":        areas = client.get_areas()        if isinstance(areas, dict) and 'error' in areas:            print(f"错误: {areas['error']}")            sys.exit(1)
        print("区域列表:")        print("-" * 50)        for area_id, area_name in areas.items():            entities = client.get_area_entities(area_id)            entity_count = len(entities) if isinstance(entities, list) else 0            print(f"  {area_name:<20} ({area_id}) [{entity_count} 个设备]")
    elif args.command == "area":        area_id = client.find_area_by_name(args.name)        if not area_id:            print(f"未找到区域: {args.name}")            print("可用区域:")            for aid, aname in client.get_areas().items():                print(f"  - {aname} ({aid})")            sys.exit(1)
        area_name = client.get_areas().get(area_id, area_id)        entities = client.get_area_entities(area_id)
        if isinstance(entities, dict) and 'error' in entities:            print(f"错误: {entities['error']}")            sys.exit(1)
        if args.domain:            entities = [e for e in entities if e.startswith(f"{args.domain}.")]
        if not entities:            print(f"{area_name} 中没有找到 {args.domain or '设备'}")            sys.exit(0)
        print(f"{area_name} ({area_id}) - {args.domain or '所有设备'}:")        print(f"{'Entity ID':<50} | {'State':<20} | {'Friendly Name'}")        print("-" * 100)
        for entity_id in sorted(entities):            state = client.get_state(entity_id)            if isinstance(state, dict) and 'error' not in state:                print(format_state(state))
    elif args.command == "list":        states = client.list_entities(args.domain)        if isinstance(states, dict) and 'error' in states:            print(f"错误: {states['error']}")            sys.exit(1)
        if not states:            print(f"未找到类型为 '{args.domain}' 的实体")            sys.exit(0)
        print(f"找到 {len(states)} 个实体:")        print(f"{'Entity ID':<50} | {'State':<20} | {'Friendly Name'}")        print("-" * 100)        for state in sorted(states, key=lambda x: x.get('entity_id', '')):            print(format_state(state))
    elif args.command == "search":        results = client.search_entities(args.keyword)        if isinstance(results, dict) and 'error' in results:            print(f"错误: {results['error']}")            sys.exit(1)
        if not results:            print(f"未找到包含 '{args.keyword}' 的实体")            sys.exit(0)
        print(f"找到 {len(results)} 个匹配实体:")        print(f"{'Entity ID':<50} | {'State':<20} | {'Friendly Name'}")        print("-" * 100)        for state in sorted(results, key=lambda x: x.get('entity_id', '')):            print(format_state(state))
    elif args.command == "get":        state = client.get_state(args.entity_id)        if isinstance(state, dict) and 'error' in state:            print(f"错误: {state['error']}")            sys.exit(1)
        print(format_state(state, verbose=args.verbose))
    elif args.command == "call":        # 解析 entity_id 获取 domain        parts = args.entity_id.split('.')        if len(parts) < 2:            print(f"错误: 无效的实体ID格式 '{args.entity_id}'，应为 domain.entity_name")            sys.exit(1)
        domain = parts[0]
        # 解析服务数据        service_data = None        if args.service_data:            try:                service_data = json.loads(args.service_data)            except json.JSONDecodeError:                print(f"错误: 无效的 JSON 格式 '{args.service_data}'")                sys.exit(1)
        result = client.call_service(domain, args.service, args.entity_id, service_data)
        if isinstance(result, dict) and 'error' in result:            print(f"错误: {result['error']}")            sys.exit(1)
        print(f"✓ 已执行: {domain}.{args.service}({args.entity_id})")        if service_data:            print(f"  参数: {json.dumps(service_data, ensure_ascii=False)}")
    elif args.command == "call-area":        area_id = client.find_area_by_name(args.area_name)        if not area_id:            print(f"未找到区域: {args.area_name}")            sys.exit(1)
        entities = client.get_area_entities_by_type(area_id, args.domain)        if isinstance(entities, dict) and 'error' in entities:            print(f"错误: {entities['error']}")            sys.exit(1)
        if not entities:            print(f"{args.area_name} 中没有 {args.domain} 类型的设备")            sys.exit(0)
        # 解析服务数据        service_data = None        if args.service_data:            try:                service_data = json.loads(args.service_data)            except json.JSONDecodeError:                print(f"错误: 无效的 JSON 格式 '{args.service_data}'")                sys.exit(1)
        print(f"正在 {args.area_name} 执行 {args.domain}.{args.service}...")        print(f"设备: {', '.join(entities)}")
        results = client.call_service_batch(args.domain, args.service, entities, service_data)
        success_count = 0        for r in results:            if isinstance(r.get('result'), dict) and 'error' in r.get('result', {}):                print(f"  ✗ {r['entity_id']}: {r['result']['error']}")            else:                print(f"  ✓ {r['entity_id']}")                success_count += 1
        print(f"\n完成: {success_count}/{len(entities)} 成功")

if __name__ == "__main__":    main()