Github Mirror / telegraf-plugins: 2728d625

verified

Move over to using a class.

Move over to using a class.

This is so that we can do object level tracking of the number of requests placed, in order to try and avoid triggering the service's rate limit

Commit 2728d625.

Authored 2023-05-14T12:04:03.000+01:00 by B Tasker in project Github Mirror / telegraf-plugins

+130 lines -114 lines

Commit Signature

Changes

diff --git a/soliscloud/soliscloud.py b/soliscloud/soliscloud.py
--- a/soliscloud/soliscloud.py
+++ b/soliscloud/soliscloud.py
# @@ -48,132 +48,150 @@ import json
# import os
# import requests
#
# -def configFromEnv():
# - ''' Build a dict of configuration settings based on environment variables
# - '''
# - return {
# - "api_id" : int(os.getenv("API_ID", 1234)),
# - "api_secret" : os.getenv("API_SECRET", "abcde"),
# - "api_url" : os.getenv("API_URL", "https://tobeconfirmed")
# - }
# -
#
# -def createHMAC(signstr,secret,algo):
# - ''' Create a HMAC of signstr using secret and algo
# -
# - https://snippets.bentasker.co.uk/page-1910021144-Generate-HMACs-with-different-algorithms-Python3.html
# - '''
# - hashedver = hmac.new(secret.encode('utf-8'),signstr.encode('utf-8'),algo)
# - return hashedver.digest()
# +class SolisCloud:
#
# + def __init__(self, config, session=False):
# + self.config = config
# + if session:
# + self.session = session
# + else:
# + self.session = requests.session()
#
# -def doAuth(key_id, secret, req_path, req_body, method="POST", content_type="application/json", datestring=False):
# - ''' Calculate an authorization header value to accompany the request
# -
# - Solis' API docs describe the method as:
# -
# - Authorization = "API " + KeyId + ":" + Sign
# - Sign = base64(HmacSHA1(KeySecret,
# - VERB + "\n"
# - + Content-MD5 + "\n"
# - + Content-Type + "\n"
# - + Date + "\n"
# - + CanonicalizedResource))
# + def createHMAC(self, signstr, secret, algo):
# + ''' Create a HMAC of signstr using secret and algo
# +
# + https://snippets.bentasker.co.uk/page-1910021144-Generate-HMACs-with-different-algorithms-Python3.html
# + '''
# + hashedver = hmac.new(secret.encode('utf-8'),signstr.encode('utf-8'),algo)
# + return hashedver.digest()
# +
# +
# + def doAuth(self, key_id, secret, req_path, req_body, method="POST", content_type="application/json", datestring=False):
# + ''' Calculate an authorization header value to accompany the request
#
# + Solis' API docs describe the method as:
#
# - Note: the API wants MD5s and SHA1s to be digests and not hexdigests
# - '''
# -
# - # Calculate an MD5 of the request body
# - # if there's no body, the hash should be empty
# - if len(req_body) > 0:
# - content_md5 = hashlib.md5(req_body.encode()).digest()
# - md5_str = base64.b64encode(content_md5).decode()
# - printDebug(f"Request body: {req_body}")
# - else:
# - md5_str = ''
# - printDebug(f"Empty Request body")
# -
# - # If there's no override, generate the current UTC date
# - # in HTTP header format
# - if not datestring:
# - printDebug(f"Calculating date")
# - d = datetime.datetime.now(tz=datetime.timezone.utc)
# - datestring = d.strftime('%a, %d %b %Y %H:%M:%S GMT')
# -
# - # Construct the string that we need to sign
# - # The entries should be seperated by \n - the API doesn't want
# - # literal newlines (presumably it evaluates them on read)
# - signstr = '\\n'.join([method,
# - md5_str,
# - content_type,
# - datestring,
# - req_path
# - ])
# -
# - printDebug(f"Signstr: {signstr}")
# -
# - # HMAC and then base64 it
# - hmacstr = createHMAC(signstr, secret, 'sha1')
# - signature = base64.b64encode(hmacstr).decode()
# -
# - # Take the values and construct the header value
# - auth_header = f"API {key_id}:{signature}"
# - printDebug(f"Calculated Auth header: {auth_header}")
# -
# - # Job done
# - return auth_header
# + Authorization = "API " + KeyId + ":" + Sign
# + Sign = base64(HmacSHA1(KeySecret,
# + VERB + "\n"
# + + Content-MD5 + "\n"
# + + Content-Type + "\n"
# + + Date + "\n"
# + + CanonicalizedResource))
# +
# +
# + Note: the API wants MD5s and SHA1s to be digests and not hexdigests
# + '''
# +
# + # Calculate an MD5 of the request body
# + # if there's no body, the hash should be empty
# + if len(req_body) > 0:
# + content_md5 = hashlib.md5(req_body.encode()).digest()
# + md5_str = base64.b64encode(content_md5).decode()
# + self.printDebug(f"Request body: {req_body}")
# + else:
# + md5_str = ''
# + self.printDebug(f"Empty Request body")
# +
# + # If there's no override, generate the current UTC date
# + # in HTTP header format
# + if not datestring:
# + self.printDebug(f"Calculating date")
# + d = datetime.datetime.now(tz=datetime.timezone.utc)
# + datestring = d.strftime('%a, %d %b %Y %H:%M:%S GMT')
# +
# + # Construct the string that we need to sign
# + # The entries should be seperated by \n - the API doesn't want
# + # literal newlines (presumably it evaluates them on read)
# + signstr = '\\n'.join([method,
# + md5_str,
# + content_type,
# + datestring,
# + req_path
# + ])
# +
# + self.printDebug(f"Signstr: {signstr}")
# +
# + # HMAC and then base64 it
# + hmacstr = self.createHMAC(signstr, secret, 'sha1')
# + signature = base64.b64encode(hmacstr).decode()
# +
# + # Take the values and construct the header value
# + auth_header = f"API {key_id}:{signature}"
# + self.printDebug(f"Calculated Auth header: {auth_header}")
# +
# + # Job done
# + return auth_header
#
#
# -def fetchStationList(config, session):
# - ''' Fetch the list of stations.
# -
# - In the Solicloud UI these are referred to as plants
# - Basically, the site at which devices are deployed.
# -
# - So, if you had multiple inverters in one location this
# - would return the total values
# -
# - TODO: may want to implement iterating through pages at some
# - point
# + def fetchStationList(self):
# + ''' Fetch the list of stations.
# +
# + In the Solicloud UI these are referred to as plants
# + Basically, the site at which devices are deployed.
# +
# + So, if you had multiple inverters in one location this
# + would return the total values
# +
# + TODO: may want to implement iterating through pages at some
# + point
# + '''
# +
# + # Construct the request body
# + req_body_d = {
# + "pageNo": 1,
# + "pageSize" : 100
# + }
# + req_body = json.dumps(req_body_d)
# + req_path = "/v1/api/userStationList"
# +
# + # Construct an auth header
# + auth_header = self.doAuth(self.config['api_id'], self.config['api_secret'], req_path, req_body)
# +
# + # Construct headers dict
# + headers = {
# + "Authorization" : auth_header,
# + "Content-Type" : "application/json"
# + }
# +
# + self.printDebug(f'Built request - Headers {headers}, body: {req_body}, path: {req_path}')
# +
# + # Place the request
# + r = self.session.post(
# + url = f"{self.config['api_url']}{req_path}",
# + headers = headers,
# + data = req_body
# + )
# +
# + return r.json()
# +
# + def printDebug(self, msg):
# + if DEBUG:
# + print(msg)
# +
# +
# +
# +# Utility functions to help with __main__ runs
# +
# +def configFromEnv():
# + ''' Build a dict of configuration settings based on environment variables
# '''
# -
# - # Construct the request body
# - req_body_d = {
# - "pageNo": 1,
# - "pageSize" : 100
# - }
# - req_body = json.dumps(req_body_d)
# - req_path = "/v1/api/userStationList"
# -
# - # Construct an auth header
# - auth_header = doAuth(config['api_id'], config['api_secret'], req_path, req_body)
# -
# - # Construct headers dict
# - headers = {
# - "Authorization" : auth_header,
# - "Content-Type" : "application/json"
# + return {
# + "api_id" : int(os.getenv("API_ID", 1234)),
# + "api_secret" : os.getenv("API_SECRET", "abcde"),
# + "api_url" : os.getenv("API_URL", "https://tobeconfirmed")
# }
# -
# - # Place the request
# - r = session.post(
# - url = f"{config['api_url']}{req_path}",
# - headers = headers,
# - data = req_body
# - )
# -
# - return r.json()
#
# -
# -def printDebug(msg):
# - if DEBUG:
# - print(msg)
#
#
# if __name__ == "__main__":
# DEBUG = True
# + config = configFromEnv()
# + soliscloud = SolisCloud(config)
# +
# # These are the example values used in the API doc
# - print(doAuth('2424',
# + print(soliscloud.doAuth('2424',
# '668018254',
# '/v1/api/userStationList',
# '{"pageNo":1,"pageSize":10}',
# @@ -182,6 +200,4 @@ if __name__ == "__main__":
# datestring='Fri, 26 Jul 2019 06:00:46 GMT')
# )
#
# - session = requests.session()
# - config = configFromEnv()
# - stations = fetchStationList(config, session)
# + stations = soliscloud.fetchStationList()
#