Thanks for this @jswim788!
In the end I opted for the polling method, as this means I only need to maintain one system’s code (openHAB).
If this is useful for anyone else I wrote the below python classes to interact with my thermostat and boiler relay, and now use these classes in my Jython rules. Note: this is the first time I have ever written classes in any language, so use with an abundance of caution! Note that I am using the openHAB LogAction
, so I guess it’s not quite a standalone python class…
import requests
from core.actions import LogAction
vera_ip = "192.168.1.91"
vera_port = "3480"
class Vera:
def __init__(self):
self.ipaddress = vera_ip
self.port = vera_port
self.DeviceNum = 1
self.ZWaveNetworkServiceId = "urn:micasaverde-com:serviceId:ZWaveNetwork1"
def build_url(self, id, DeviceNum, serviceId, action):
url = "http://{}:{}".format(self.ipaddress, self.port)
url = url + "/data_request?"
url = url + "id={}".format(id)
url = url + "&DeviceNum={}".format(DeviceNum)
url = url + "&serviceId={}".format(serviceId)
url = url + "&{}".format(action)
url = url + "&output_format=text"
return str(url)
def set_parameter(self, DeviceNum, serviceId, action_name, parameter_name, parameter_value):
action = "action={}&{}={}".format(action_name, parameter_name, parameter_value)
url = self.build_url("action", DeviceNum, serviceId, action)
r = self.send_request(url)
if r == "OK":
return True
else:
return str(r)
def get_parameter(self, DeviceNum, serviceId, parameter_name):
action = "Variable={}".format(parameter_name)
url = self.build_url("variableget", DeviceNum, serviceId, action)
r = self.send_request(url)
return str(r)
def send_request(self, url):
r = requests.get(url)
return r.content
def get_zwave_network_status(self):
r = self.get_parameter(self.DeviceNum, self.ZWaveNetworkServiceId,"NetStatusText")
return str(r.upper())
def get_zwave_version(self):
r = self.get_parameter(self.DeviceNum, self.ZWaveNetworkServiceId,"VersionInfo")
return str(r.upper())
def get_zwave_last_update(self):
r = self.get_parameter(self.DeviceNum, self.ZWaveNetworkServiceId,"LastUpdate")
return str(r.upper())
class relay:
def __init__(self):
self.DeviceNum = 6
self.SwitchServiceId = "urn:upnp-org:serviceId:SwitchPower1"
self.HaDeviceServiceId = "urn:micasaverde-com:serviceId:HaDevice1"
self.HVACModeServiceId = "urn:upnp-org:serviceId:HVAC_UserOperatingMode1"
self.HVACStateServiceId = "urn:micasaverde-com:serviceId:HVAC_OperatingState1"
def set_switch_status(self, newTargetValue):
r = Vera().set_parameter(self.DeviceNum, self.SwitchServiceId, "SetTarget", "newTargetValue", newTargetValue)
if r == True:
LogAction.logInfo("set_switch_status", "Switched to {}".format(newTargetValue))
else:
LogAction.logWarn("set_switch_status","Switch change failed. {}".format(r))
def get_switch_status(self):
r = Vera().get_parameter(self.DeviceNum, self.SwitchServiceId,"Status")
#LogAction.logInfo("get_switch_status", "Switch status: {}".format(r))
return str(r)
def get_comms_status(self):
r = Vera().get_parameter(self.DeviceNum, self.HaDeviceServiceId, "CommFailure")
return int(r)
def get_mode_status(self):
r = Vera().get_parameter(self.DeviceNum, self.HVACModeServiceId,"ModeStatus")
return str(r.upper())
def get_mode_state(self):
r = Vera().get_parameter(self.DeviceNum, self.HVACStateServiceId,"ModeState")
return str(r.upper())
def get_last_update(self):
r = Vera().get_parameter(self.DeviceNum, self.HaDeviceServiceId,"LastUpdate")
return str(r)
class thermostat:
def __init__(self):
self.DeviceNum = 4
self.SetpointServiceId = "urn:upnp-org:serviceId:TemperatureSetpoint1_Heat"
self.SensorServiceId = "urn:upnp-org:serviceId:TemperatureSensor1"
self.HaDeviceServiceId = "urn:micasaverde-com:serviceId:HaDevice1"
self.HVACModeServiceId = "urn:upnp-org:serviceId:HVAC_UserOperatingMode1"
def set_setpoint(self, setpointTemp):
r = Vera().set_parameter(self.DeviceNum, self.SetpointServiceId, "SetCurrentSetpoint", "NewCurrentSetpoint", setpointTemp)
if r == True:
LogAction.logInfo("set_setpoint", "Thermostat changed to {}".format(setpointTemp))
else:
LogAction.logWarn("set_setpoint","Thermostat change failed. {}".format(r))
def get_current_setpoint(self):
r = Vera().get_parameter(self.DeviceNum, self.SetpointServiceId,"CurrentSetpoint")
#LogAction.logInfo("get_current_setpoint", "Current setpoint: {}".format(r))
return str(r)
def get_target_setpoint(self):
r = Vera().get_parameter(self.DeviceNum, self.SetpointServiceId,"SetpointTarget")
#LogAction.logInfo("get_target_setpoint", "Target setpoint: {}".format(r))
return str(r)
def get_current_temperature(self):
r = Vera().get_parameter(self.DeviceNum, self.SensorServiceId,"CurrentTemperature")
#LogAction.logInfo("get_current_temperature", "Current temperature: {}".format(r))
return str(r)
def get_battery_level(self):
r = Vera().get_parameter(self.DeviceNum, self.HaDeviceServiceId,"BatteryLevel")
#LogAction.logInfo("get_battery_level", "Battery level: {}".format(r))
return str(r)
def get_battery_level_last_update(self):
r = Vera().get_parameter(self.DeviceNum, self.HaDeviceServiceId,"BatteryDate")
return str(r)
def get_last_update(self):
r = Vera().get_parameter(self.DeviceNum, self.HaDeviceServiceId,"LastUpdate")
return str(r)
def get_mode_status(self):
r = Vera().get_parameter(self.DeviceNum, self.HVACModeServiceId,"ModeStatus")
return str(r.upper())
def get_comms_status(self):
r = Vera().get_parameter(self.DeviceNum, self.HaDeviceServiceId, "CommFailure")
return int(r)
To use these classes within a rule I:
- Saved the above into
$OPENHAB_CONF/automation/lib/python/personal/heating_functions.py
- Add the following to the top of any rules file which will use the classes:
import personal.heating_functions
reload(personal.heating_functions)
from personal.heating_functions import Vera
- To set the thermostat setpoint in a rule, for example:
Vera().thermostat().set_setpoint(20)
- To get the Z-Wave network status:
Vera().get_zwave_network_status()
EDIT 23/11/2020: Added more class methods.