alpha.py

import requests
from datetime import datetime
import hashlib
import argparse

parser = argparse.ArgumentParser(prog='alpha.py')
parser.add_argument('-u', '--username')
parser.add_argument('-p', '--password')
parser.add_argument('-m', '--max_current', type=int, help="set max current of charging pile (6-16)")
parser.add_argument('--start', help="Start charging")
parser.add_argument('--stop', help="Stop charging")
args = parser.parse_args()

# thanks to https://bitbucket.org/vdwals/alphaesscloud2mqtt/src/master/src/main/java/de/vdw/io/alpha2mqtt/config/Constants.java
AUTH_SIGNATURE_START = "al8e4s"
AUTH_SIGNATURE_HASH = "LS885ZYDA95JVFQKUIUUUV7PQNODZRDZIS4ERREDS0EED8BCWSS"
AUTH_SIGNATURE_END = "ui893ed"

MAX_CURRENT_RANGE = range(6, 16)


def get_auth_headers():
    now = datetime.now()
    auth_ts = str(int(datetime.timestamp(now)))    
    
    hash_str = AUTH_SIGNATURE_HASH + auth_ts
    auth_sig = AUTH_SIGNATURE_START + hashlib.sha512(hash_str.encode()).hexdigest() + AUTH_SIGNATURE_END
    return {
        "Authsignature": auth_sig,
        "Authtimestamp": auth_ts,
    }


def login(username, password):
    url = "https://cloud.alphaess.com/api/Account/Login"
    req = requests.post(url, json={"username": username, "password": password}, headers=get_auth_headers())
    data = req.json().get('data')
    return {
        'AccessToken': data['AccessToken'],
        'ExpiresIn': data['ExpiresIn'],
        'RefreshTokenKey': data['RefreshTokenKey'],                
    }
    

def get_settings(login_data, system_id=None):
    url = "https://cloud.alphaess.com/api/Account/GetCustomUseESSSetting"
    headers = get_auth_headers()
    headers["Authorization"] = "Bearer " + login_data["AccessToken"]
    #print(headers)
    params = {}
    if system_id:
        params = {
            'system_id': system_id
        }
    req = requests.get(url, params=params, headers=headers)
    #print(req)
    return req.json()
    
def post_settings(login_data, settings):
    url = "https://cloud.alphaess.com/api/Account/CustomUseESSSetting"
    headers = get_auth_headers()
    headers["Authorization"] = "Bearer " + login_data["AccessToken"]
    req = requests.post(url, json=settings, headers=headers)
    return req.json()
    

def get_system_list(login_data):
    url = "https://cloud.alphaess.com/api/Account/GetCustomUseESSList"
    headers = get_auth_headers()
    headers["Authorization"] = "Bearer " + login_data["AccessToken"]
    req = requests.get(url, headers=headers)
    return req.json()
    
def get_cp_status(login_data, sys_sn, chargingpile_id="EV1"):
    url = "https://cloud.alphaess.com/api/ESS/GetChargPileStatusByPileSn"
    headers = get_auth_headers()
    headers["Authorization"] = "Bearer " + login_data["AccessToken"]
    req = requests.post(url, json={'sys_sn': sys_sn, 'chargingpile_id': chargingpile_id}, headers=headers)
    j = req.json()
    pile_status_code = int(j['data'])
    
    if pile_status_code == 3:
        return "Lädt"
    if pile_status_code == 6:
        return "Ladevorgang beendet"
    return "Unbekannter Code %s, bitte auf der Webseite nachschauen" % pile_status_code
    
    
def start_charging(login_data, sys_sn, chargingpile_sn):
    url = "https://cloud.alphaess.com/api/ESS/StartCharging"
    headers = get_auth_headers()
    headers["Authorization"] = "Bearer " + login_data["AccessToken"]
    resp = requests.post(url, json={'sys_sn': sys_sn, 'chargingpile_sn': chargingpile_sn}, headers=headers)
    return resp.json()['code'] == 200
    
def stop_charging(login_data, sys_sn, chargingpile_sn):
    url = "https://cloud.alphaess.com/api/ESS/StopCharging"
    headers = get_auth_headers()
    headers["Authorization"] = "Bearer " + login_data["AccessToken"]
    resp = requests.post(url, json={'sys_sn': sys_sn, 'chargingpile_sn': chargingpile_sn}, headers=headers)
    return resp.json()['code'] == 200


login_data = login(username=args.username, password=args.password)
#settings_json = get_settings(login_data=login_data)
system_list_json = get_system_list(login_data=login_data)
first_system_id = system_list_json['data'][0]['system_id']
settings_json = get_settings(login_data=login_data, system_id=first_system_id)

cp_data = settings_json['data']['charging_pile_list'][0]
# Status Wallbox
status_text = get_cp_status(login_data, sys_sn=settings_json['data']['sys_sn'], chargingpile_id=cp_data['chargingpile_id'])
print("Status Wallbox: " + status_text)

# Lademodus, convert to int
max_current = int(cp_data['max_current'])
print("Aktueller Ladestrom %sA" % str(max_current))

if args.max_current:
    if int(args.max_current) in MAX_CURRENT_RANGE:
        max_current = int(args.max_current)
    else:
        print("Ladestrom ausserhalb 6-16A Bereich: %s" % str(args.max_current))
    print("Neuer Ladestrom %sA" % str(max_current))
    cp_data['max_current'] = str(max_current)
    cp_data['system_id'] = first_system_id

    settings_json['data']['charging_pile_list'][0] = cp_data
    settings_json['data'].update(cp_data)

    settings_data_json = settings_json['data']

    #import pprint
    #pprint.pprint(settings_json)

    post_settings(login_data=login_data, settings=settings_data_json)
    
if args.start:
    successful = start_charging(login_data, sys_sn=settings_json['data']['sys_sn'], chargingpile_sn=cp_data['chargingpile_sn'])
    if successful:
        print("Charging starting")
    
if args.stop:
    successful = stop_charging(login_data, sys_sn=settings_json['data']['sys_sn'], chargingpile_sn=cp_data['chargingpile_sn'])  
    if successful:
        print("Charging stopping")