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

Retrieve network and device details

Browse Source
This commit is contained in:
Meatballs1
2024-03-22 02:55:19 +00:00
parent 953307484d
commit 57549abbc6
2 changed files with 100 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
HCDevice.py
View File

@@ -36,6 +36,7 @@
# /ce/status
#
# /ni/config
# /ni/info
#
# /iz/services
@@ -63,6 +64,9 @@ class HCDevice:
self.device_id = "0badcafe"
self.debug = False
self.name = name
self.services_initialized = False
self.services = {}
self.token = None
def parse_values(self, values):
if not self.features:
@@ -224,6 +228,15 @@ class HCDevice:
# send a message to the device
def get(self, resource, version=1, action="GET", data=None):
if self.services_initialized:
resource_parts = resource.split("/")
if len(resource_parts) > 1:
service = resource.split("/")[1]
if service in self.services.keys():
version = self.services[service]["version"]
else:
print(now(), self.name, "ERROR service not known")
msg = {
"sID": self.session_id,
"msgID": self.tx_msg_id,
@@ -252,6 +265,48 @@ class HCDevice:
print(self.name, "Failed to send", e, msg, traceback.format_exc())
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()
if self.services_initialized:
break
# We override the version based on the registered services received above
# the clothes washer wants this, the token doesn't matter,
# although they do not handle padding characters
# they send a response, not sure how to interpet it
self.token = base64url_encode(get_random_bytes(32)).decode("UTF-8")
self.token = re.sub(r"=", "", self.token)
self.get("/ci/authentication", version=2, data={"nonce": self.token})
self.get("/ci/info") # clothes washer
self.get("/iz/info") # dish washer
# Retrieves registered clients like phone/hcpy itself
self.get("/ci/registeredDevices")
# tzInfo all returns empty?
#self.get("/ci/tzInfo")
# We need to send deviceReady for some devices or /ni/ will come back as 403 unauth
self.get("/ei/deviceReady", version=2, action="NOTIFY")
self.get("/ni/info")
#self.get("/ni/config", data={"interfaceID": 0})
#self.get("/ro/allDescriptionChanges")
self.get("/ro/allMandatoryValues")
#self.get("/ro/values")
def handle_message(self, buf):
msg = json.loads(buf)
if self.debug:
@@ -264,7 +319,6 @@ class HCDevice:
values = {}
if "code" in msg:
print(now(), self.name, "ERROR", msg["code"])
values = {
"error": msg["code"],
"resource": msg.get("resource", ""),
@@ -284,41 +338,27 @@ class HCDevice:
"deviceID": self.device_id,
},
)
# ask the device which services it supports
self.get("/ci/services")
# the clothes washer wants this, the token doesn't matter,
# although they do not handle padding characters
# they send a response, not sure how to interpet it
token = base64url_encode(get_random_bytes(32)).decode("UTF-8")
token = re.sub(r"=", "", token)
self.get("/ci/authentication", version=2, data={"nonce": token})
self.get("/ci/info", version=2) # clothes washer
self.get("/iz/info") # dish washer
# self.get("/ci/tzInfo", version=2)
self.get("/ni/info")
# self.get("/ni/config", data={"interfaceID": 0})
self.get("/ei/deviceReady", version=2, action="NOTIFY")
self.get("/ro/allDescriptionChanges")
self.get("/ro/allDescriptionChanges")
self.get("/ro/allMandatoryValues")
# self.get("/ro/values")
else:
print(now(), self.name, "Unknown resource", resource, file=sys.stderr)
elif action == "RESPONSE" or action == "NOTIFY":
if resource == "/iz/info" or resource == "/ci/info":
# we could validate that this matches our machine
pass
if "data" in msg and len(msg["data"]) > 0:
# Return Device Information such as Serial Number, SW Versions, MAC Address
values = msg["data"][0]
elif resource == "/ro/descriptionChange" or resource == "/ro/allDescriptionChanges":
# we asked for these but don't know have to parse yet
### we asked for these but don't know have to parse yet
pass
elif resource == "/ni/info":
# we're already talking, so maybe we don't care?
if "data" in msg and len(msg["data"]) > 0:
# Return Network Information/IP Address etc
values = msg["data"][0]
elif resource == "/ni/config":
# Returns some data about network interfaces e.g.
# [{'interfaceID': 0, 'automaticIPv4': True, 'automaticIPv6': True}]
pass
elif resource == "/ro/allMandatoryValues" or resource == "/ro/values":
@@ -326,19 +366,33 @@ class HCDevice:
values = self.parse_values(msg["data"])
else:
print(now(), self.name, f"received {msg}")
elif resource == "/ci/registeredDevices":
# we don't care
# This contains details of Phone/HCPY registered as clients to the device
pass
elif resource == "/ci/tzInfo":
pass
elif resource == "/ci/authentication":
if "data" in msg and len(msg["data"]) > 0:
# Grab authentication token - unsure if this is for us to use
# or to authenticate the server. Doesn't appear to be needed
self.token = msg["data"][0]["response"]
elif resource == "/ci/services":
self.services = {}
for service in msg["data"]:
self.services[service["service"]] = {
"version": service["version"],
}
self.services_initialized = True
else:
print(now(), self.name, "Unknown response or notify:", msg)
pass
else:
print(now(), self.name, "Unknown", msg)
print(now(), self.name, "Unknown message", msg)
# return whatever we've parsed out of it
return values

43
hc2mqtt
View File

@@ -147,35 +147,21 @@ def client_connect(client, device, mqtt_topic):
host = device["host"]
device_topics = topics.copy()
for value in device["features"]:
if (
"access" in device["features"][value]
and "read" in device["features"][value]["access"].lower()
):
name = device["features"][value]["name"].split(".")
device_topics[name[-1]] = name[-1]
device_topics[value] = name[
-1
] # sometimes the returned key is a digit, making translation possible
state = {}
for topic in device_topics:
if not topic.isdigit(): # We only want the named topics
state[device_topics[topic]] = None
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(20)
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
ws.reconnect()
dev[device["name"]].reconnect()
while True:
msg = dev[device["name"]].recv()
@@ -185,14 +171,19 @@ def client_connect(client, device, mqtt_topic):
print(now(), device["name"], msg)
update = False
for topic in device_topics:
value = msg.get(topic, None)
if value is None:
continue
new_topic = device_topics[topic]
state[new_topic] = value
update = True
for key in msg.keys():
val = msg.get(key, None)
if key in state:
# Override existing values with None if they have changed
state[key] = val
update = True
else:
# Dont store None values until something useful is populated?
if val is None:
continue
else:
state[key] = val
update = True
if not update:
continue
@@ -209,9 +200,9 @@ def client_connect(client, device, mqtt_topic):
)
except Exception as e:
print(device["name"], "ERROR", e, file=sys.stderr)
print(device["name"], "hc2mqtt", e, file=sys.stderr)
time.sleep(40)
time.sleep(57)
if __name__ == "__main__":