711 lines
23 KiB
Python
711 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Home Assistant 연동 — 엔티티 스위치 JSON + light/climate/TV/washer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
HOOKS = Path(__file__).resolve().parent
|
|
BASE_DIR = HOOKS.parent
|
|
ENV_PATH = HOOKS / "ha.env"
|
|
ALIAS_PATH = HOOKS / "aliases.json"
|
|
ENTITIES_PATH = Path(
|
|
os.environ.get("JARVIS_HA_ENTITIES", str(BASE_DIR / "jarvis_ha_entities.json"))
|
|
)
|
|
|
|
_entities_mtime: float = -1.0
|
|
_entities_cache: dict[str, dict] = {}
|
|
|
|
DEFAULT_ON_DOMAINS = frozenset({"light", "climate"})
|
|
DEFAULT_OFF_DOMAINS = frozenset(
|
|
{
|
|
"lock",
|
|
"alarm_control_panel",
|
|
"cover",
|
|
"person",
|
|
"device_tracker",
|
|
"camera",
|
|
"image",
|
|
"update",
|
|
"automation",
|
|
"script",
|
|
"scene",
|
|
"button",
|
|
"event",
|
|
"notify",
|
|
"tts",
|
|
"stt",
|
|
"conversation",
|
|
"ai_task",
|
|
"todo",
|
|
"calendar",
|
|
"zone",
|
|
}
|
|
)
|
|
TV_NAME_KEYS = ("tv", "webos", "티비", "텔레비전", "television", "엘지티비")
|
|
|
|
|
|
def load_env() -> dict[str, str]:
|
|
env: dict[str, str] = {}
|
|
if not ENV_PATH.is_file():
|
|
raise SystemExit(f"ha.env 없음: {ENV_PATH}")
|
|
for line in ENV_PATH.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
env[k.strip()] = v.strip()
|
|
if not env.get("HA_TOKEN"):
|
|
raise SystemExit("HA_TOKEN 비어 있음 (ha.env 저장 확인)")
|
|
env["HA_URL"] = env.get("HA_URL", "http://192.168.0.148:8123").rstrip("/")
|
|
return env
|
|
|
|
|
|
def ha_request(method: str, path: str, data: dict | None = None) -> object:
|
|
env = load_env()
|
|
body = None if data is None else json.dumps(data).encode()
|
|
req = urllib.request.Request(
|
|
env["HA_URL"] + path,
|
|
data=body,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"Bearer {env['HA_TOKEN']}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
raw = resp.read()
|
|
if not raw:
|
|
return None
|
|
return json.loads(raw.decode())
|
|
except urllib.error.HTTPError as e:
|
|
raise RuntimeError(f"HA HTTP {e.code}: {e.read()[:200]!r}") from e
|
|
|
|
|
|
def get_states() -> list[dict]:
|
|
data = ha_request("GET", "/api/states")
|
|
if not isinstance(data, list):
|
|
raise RuntimeError(f"states 이상: {data!r}"[:200])
|
|
return data
|
|
|
|
|
|
def _compact(s: str) -> str:
|
|
return re.sub(r"\s+", "", s)
|
|
|
|
|
|
def _looks_like_tv(entity_id: str, friendly_name: str) -> bool:
|
|
blob = f"{entity_id} {friendly_name}".lower()
|
|
return any(k in blob for k in TV_NAME_KEYS)
|
|
|
|
|
|
def _default_enabled(entity_id: str, friendly_name: str) -> bool:
|
|
domain = entity_id.split(".", 1)[0]
|
|
if domain in DEFAULT_ON_DOMAINS:
|
|
return True
|
|
if entity_id.startswith(("sensor.setaggi", "binary_sensor.setaggi")):
|
|
return True
|
|
if "setaggi" in entity_id:
|
|
return True
|
|
if domain == "media_player":
|
|
return _looks_like_tv(entity_id, friendly_name)
|
|
if domain in DEFAULT_OFF_DOMAINS:
|
|
return False
|
|
return False
|
|
|
|
|
|
def load_aliases() -> dict[str, str]:
|
|
if not ALIAS_PATH.is_file():
|
|
return {}
|
|
try:
|
|
data = json.loads(ALIAS_PATH.read_text(encoding="utf-8"))
|
|
return {
|
|
str(k): str(v)
|
|
for k, v in (data or {}).items()
|
|
if not str(k).startswith("_")
|
|
}
|
|
except (OSError, json.JSONDecodeError, TypeError):
|
|
return {}
|
|
|
|
|
|
def load_entities_map(*, force: bool = False) -> dict[str, dict]:
|
|
"""jarvis_ha_entities.json — mtime 변경 시 재로드."""
|
|
global _entities_mtime, _entities_cache
|
|
if not ENTITIES_PATH.is_file():
|
|
_entities_mtime = -1.0
|
|
_entities_cache = {}
|
|
return {}
|
|
try:
|
|
mtime = ENTITIES_PATH.stat().st_mtime
|
|
except OSError:
|
|
return _entities_cache
|
|
if not force and mtime == _entities_mtime and _entities_cache:
|
|
return _entities_cache
|
|
try:
|
|
data = json.loads(ENTITIES_PATH.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
print(f"[ha_entities] 읽기 실패: {e}")
|
|
return _entities_cache
|
|
raw = data.get("entities") if isinstance(data, dict) else None
|
|
if not isinstance(raw, dict):
|
|
return _entities_cache
|
|
cleaned: dict[str, dict] = {}
|
|
for eid, meta in raw.items():
|
|
if not isinstance(meta, dict):
|
|
continue
|
|
cleaned[str(eid)] = {
|
|
"name": str(meta.get("name") or eid),
|
|
"enabled": bool(meta.get("enabled", False)),
|
|
}
|
|
_entities_cache = cleaned
|
|
_entities_mtime = mtime
|
|
return _entities_cache
|
|
|
|
|
|
def is_enabled(entity_id: str) -> bool:
|
|
ents = load_entities_map()
|
|
if not ents:
|
|
domain = entity_id.split(".", 1)[0]
|
|
return domain in DEFAULT_ON_DOMAINS or "setaggi" in entity_id
|
|
meta = ents.get(entity_id)
|
|
return bool(meta and meta.get("enabled"))
|
|
|
|
|
|
def assert_enabled(entity_id: str) -> str | None:
|
|
"""허용이면 None, 아니면 거부 메시지."""
|
|
ents = load_entities_map()
|
|
if not ents:
|
|
return None
|
|
if entity_id not in ents:
|
|
return f"{entity_id} 는 엔티티 JSON에 없습니다. sync 후 enabled를 확인하세요."
|
|
if not ents[entity_id].get("enabled"):
|
|
name = ents[entity_id].get("name") or entity_id
|
|
return f"{name}({entity_id}) 는 스위치가 꺼져 있어 조작할 수 없습니다."
|
|
return None
|
|
|
|
|
|
def sync_entities_file(*, path: Path | None = None) -> Path:
|
|
"""HA 전 엔티티를 JSON에 넣는다. 기존 enabled는 유지, 신규만 기본값."""
|
|
out = path or ENTITIES_PATH
|
|
states = get_states()
|
|
existing: dict[str, dict] = {}
|
|
if out.is_file():
|
|
try:
|
|
prev = json.loads(out.read_text(encoding="utf-8"))
|
|
raw = prev.get("entities") if isinstance(prev, dict) else {}
|
|
if isinstance(raw, dict):
|
|
for eid, meta in raw.items():
|
|
if isinstance(meta, dict):
|
|
existing[str(eid)] = {
|
|
"name": str(meta.get("name") or eid),
|
|
"enabled": bool(meta.get("enabled", False)),
|
|
}
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
|
|
entities: dict[str, dict] = {}
|
|
for st in states:
|
|
eid = str(st.get("entity_id") or "")
|
|
if not eid or "." not in eid:
|
|
continue
|
|
fn = str((st.get("attributes") or {}).get("friendly_name") or eid)
|
|
if eid in existing:
|
|
entities[eid] = {
|
|
"name": existing[eid].get("name") or fn,
|
|
"enabled": bool(existing[eid].get("enabled")),
|
|
}
|
|
else:
|
|
entities[eid] = {
|
|
"name": fn,
|
|
"enabled": _default_enabled(eid, fn),
|
|
}
|
|
|
|
for eid, meta in existing.items():
|
|
if eid not in entities:
|
|
entities[eid] = meta
|
|
|
|
payload = {
|
|
"_help": (
|
|
"enabled true만 자비스가 읽고 조작. 저장 즉시 반영(재시작 불필요). "
|
|
"동기화: python3 jarvis_hooks/ha_api.py sync"
|
|
),
|
|
"entities": dict(sorted(entities.items())),
|
|
}
|
|
out.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
load_entities_map(force=True)
|
|
enabled_n = sum(1 for m in entities.values() if m.get("enabled"))
|
|
print(f"[ha_entities] {out} — 전체 {len(entities)} / enabled {enabled_n}")
|
|
return out
|
|
|
|
|
|
def _display_name(entity_id: str, friendly_name: str) -> str:
|
|
ents = load_entities_map()
|
|
meta = ents.get(entity_id)
|
|
if meta and meta.get("name"):
|
|
return str(meta["name"])
|
|
return friendly_name or entity_id
|
|
|
|
|
|
def _apply_aliases(utterance: str) -> str:
|
|
c = _compact(utterance)
|
|
aliases = load_aliases()
|
|
for alias, target in sorted(aliases.items(), key=lambda x: -len(x[0])):
|
|
if _compact(alias) and _compact(alias) in c:
|
|
return target
|
|
return utterance
|
|
|
|
|
|
def _match_by_name(
|
|
utterance: str,
|
|
candidates: list[tuple[str, str, str]],
|
|
suffixes: tuple[str, ...] = (),
|
|
) -> tuple[str, str] | None:
|
|
"""candidates: (display_name, entity_id, state) — 긴 이름 우선."""
|
|
text = _apply_aliases(utterance)
|
|
c = _compact(text)
|
|
scored: list[tuple[int, str, str]] = []
|
|
for fn, eid, _ in candidates:
|
|
cf = _compact(fn)
|
|
if not cf:
|
|
continue
|
|
if cf in c or (len(c) >= 2 and c in cf):
|
|
scored.append((len(cf), fn, eid))
|
|
continue
|
|
for suffix in suffixes:
|
|
if cf + suffix in c or cf in c.replace(suffix, ""):
|
|
scored.append((len(cf), fn, eid))
|
|
break
|
|
ce = _compact(eid.split(".", 1)[-1])
|
|
if ce and len(ce) >= 3 and ce in c:
|
|
scored.append((len(ce), fn, eid))
|
|
if scored:
|
|
scored.sort(reverse=True)
|
|
return scored[0][1], scored[0][2]
|
|
return None
|
|
|
|
|
|
def list_lights(states: list[dict] | None = None) -> list[tuple[str, str, str]]:
|
|
"""(friendly_name, entity_id, state) — enabled만."""
|
|
states = states or get_states()
|
|
out = []
|
|
for st in states:
|
|
eid = st.get("entity_id") or ""
|
|
if not eid.startswith("light.") or not is_enabled(eid):
|
|
continue
|
|
fn = (st.get("attributes") or {}).get("friendly_name") or eid
|
|
out.append((_display_name(eid, fn), eid, st.get("state") or "unknown"))
|
|
return out
|
|
|
|
|
|
def resolve_light(
|
|
utterance: str, states: list[dict] | None = None
|
|
) -> tuple[str, str] | None:
|
|
states = states or get_states()
|
|
lights = list_lights(states)
|
|
hit = _match_by_name(
|
|
utterance, lights, suffixes=("불", "등", "조명", "라이트")
|
|
)
|
|
if hit:
|
|
return hit
|
|
c = _compact(_apply_aliases(utterance))
|
|
prefers: list[str] = []
|
|
for room in ("거실", "아기방", "안방", "주방", "화장실"):
|
|
if room in c:
|
|
prefers.append(room)
|
|
prefers.extend(("메인등", "거실", "아기방"))
|
|
if any(
|
|
k in c
|
|
for k in ("불켜", "불꺼", "라이트온", "라이트오프", "조명켜", "조명꺼")
|
|
) or prefers:
|
|
seen: set[str] = set()
|
|
for prefer in prefers:
|
|
if prefer in seen:
|
|
continue
|
|
seen.add(prefer)
|
|
for fn, eid, _ in lights:
|
|
if prefer in fn:
|
|
return fn, eid
|
|
if lights:
|
|
return lights[0][0], lights[0][1]
|
|
return None
|
|
|
|
|
|
def light_service(entity_id: str, turn_on: bool) -> None:
|
|
err = assert_enabled(entity_id)
|
|
if err:
|
|
raise RuntimeError(err)
|
|
svc = "turn_on" if turn_on else "turn_off"
|
|
ha_request("POST", f"/api/services/light/{svc}", {"entity_id": entity_id})
|
|
|
|
|
|
def list_climates(
|
|
states: list[dict] | None = None,
|
|
) -> list[tuple[str, str, str, float | None]]:
|
|
"""(name, entity_id, state, temperature)."""
|
|
states = states or get_states()
|
|
out = []
|
|
for st in states:
|
|
eid = st.get("entity_id") or ""
|
|
if not eid.startswith("climate.") or not is_enabled(eid):
|
|
continue
|
|
attrs = st.get("attributes") or {}
|
|
fn = attrs.get("friendly_name") or eid
|
|
temp = attrs.get("temperature")
|
|
try:
|
|
temp_f = float(temp) if temp is not None else None
|
|
except (TypeError, ValueError):
|
|
temp_f = None
|
|
out.append(
|
|
(_display_name(eid, fn), eid, st.get("state") or "unknown", temp_f)
|
|
)
|
|
return out
|
|
|
|
|
|
def resolve_climate(
|
|
utterance: str, states: list[dict] | None = None
|
|
) -> tuple[str, str] | None:
|
|
climates = [(n, e, s) for n, e, s, _ in list_climates(states)]
|
|
hit = _match_by_name(
|
|
utterance, climates, suffixes=("에어컨", "냉방", "공조")
|
|
)
|
|
if hit:
|
|
return hit
|
|
c = _compact(_apply_aliases(utterance))
|
|
if any(k in c for k in ("에어컨", "더워", "추워", "냉방", "난방")):
|
|
for prefer in ("스탠드", "거실", "이문동"):
|
|
for fn, eid, _ in climates:
|
|
if prefer in fn:
|
|
return fn, eid
|
|
if climates:
|
|
return climates[0][0], climates[0][1]
|
|
return None
|
|
|
|
|
|
def climate_service(
|
|
entity_id: str,
|
|
action: str,
|
|
temperature: int | float | None = None,
|
|
) -> str:
|
|
err = assert_enabled(entity_id)
|
|
if err:
|
|
return err
|
|
action = (action or "").strip().lower()
|
|
if action in ("on", "turn_on", "켜", "켜줘"):
|
|
ha_request(
|
|
"POST",
|
|
"/api/services/climate/turn_on",
|
|
{"entity_id": entity_id},
|
|
)
|
|
temp = 24 if temperature is None else int(temperature)
|
|
ha_request(
|
|
"POST",
|
|
"/api/services/climate/set_temperature",
|
|
{"entity_id": entity_id, "temperature": temp},
|
|
)
|
|
return f"{entity_id} 를 {temp}도로 켰습니다."
|
|
if action in ("off", "turn_off", "꺼", "꺼줘"):
|
|
ha_request(
|
|
"POST",
|
|
"/api/services/climate/turn_off",
|
|
{"entity_id": entity_id},
|
|
)
|
|
return f"{entity_id} 를 껐습니다."
|
|
if action in ("set", "set_temperature", "온도"):
|
|
if temperature is None:
|
|
return "온도 숫자가 필요합니다."
|
|
temp = int(temperature)
|
|
ha_request(
|
|
"POST",
|
|
"/api/services/climate/set_temperature",
|
|
{"entity_id": entity_id, "temperature": temp},
|
|
)
|
|
return f"{entity_id} 온도를 {temp}도로 맞췄습니다."
|
|
return f"알 수 없는 climate action: {action}"
|
|
|
|
|
|
def list_tvs(states: list[dict] | None = None) -> list[tuple[str, str, str]]:
|
|
states = states or get_states()
|
|
out = []
|
|
for st in states:
|
|
eid = st.get("entity_id") or ""
|
|
if not eid.startswith("media_player.") or not is_enabled(eid):
|
|
continue
|
|
fn = (st.get("attributes") or {}).get("friendly_name") or eid
|
|
out.append((_display_name(eid, fn), eid, st.get("state") or "unknown"))
|
|
return out
|
|
|
|
|
|
def resolve_tv(
|
|
utterance: str = "", states: list[dict] | None = None
|
|
) -> tuple[str, str] | None:
|
|
tvs = list_tvs(states)
|
|
if not tvs:
|
|
return None
|
|
text = utterance or "티비"
|
|
hit = _match_by_name(
|
|
text, tvs, suffixes=("티비", "TV", "텔레비전", "티브이")
|
|
)
|
|
if hit:
|
|
return hit
|
|
for fn, eid, _ in tvs:
|
|
if _looks_like_tv(eid, fn):
|
|
return fn, eid
|
|
return tvs[0][0], tvs[0][1]
|
|
|
|
|
|
def tv_service(entity_id: str, turn_on: bool) -> None:
|
|
err = assert_enabled(entity_id)
|
|
if err:
|
|
raise RuntimeError(err)
|
|
svc = "turn_on" if turn_on else "turn_off"
|
|
ha_request(
|
|
"POST", f"/api/services/media_player/{svc}", {"entity_id": entity_id}
|
|
)
|
|
|
|
|
|
def home_snapshot(states: list[dict] | None = None) -> str:
|
|
"""enabled 기기만 짧게. 제미나이 컨텍스트용."""
|
|
states = states or get_states()
|
|
by_id = {st.get("entity_id"): st for st in states}
|
|
lines: list[str] = []
|
|
|
|
for fn, eid, st in list_lights(states):
|
|
lines.append(f"light | {fn} | {eid} | {st}")
|
|
for fn, eid, st, temp in list_climates(states):
|
|
extra = f", set={temp}" if temp is not None else ""
|
|
cur = (by_id.get(eid) or {}).get("attributes", {}).get(
|
|
"current_temperature"
|
|
)
|
|
if cur is not None:
|
|
extra += f", now={cur}"
|
|
lines.append(f"climate | {fn} | {eid} | {st}{extra}")
|
|
for fn, eid, st in list_tvs(states):
|
|
lines.append(f"tv/media | {fn} | {eid} | {st}")
|
|
|
|
washer_bits = []
|
|
for eid, st in by_id.items():
|
|
if not eid or "setaggi" not in eid:
|
|
continue
|
|
if not eid.startswith(("sensor.", "binary_sensor.")):
|
|
continue
|
|
if not is_enabled(eid):
|
|
continue
|
|
washer_bits.append(f"{eid}={st.get('state')}")
|
|
if washer_bits:
|
|
lines.append("washer | " + ", ".join(washer_bits[:12]))
|
|
|
|
if not lines:
|
|
return "enabled 기기가 없습니다. jarvis_ha_entities.json 을 확인하세요."
|
|
return "\n".join(lines)
|
|
|
|
|
|
def washer_status(states: list[dict] | None = None) -> str:
|
|
states = states or get_states()
|
|
by_id = {st.get("entity_id"): st for st in states}
|
|
|
|
remain = None
|
|
power = None
|
|
run_state = None
|
|
pre = None
|
|
completed = None
|
|
|
|
if "sensor.setaggi" in by_id:
|
|
st = by_id["sensor.setaggi"]
|
|
attrs = st.get("attributes") or {}
|
|
power = st.get("state")
|
|
remain = attrs.get("remain_time")
|
|
run_state = attrs.get("run_state")
|
|
completed = attrs.get("run_completed")
|
|
|
|
if "sensor.setaggi_remaining_time" in by_id:
|
|
remain = remain or by_id["sensor.setaggi_remaining_time"].get("state")
|
|
if "sensor.setaggi_remaining_time_2" in by_id:
|
|
r2 = by_id["sensor.setaggi_remaining_time_2"].get("state")
|
|
if r2 and r2 not in ("unknown", "unavailable", "none"):
|
|
remain = remain or r2
|
|
if "sensor.setaggi_current_status" in by_id:
|
|
power = power or by_id["sensor.setaggi_current_status"].get("state")
|
|
if "sensor.setaggi_run_state" in by_id:
|
|
run_state = run_state or by_id["sensor.setaggi_run_state"].get("state")
|
|
if "sensor.setaggi_pre_state" in by_id:
|
|
pre = by_id["sensor.setaggi_pre_state"].get("state")
|
|
if "binary_sensor.setaggi_run_completed" in by_id:
|
|
completed = completed or by_id["binary_sensor.setaggi_run_completed"].get(
|
|
"state"
|
|
)
|
|
|
|
def fmt_remain(v) -> str | None:
|
|
if v is None:
|
|
return None
|
|
s = str(v).strip()
|
|
if s in ("", "unknown", "unavailable", "none", "-", "0:00:00", "00:00:00"):
|
|
return None
|
|
m = re.match(r"^(\d+):(\d+):(\d+)$", s)
|
|
if m:
|
|
h, mi, sec = map(int, m.groups())
|
|
total = h * 60 + mi + (1 if sec else 0)
|
|
if total <= 0:
|
|
return None
|
|
if h > 0:
|
|
return f"{h}시간 {mi}분"
|
|
return f"{mi}분"
|
|
return s
|
|
|
|
remain_s = fmt_remain(remain)
|
|
off_like = str(power).lower() in (
|
|
"off",
|
|
"power_off",
|
|
"unavailable",
|
|
"unknown",
|
|
"",
|
|
)
|
|
|
|
if remain_s:
|
|
extra = (
|
|
f", 상태 {run_state}"
|
|
if run_state and run_state not in ("-", "unknown")
|
|
else ""
|
|
)
|
|
return f"세탁기 남은 시간은 약 {remain_s}입니다{extra}."
|
|
|
|
if completed in ("on", "True", True) or (pre and "완료" in str(pre)):
|
|
return (
|
|
f"세탁기는 지금 꺼져 있고, 세탁은 끝난 상태입니다. "
|
|
f"{('이전 상태: ' + str(pre)) if pre else ''}"
|
|
).strip()
|
|
|
|
if off_like:
|
|
return "세탁기는 지금 전원 꺼짐 상태라 남은 시간이 없습니다."
|
|
|
|
return f"세탁기 상태: 전원 {power}, 남은 시간 정보 없음."
|
|
|
|
|
|
def tool_get_home_snapshot() -> str:
|
|
"""지금 집안에서 자비스가 조작 가능한(enabled) 기기 목록과 상태를 조회한다."""
|
|
try:
|
|
return home_snapshot()
|
|
except Exception as e:
|
|
return f"스냅샷 실패: {e}"
|
|
|
|
|
|
def tool_control_light(name: str, action: str) -> str:
|
|
"""조명을 켠다/끈다. name은 방/조명 이름(예: 거실, 메인등, 아기방). action은 on 또는 off."""
|
|
try:
|
|
turn_on = str(action).strip().lower() in (
|
|
"on",
|
|
"turn_on",
|
|
"켜",
|
|
"켜줘",
|
|
"켜라",
|
|
)
|
|
hit = resolve_light(name or "")
|
|
if not hit:
|
|
return f"조명을 찾지 못했습니다: {name}"
|
|
fn, eid = hit
|
|
light_service(eid, turn_on)
|
|
return f"{fn}을 {'켰습니다' if turn_on else '껐습니다'}."
|
|
except Exception as e:
|
|
return f"조명 제어 실패: {e}"
|
|
|
|
|
|
def tool_control_climate(
|
|
name: str = "",
|
|
action: str = "turn_on",
|
|
temperature: int = 24,
|
|
) -> str:
|
|
"""에어컨/난방을 켠다/끈다/온도를 맞춘다. action: turn_on, turn_off, set. temperature 기본 24."""
|
|
try:
|
|
hit = resolve_climate(name or "에어컨")
|
|
if not hit:
|
|
return f"에어컨을 찾지 못했습니다: {name or '(기본)'}"
|
|
fn, eid = hit
|
|
msg = climate_service(eid, action, temperature)
|
|
return f"{fn}: {msg}"
|
|
except Exception as e:
|
|
return f"에어컨 제어 실패: {e}"
|
|
|
|
|
|
def tool_control_tv(name: str = "", action: str = "turn_on") -> str:
|
|
"""티비를 켠다/끈다. name 생략 시 거실 LG 티비. action: on/off."""
|
|
try:
|
|
turn_on = str(action).strip().lower() in (
|
|
"on",
|
|
"turn_on",
|
|
"켜",
|
|
"켜줘",
|
|
"켜라",
|
|
)
|
|
hit = resolve_tv(name or "티비")
|
|
if not hit:
|
|
return (
|
|
"티비(media_player)를 찾지 못했습니다. "
|
|
"entities JSON에서 enabled를 확인하세요."
|
|
)
|
|
fn, eid = hit
|
|
tv_service(eid, turn_on)
|
|
return f"{fn}을 {'켰습니다' if turn_on else '껐습니다'}."
|
|
except Exception as e:
|
|
return f"티비 제어 실패: {e}"
|
|
|
|
|
|
def tool_get_washer_status() -> str:
|
|
"""세탁기 남은 시간·전원·완료 여부를 조회한다."""
|
|
try:
|
|
return washer_status()
|
|
except Exception as e:
|
|
return f"세탁기 조회 실패: {e}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "washer"
|
|
if cmd == "sync":
|
|
sync_entities_file()
|
|
elif cmd == "snapshot":
|
|
if not ENTITIES_PATH.is_file():
|
|
sync_entities_file()
|
|
print(home_snapshot())
|
|
elif cmd == "washer":
|
|
print(washer_status())
|
|
elif cmd == "lights":
|
|
for fn, eid, st in list_lights():
|
|
print(f"{fn}\t{eid}\t{st}")
|
|
elif cmd == "tvs":
|
|
for fn, eid, st in list_tvs():
|
|
print(f"{fn}\t{eid}\t{st}")
|
|
elif cmd == "climates":
|
|
for fn, eid, st, temp in list_climates():
|
|
print(f"{fn}\t{eid}\t{st}\t{temp}")
|
|
elif cmd in ("on", "off") and len(sys.argv) > 2:
|
|
name = " ".join(sys.argv[2:])
|
|
hit = resolve_light(name)
|
|
if not hit:
|
|
print("FAIL no light", name)
|
|
sys.exit(1)
|
|
fn, eid = hit
|
|
light_service(eid, cmd == "on")
|
|
print(f"OK {cmd} {fn} {eid}")
|
|
elif cmd in ("tv_on", "tv_off"):
|
|
name = " ".join(sys.argv[2:]) if len(sys.argv) > 2 else "티비"
|
|
hit = resolve_tv(name)
|
|
if not hit:
|
|
print("FAIL no tv", name)
|
|
sys.exit(1)
|
|
fn, eid = hit
|
|
tv_service(eid, cmd == "tv_on")
|
|
print(f"OK {cmd} {fn} {eid}")
|
|
else:
|
|
print(
|
|
"usage: ha_api.py sync|snapshot|washer|lights|tvs|climates|"
|
|
"on <이름>|off <이름>|tv_on [이름]|tv_off [이름]"
|
|
)
|