bartek/hcpy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #43 from Meatballs1/websocket_keepalive

Browse Source 
Websocket keepalive
This commit is contained in:
pmagyar
2024-04-15 08:51:49 +02:00
committed by GitHub
parent 0bf9f33bea d98b568895
commit f0f6b5193f
3 changed files with 122 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
HCDevice.py
View File

@@ -43,6 +43,8 @@
import json
import re
import sys
import threading
import time
import traceback
from base64 import urlsafe_b64encode as base64url_encode
from datetime import datetime
@@ -55,18 +57,19 @@ def now():
class HCDevice:
def __init__(self, ws, features, name):
def __init__(self, ws, device):
self.ws = ws
self.features = features
self.features = device.get("features")
self.name = device.get("name")
self.session_id = None
self.tx_msg_id = None
self.device_name = "hcpy"
self.device_id = "0badcafe"
self.debug = False
self.name = name
self.services_initialized = False
self.services = {}
self.token = None
self.connected = False
def parse_values(self, values):
if not self.features:
@@ -152,7 +155,7 @@ class HCDevice:
feature = self.features[uid]
# check the access level of the feature
print(now(), self.name, f"Processing feature {feature['name']} with uid {uid}")
self.print(f"Processing feature {feature['name']} with uid {uid}")
if "access" not in feature:
raise Exception(
"Unable to configure appliance. "
@@ -209,7 +212,7 @@ class HCDevice:
try:
return self.handle_message(buf)
except Exception as e:
print(self.name, "error handling msg", e, buf, traceback.format_exc())
self.print("error handling msg", e, buf, traceback.format_exc())
return None
# reply to a POST or GET message with new data
@@ -234,7 +237,7 @@ class HCDevice:
if service in self.services.keys():
version = self.services[service]["version"]
else:
print(now(), self.name, "ERROR service not known")
self.print("ERROR service not known")
msg = {
"sID": self.session_id,
@@ -265,16 +268,14 @@ class HCDevice:
self.tx_msg_id += 1
def reconnect(self):
self.ws.reconnect()
# Receive initialization message /ei/initialValues
# Automatically responds in the handle_message function
self.recv()
# ask the device which services it supports
# registered devices gets pushed down too hence the loop
self.get("/ci/services")
while True:
self.recv()
time.sleep(1)
if self.services_initialized:
break
@@ -308,7 +309,7 @@ class HCDevice:
def handle_message(self, buf):
msg = json.loads(buf)
if self.debug:
print(now(), self.name, "RX:", msg)
self.print("RX:", msg)
sys.stdout.flush()
resource = msg["resource"]
@@ -336,8 +337,10 @@ class HCDevice:
"deviceID": self.device_id,
},
)
threading.Thread(target=self.reconnect).start()
else:
print(now(), self.name, "Unknown resource", resource, file=sys.stderr)
self.print("Unknown resource", resource, file=sys.stderr)
elif action == "RESPONSE" or action == "NOTIFY":
if resource == "/iz/info" or resource == "/ci/info":
@@ -363,7 +366,7 @@ class HCDevice:
if "data" in msg:
values = self.parse_values(msg["data"])
else:
print(now(), self.name, f"received {msg}")
self.print(f"received {msg}")
elif resource == "/ci/registeredDevices":
# This contains details of Phone/HCPY registered as clients to the device
@@ -386,10 +389,33 @@ class HCDevice:
self.services_initialized = True
else:
print(now(), self.name, "Unknown response or notify:", msg)
self.print("Unknown response or notify:", msg)
else:
print(now(), self.name, "Unknown message", msg)
self.print("Unknown message", msg)
# return whatever we've parsed out of it
return values
def run_forever(self, on_message, on_open, on_close):
def _on_message(ws, message):
values = self.handle_message(message)
on_message(values)
def _on_open(ws):
self.connected = True
on_open(ws)
def _on_close(ws, code, message):
self.connected = False
on_close(ws, code, message)
def on_error(ws, message):
self.print("Websocket error:", message)
self.ws.run_forever(
on_message=_on_message, on_open=_on_open, on_close=_on_close, on_error=on_error
)
def print(self, *args):
print(now(), self.name, *args)

62
HCSocket.py
View File

@@ -131,10 +131,6 @@ class HCSocket:
def reconnect(self):
self.reset()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 30)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 3)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)
sock.connect((self.host, self.port))
if not self.http:
@@ -153,10 +149,9 @@ class HCSocket:
buf = json.dumps(msg, separators=(",", ":"))
# swap " for '
buf = re.sub("'", '"', buf)
if self.debug:
print(now(), "TX:", buf)
self.dprint("TX:", buf)
if self.http:
self.ws.send_binary(self.encrypt(buf))
self.ws.send_bytes(self.encrypt(buf))
else:
self.ws.send(buf)
@@ -170,6 +165,55 @@ class HCSocket:
if buf is None:
return None
if self.debug:
print(now(), "RX:", buf)
self.dprint("RX:", buf)
return buf
def run_forever(self, on_message, on_open, on_close, on_error):
self.reset()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
if not self.http:
sock = sslpsk.wrap_socket(
sock,
ssl_version=ssl.PROTOCOL_TLSv1_2,
ciphers="ECDHE-PSK-CHACHA20-POLY1305",
psk=self.psk,
)
def _on_open(ws):
self.dprint("on connect")
on_open(ws)
def _on_close(ws, close_status_code, close_msg):
self.dprint(f"close: {close_msg}")
on_close(ws, close_status_code, close_msg)
def _on_message(ws, message):
if self.http:
message = self.decrypt(message)
self.dprint("RX:", message)
on_message(ws, message)
def _on_error(ws, error):
self.dprint(f"error {error}")
on_error(ws, error)
print(now(), "CON:", self.uri)
self.ws = websocket.WebSocketApp(
self.uri,
socket=sock,
on_open=_on_open,
on_message=_on_message,
on_close=_on_close,
on_error=_on_error,
)
websocket.setdefaulttimeout(30)
self.ws.run_forever(ping_interval=120, ping_timeout=10)
# Debug print
def dprint(self, *args):
if self.debug:
print(now(), *args)

61
hc2mqtt.py
View File

@@ -26,7 +26,7 @@ from HCSocket import HCSocket, now
@click.option("--mqtt_cafile")
@click.option("--mqtt_certfile")
@click.option("--mqtt_keyfile")
@click.option("--mqtt_clientname", default="hcpy")
@click.option("--mqtt_clientname", default="hcpy1")
@click_config_file.configuration_option()
def hc2mqtt(
devices_file: str,
@@ -90,7 +90,10 @@ def hc2mqtt(
else:
raise Exception(f"Payload topic {topic} is unknown.")
dev[device_name].get(resource, 1, "POST", msg)
if dev[device_name].connected:
dev[device_name].get(resource, 1, "POST", msg)
else:
print(now(), device_name, "ERROR cant send message as websocket is not connected")
except Exception as e:
print(now(), device_name, "ERROR", e, file=sys.stderr)
@@ -127,7 +130,6 @@ def hc2mqtt(
for device in devices:
mqtt_topic = mqtt_prefix + device["name"]
print(now(), f"topic: {mqtt_topic}")
thread = Thread(target=client_connect, args=(client, device, mqtt_topic))
thread.start()
@@ -140,35 +142,15 @@ dev = {}
def client_connect(client, device, mqtt_topic):
host = device["host"]
name = device["name"]
# HCDevice should maintain its own state?
state = {}
mqtt_set_topic = mqtt_topic + "/set"
print(now(), device["name"], f"set topic: {mqtt_set_topic}")
client.subscribe(mqtt_set_topic)
while True:
time.sleep(3)
try:
print(now(), device["name"], f"connecting to {host}")
ws = HCSocket(host, device["key"], device.get("iv", None))
dev[device["name"]] = HCDevice(ws, device.get("features", None), device["name"])
# ws.debug = True
dev[device["name"]].reconnect()
while True:
if client.is_connected():
client.publish(f"{mqtt_topic}/LWT", "", retain=True)
break
while True:
msg = dev[device["name"]].recv()
client.publish(f"{mqtt_topic}/LWT", "online")
if msg is None:
break
if len(msg) > 0:
print(now(), device["name"], msg)
def on_message(msg):
if msg is not None:
if len(msg) > 0:
print(now(), name, msg)
update = False
for key in msg.keys():
@@ -186,19 +168,34 @@ def client_connect(client, device, mqtt_topic):
update = True
if not update:
continue
return
if client.is_connected():
msg = json.dumps(state)
print(now(), device["name"], f"publish to {mqtt_topic} with {msg}")
print(now(), name, f"publish to {mqtt_topic} with {msg}")
client.publish(f"{mqtt_topic}/state", msg, retain=True)
else:
print(
now(),
device["name"],
name,
"ERROR Unable to publish update as mqtt is not connected.",
)
def on_open(ws):
client.publish(f"{mqtt_topic}/LWT", "", retain=True)
client.publish(f"{mqtt_topic}/LWT", "online")
def on_close(ws, code, message):
client.publish(f"{mqtt_topic}/LWT", "offline", retain=True)
print(now(), device["name"], "websocket closed, reconnecting...")
while True:
time.sleep(3)
try:
print(now(), name, f"connecting to {host}")
ws = HCSocket(host, device["key"], device.get("iv", None))
dev[name] = HCDevice(ws, device)
dev[name].run_forever(on_message=on_message, on_open=on_open, on_close=on_close)
except Exception as e:
print(now(), device["name"], "ERROR", e, file=sys.stderr)
client.publish(f"{mqtt_topic}/LWT", "offline", retain=True)