Hello,
Newbie here.
Can some one help me implement this python code thru node-red?
import json
import requests
class xComfortAPI:
def init(self, url, username, password, verbose=False):
"""
Class to access a Smart Home Controller (SHC) via its web API.
:param url: the URL of the SHC, for example: http://my_home.dyndns.org:8080
:param username: normally "admin" - the user you normally use to log on to the SHC web page
:param password: the password for the user
"""
self.url = url
self.username = username
self.password = password
self.verbose = verbose
self.session_ID = ''
self._get_session_id()
if self.verbose:
print('Got session id "', self.session_ID, '"', sep='')
def _get_session_id(self):
"""
Logs in to the SHC and gives us a session ID cookie that is authenticated and we reuse to make our JSON RPC calls
:return: session ID cookie(string)
"""
headers = {'User-Agent': 'Mozilla/5.0'}
session = requests.Session()
try:
session.get(self.url)
except Exception as err:
if 'failed to respond' in str(err):
print('ERROR:', self.url, 'did not respond to our connection attempt. Please check your URL address\nAborting...')
exit(1)
else:
print('ERROR:', err, '\nAborting...')
exit(1)
response = session.post(self.url, headers=headers, auth=(self.username, self.password))
if response.status_code != 200:
if response.status_code == 401:
print('Invalid username/password\nAborting...')
exit(1)
else:
print('Server responded with status code', str(response.status_code), '\nAborting...')
else:
self.session_ID = requests.utils.dict_from_cookiejar(session.cookies)['JSESSIONID']
def query(self, method, params=['', '']):
"""
Sends query to SHC's JSON RPC interface
:param method: the JSON RPC method
:param params: parameters for the JSON RPC method (defaults to list of two empty string)
:return: dict of response from SHC
"""
json_url = self.url + '/remote/json-rpc'
data = {
"jsonrpc": "2.0",
'method': method, # Settings/getSystemInfo
'params': params,
'id': 1
}
headers = {
'Cookie': 'JSESSIONID=' + self.session_ID,
'Accept-Encoding': 'gzip, deflate',
'Content-Type': 'application/json',
'Accept': 'application/json, text/javascript, */*; q=0.01',
}
try:
response = requests.post(json_url, data=json.dumps(data), headers=headers).json()
except ValueError as err:
if err == 'Expecting value: line 1 column 1 (char 0)':
print('The SHC gave an invalid response. Most likely, this is due to an incorrect URL in your INI-file')
print('Please verify that your URL is of the format "http://IP:port", or just "http://IP" if you use default port 80')
exit(1)
if 'result' not in response:
response['result'] = [{}] # In case we have a zone without devices for example
if self.verbose:
print('-> NO datapoints returned')
elif self.verbose:
print('->', len(response['result']), 'datapoints returned')
return response['result']
def get_zone_devices(self):
"""
Getting all zones and functions
:return: list of zones with all devices in each zone
"""
zone_list = self.query('HFM/getZones') # Returns a list of dicts in this case
for zone in zone_list:
zone.pop('virtual', None)
zone.pop('mainIndoorTemperature', None)
zone['devices'] = self.query('StatusControlFunction/getDevices', params=[zone['zoneId'], ''])
return zone_list
@staticmethod
def print_zones(zones):
"""
Prints all zones and their devices in human readable form
Static method
:param zones:
"""
for zone in zones:
print(zone['zoneName'], '(' + zone['zoneId'] + ')')
for device in zone['devices']:
if 'name' in device:
print('\t{:<55}{:>10}{}'.format(device['name'], device['value'], device['unit']))
def show_diagnostics(self):
"""
Prints the SHC diagnostics info
"""
all_diagnostics = self.query('Diagnostics/getAllSystemStates')
print('\n### SHC diagnostic information ###')
for diagnostics in all_diagnostics:
print(diagnostics['statusValue'])
print('\n')
def switch(self, zone, dev_id, state):
"""
Turns off a device
:param dev_id: the SHC ID of the device (like 'xCo:5355820_u0')
:param zone: the zone the device is in (like 'hz_1')
:param state: 'on', 'off' or '' - if '' it means "switch to other" (on if off, off if on)
:return: True if successful, False otherwise
"""
if not state == 'on' and not state == 'off' and not state == '':
print('Error: switch() state must be "on", "off" or "" (blank), not "' + str(state) + '"')
exit(1)
result = self.query('StatusControlFunction/controlDevice', params=[zone, dev_id, 'off'])
if not result['status'] == 'ok':
return False
else:
return True