bernd
d69e07f00d
The lock in this WG is a Homematic IP door lock (keypad + alarm), not Nuki/Tuya as originally specced — confirmed with the user. Homematic IP has no official self-service partner API for Access-Point-only setups (no local CCU), so this uses the actively-maintained open-source `homematicip` Python library (implements the same reverse-engineered cloud protocol used by the official app) via two scripts rather than reimplementing the HTTP/HMAC handshake from scratch — lower risk of subtly wrong protocol details. - scripts/hmip_register.py: non-interactive pairing (connection request -> wait for the physical blue-button press on the Access Point -> auth token). Writes progress to a status file so the API can poll it instead of blocking a request for up to ~6 minutes. - scripts/hmip_control.py: list-devices / lock / unlock via HMIP_ACCESS_POINT + HMIP_AUTH_TOKEN env vars (not CLI args, so they don't leak into process listings). - routes/smartLock.ts: POST /smartlock/pairing/start + GET .../status (LANDLORD/ADMIN, one-time setup), GET /smartlock/devices (LANDLORD/ADMIN), POST /smartlock/devices/:id/:lock|unlock (any authenticated user — this is the tenant-facing "keyless door" use case from the spec). All lock/unlock routes 409 until pairing has produced HMIP_ACCESS_POINT/HMIP_AUTH_TOKEN. Does NOT cover temporary guest PIN codes on the keypad itself — that's handled through Homematic IP's own "eSchlüssel" app feature, which isn't exposed by this API; guest codes remain DB-only as before.
78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Nicht-interaktive Registrierung eines neuen Homematic-IP-Clients (analog zu
|
|
homematicip.cli.hmip_generate_auth_token, aber ohne input()-Prompts, damit
|
|
die Node-API sie als Hintergrundprozess starten und den Fortschritt pollen
|
|
kann).
|
|
|
|
Usage: python3 hmip_register.py <SGTIN> <STATUS_FILE>
|
|
|
|
Schreibt fortlaufend JSON-Status nach STATUS_FILE:
|
|
{"state": "waiting_for_button"}
|
|
{"state": "success", "authToken": "...", "accessPoint": "...", "clientId": "..."}
|
|
{"state": "error", "message": "..."}
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
|
|
import homematicip.auth
|
|
from homematicip.connection.connection_context import ConnectionContextBuilder
|
|
from homematicip.connection.rest_connection import RestConnection
|
|
|
|
|
|
def write_status(path, data):
|
|
with open(path, "w") as f:
|
|
json.dump(data, f)
|
|
|
|
|
|
async def main():
|
|
if len(sys.argv) != 3:
|
|
print("Usage: hmip_register.py <SGTIN> <STATUS_FILE>", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
access_point = sys.argv[1].replace("-", "").upper()
|
|
status_file = sys.argv[2]
|
|
|
|
if len(access_point) != 24:
|
|
write_status(status_file, {"state": "error", "message": "Ungültige SGTIN (erwartet 24 Zeichen ohne Bindestriche)"})
|
|
sys.exit(1)
|
|
|
|
try:
|
|
context = await ConnectionContextBuilder.build_context_async(access_point)
|
|
connection = RestConnection(context, log_status_exceptions=False)
|
|
auth = homematicip.auth.Auth(connection, context.client_auth_token, access_point)
|
|
|
|
response = await auth.connection_request(access_point, "wg-verwaltung")
|
|
if response.status != 200:
|
|
body = json.loads(response.text)
|
|
write_status(status_file, {"state": "error", "message": body.get("errorCode", "Unbekannter Fehler")})
|
|
sys.exit(1)
|
|
|
|
write_status(status_file, {"state": "waiting_for_button"})
|
|
|
|
attempts = 0
|
|
while not await auth.is_request_acknowledged():
|
|
attempts += 1
|
|
if attempts > 180: # 180 * 2s = 6 Minuten Timeout
|
|
write_status(status_file, {"state": "error", "message": "Zeitüberschreitung: Knopf am Access Point wurde nicht rechtzeitig gedrückt"})
|
|
sys.exit(1)
|
|
await asyncio.sleep(2)
|
|
|
|
auth_token = await auth.request_auth_token()
|
|
client_id = await auth.confirm_auth_token(auth_token)
|
|
|
|
write_status(status_file, {
|
|
"state": "success",
|
|
"authToken": auth_token,
|
|
"accessPoint": access_point,
|
|
"clientId": client_id,
|
|
})
|
|
except Exception as e:
|
|
write_status(status_file, {"state": "error", "message": str(e)})
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|