SoC module for OVMS (#2847)

* soc module for OVMS * flake8 fix --------- Co-authored-by: rleidner <[email protected]>
snaptec · Oct 10, 2024 · 3066029 · 3066029
1 parent 2995237
commit 3066029
Show file tree

Hide file tree

Showing 4 changed files with 520 additions and 0 deletions.
diff --git a/modules/soc_ovms/main.sh b/modules/soc_ovms/main.sh
@@ -0,0 +1,103 @@
+#!/bin/bash
+OPENWBBASEDIR=$(cd `dirname $0`/../../ && pwd)
+RAMDISKDIR="$OPENWBBASEDIR/ramdisk"
+MODULEDIR=$(cd `dirname $0` && pwd)
+DMOD="EVSOC"
+CHARGEPOINT=$1
+export OPENWBBASEDIR RAMDISKDIR MODULEDIR
+
+# check if config file is already in env
+if [[ -z "$debug" ]]; then
+	echo "soc_ovms: Seems like openwb.conf is not loaded. Reading file."
+	# try to load config
+	. $OPENWBBASEDIR/loadconfig.sh
+	# load helperFunctions
+	. $OPENWBBASEDIR/helperFunctions.sh
+fi
+
+case $CHARGEPOINT in
+	2)
+		# second charge point
+		ladeleistung=$(<$RAMDISKDIR/llaktuells1)
+		soctimerfile="$RAMDISKDIR/soctimer1"
+		socfile="$RAMDISKDIR/soc1"
+		fztype=$soc2type
+		server=$soc2server
+		username=$soc2user
+		password=$soc2pass
+		vehicleId=$soc2vehicleid
+		intervall=$(( soc2intervall * 6 ))
+		intervallladen=$(( soc2intervallladen * 6 ))
+		;;
+	*)
+		# defaults to first charge point for backward compatibility
+		# set CHARGEPOINT in case it is empty (needed for logging)
+		CHARGEPOINT=1
+		ladeleistung=$(<$RAMDISKDIR/llaktuell)
+		soctimerfile="$RAMDISKDIR/soctimer"
+		socfile="$RAMDISKDIR/soc"
+		server=$soc_ovms_server
+		username=$soc_ovms_username
+		password=$soc_ovms_passwort
+		vehicleId=$soc_ovms_vehicleid
+		intervall=$(( soc_ovms_intervall * 6 ))
+		intervallladen=$(( soc_ovms_intervallladen * 6 ))
+		;;
+esac
+
+incrementTimer(){
+	case $dspeed in
+		1)
+			# Regelgeschwindigkeit 10 Sekunden
+			ticksize=1
+			;;
+		2)
+			# Regelgeschwindigkeit 20 Sekunden
+			ticksize=2
+			;;
+		3)
+			# Regelgeschwindigkeit 60 Sekunden
+			ticksize=1
+			;;
+		*)
+			# Regelgeschwindigkeit unbekannt
+			ticksize=1
+			;;
+	esac
+	soctimer=$((soctimer+$ticksize))
+	echo $soctimer > $soctimerfile
+}
+
+getAndWriteSoc(){
+	openwbDebugLog ${DMOD} 2 "Lp$CHARGEPOINT: Requesting SoC"
+	echo 0 > $soctimerfile
+	echo $($MODULEDIR/soc_ovms.py --server "$server" --user "$username" --password "$password" --vehicleId "$vehicleId" --chargepoint "$CHARGEPOINT" 2>>$RAMDISKDIR/soc.log)
+	answer=$($MODULEDIR/soc_ovms.py --server "$server" --user "$username" --password "$password" --vehicleId "$vehicleId" --chargepoint "$CHARGEPOINT" 2>>$RAMDISKDIR/soc.log)
+	if [ $? -eq 0 ]; then
+		# we got a valid answer
+		echo $answer > $socfile
+		openwbDebugLog ${DMOD} 2 "Lp$CHARGEPOINT: SoC: $answer"
+	else
+		# we have a problem
+		openwbDebugLog ${DMOD} 0 "Lp$CHARGEPOINT: Error from soc_ovms: $answer"
+	fi
+}
+
+soctimer=$(<$soctimerfile)
+if (( ladeleistung > 500 )); then
+	if (( soctimer < intervallladen )); then
+		openwbDebugLog ${DMOD} 2 "Lp$CHARGEPOINT: Charging, but nothing to do yet. Incrementing timer."
+		incrementTimer
+	else
+		getAndWriteSoc
+	fi
+else
+	if (( soctimer < intervall )); then
+		openwbDebugLog ${DMOD} 2 "Lp$CHARGEPOINT: Nothing to do yet. Incrementing timer."
+		incrementTimer
+
+	else
+		getAndWriteSoc
+	fi
+fi
+
diff --git a/modules/soc_ovms/soc_ovms.py b/modules/soc_ovms/soc_ovms.py
@@ -0,0 +1,269 @@
+#!/usr/bin/python3
+
+from argparse import ArgumentParser
+import requests
+import logging
+import os
+import time
+from datetime import datetime
+from typing import Union
+from json import loads, dumps
+
+TS_FMT = "%Y-%m-%dT%H:%M:%S"
+
+# OVMS_SERVER = "https://ovms.dexters-web.de:6869"
+TOKEN_CMD = "/api/token"
+STATUS_CMD = "/api/status"
+OVMS_APPL_LABEL = "application"
+OVMS_APPL_VALUE = "owb-ovms-1.9"
+OVMS_PURPOSE_LABEL = "purpose"
+OVMS_PURPOSE_VALUE = "get soc"
+
+
+def utc2local(utc):
+    global log, session, token, vehicleId
+    epoch = time.mktime(utc.timetuple())
+    offset = datetime.fromtimestamp(epoch) - datetime.utcfromtimestamp(epoch)
+    return utc + offset
+
+
+# sync function
+def fetch_soc(id: str, pw: str, chargepoint: int) -> int:
+    global log, session, token, vehicleId
+
+    # get soc, from server
+    soc = _fetch_soc(id, pw, chargepoint)
+
+    return soc
+
+
+def read_token_file() -> dict:
+    global tokenFile
+    try:
+        with open(tokenFile, "r") as f:
+            jsonstr = f.read()
+            confDict = loads(jsonstr)
+    except Exception as e:
+        log.exception("Token file read exception" + str(e))
+        token = ""
+        confDict = {}
+        confDict['configuration'] = {}
+        confDict['configuration']['token'] = token
+    return confDict
+
+
+def write_token_file(confDict: dict):
+    global tokenFile
+    try:
+        with open(tokenFile, "w") as f:
+            jsonstr = dumps(confDict, indent=4)
+            f.write(jsonstr)
+    except Exception as e:
+        log.exception("Token file write exception" + str(e))
+
+    try:
+        os.chmod(tokenFile, 0o777)
+    except Exception as e:
+        log.exception("chmod tokenFile exception, e="+str(e))
+        os.system("sudo chmod 0777 "+tokenFile)
+
+
+def main():
+    global log, session, token, vehicleId, tokenFile, OVMS_SERVER
+
+    log = logging.getLogger("soc_ovms")
+    token = ""
+
+    parser = ArgumentParser()
+    parser.add_argument("-s", "--server",
+                        help="server", metavar="server", required=True)
+    parser.add_argument("-u", "--user",
+                        help="user", metavar="user", required=True)
+    parser.add_argument("-p", "--password",
+                        help="password", metavar="password", required=True)
+    parser.add_argument("-v", "--vehicleId",
+                        help="vehicleId", metavar="vehicleId", required=True)
+    parser.add_argument("-c", "--chargepoint",
+                        help="chargepoint", metavar="chargepoint", required=True)
+
+    args = vars(parser.parse_args())
+    OVMS_SERVER = args['server']
+    id = args['user']
+    pw = args['password']
+    vehicleId = args['vehicleId']
+    chargepoint = args['chargepoint']
+
+    # logging setup
+    debug = os.environ.get('debug', '0')
+    LOGLEVEL = 'WARN'
+    if debug == '1':
+        LOGLEVEL = 'INFO'
+    if debug == '2':
+        LOGLEVEL = 'DEBUG'
+    RAMDISKDIR = os.environ.get("RAMDISKDIR", "undefined")
+    logFile = RAMDISKDIR+'/soc.log'
+    format = '%(asctime)s %(levelname)s:%(name)s:Lp' + str(chargepoint) + ' %(message)s'
+    datefmt = '%Y-%m-%d %H:%M:%S'
+    logging.basicConfig(filename=logFile,
+                        filemode='a',
+                        format=format,
+                        datefmt=datefmt,
+                        level=LOGLEVEL)
+
+    log.debug("server=" + OVMS_SERVER +
+              ", user=" + id +
+              ", pw=" + pw +
+              ", vehicleId=" + vehicleId +
+              ", cp=" + chargepoint)
+    RAMDISKDIR = os.environ.get("RAMDISKDIR", "undefined")
+    tokenFile = RAMDISKDIR+'/soc_ovms_tokenlp'+chargepoint
+
+    with requests.Session() as session:
+        soc = fetch_soc(id, pw, chargepoint)
+        print(str(soc))
+
+
+# create a new token and store it in the soc_module configuration
+def create_token(user_id: str, password: str) -> str:
+    global log, session, token, vehicleId, OVMS_SERVER
+    token_url = OVMS_SERVER + TOKEN_CMD
+    data = {
+        "username": user_id,
+        "password": password
+    }
+    form_data = {
+        OVMS_APPL_LABEL: OVMS_APPL_VALUE,
+        OVMS_PURPOSE_LABEL: OVMS_PURPOSE_VALUE
+    }
+    try:
+        resp = session.post(token_url, params=data, files=form_data)
+    except Exception as e:
+        resp = e.response
+
+    log.debug("create_token status_code=" + str(resp.status_code))
+    tokenDict = loads(resp.text)
+    log.debug("create_token response=" + dumps(tokenDict, indent=4))
+    token = tokenDict['token']
+    confDict = {}
+    confDict['configuration'] = {}
+    confDict["configuration"]["token"] = resp.text.rstrip()
+    log.debug("create_token confDict=" + dumps(confDict, indent=4))
+    write_token_file(confDict)
+
+    return token
+
+
+# check list of token on OVMS server for unused token created by the soc mudule
+# if any obsolete token are found these are deleted.
+def cleanup_token(user_id: str, password: str):
+    global log, session, token, vehicleId, OVMS_SERVER
+    tokenlist_url = OVMS_SERVER + TOKEN_CMD + '?username=' + user_id + '&' + 'password=' + token
+
+    log.debug("tokenlist_url=" + tokenlist_url)
+    try:
+        resp = session.get(tokenlist_url)
+    except Exception as e:
+        log.error("cleanup_token: exception = " + str(e))
+        resp = e.response
+
+    status_code = resp.status_code
+    if status_code > 299:
+        log.error("cleanup_token status_code=" + str(status_code))
+        full_tokenlist = {}
+    else:
+        response = resp.text
+        full_tokenlist = loads(response)
+        log.debug("cleanup_token status_code=" +
+                  str(status_code) + ", full_tokenlist=\n" +
+                  dumps(full_tokenlist, indent=4))
+        obsolete_tokenlist = list(filter(lambda token:
+                                         token[OVMS_APPL_LABEL] == OVMS_APPL_VALUE and token["token"] != token,
+                                         full_tokenlist))
+        log.debug("cleanup_token: obsolete_tokenlist=\n" +
+                  dumps(obsolete_tokenlist, indent=4))
+        if len(obsolete_tokenlist) > 0:
+            log.debug("cleanup_token obsolete_tokenlist=\n" + dumps(obsolete_tokenlist, indent=4))
+            for tok in obsolete_tokenlist:
+                token_to_delete = tok["token"]
+                if token_to_delete != token:
+                    log.debug("cleanup_token: token_to_delete=" + dumps(tok, indent=4))
+                    token_del_url = OVMS_SERVER + TOKEN_CMD + '/' + token_to_delete
+                    token_del_url = token_del_url + '?username=' + user_id + '&password=' + token
+                    log.debug("token_del_url=" + token_del_url)
+                    try:
+                        resp = session.delete(token_del_url)
+                    except Exception as e:
+                        log.error("delete_token: exception = " + str(e))
+                        resp = e.response
+
+                    status_code = resp.status_code
+                else:
+                    log.debug("cleanup_token: skip active token: " + token)
+        else:
+            log.debug("cleanup_token: no obsolete token found")
+
+    return
+
+
+# get status for vehicleId
+def get_status(user_id: str) -> Union[int, dict]:
+    global log, session, token, vehicleId, OVMS_SERVER
+    status_url = OVMS_SERVER + STATUS_CMD + '/' + vehicleId + '?username=' + user_id + '&password=' + token
+
+    log.debug("status-url=" + status_url)
+    try:
+        resp = session.get(status_url)
+    except Exception as e:
+        resp = e.response
+
+    status_code = resp.status_code
+    if status_code > 299:
+        log.error("get_status status_code=" + str(status_code) + ", create new token")
+        respDict = {}
+    else:
+        response = resp.text
+        respDict = loads(response)
+        log.debug("get_status status_code=" + str(status_code) + ", response=" + dumps(respDict, indent=4))
+    return int(status_code), respDict
+
+
+# async function to fetch soc, range, soc_ts
+def _fetch_soc(user_id: str, password: str, chargepoint: int) -> int:
+    global log, session, token, vehicleId
+
+    confDict = read_token_file()
+    tokenstr = confDict['configuration']['token']
+    tokdict = loads(tokenstr)
+    token = tokdict['token']
+    log.debug("read token: " + token)
+    if token is None or token == "":
+        token = create_token(user_id, password)
+        log.debug("create_token: " + token)
+
+    log.debug("call get_status, token:" + token)
+    status_code, statusDict = get_status(user_id)
+    if status_code > 299:
+        token = create_token(user_id, password)
+        status_code, statusDict = get_status(user_id)
+        if status_code > 299:
+            raise "Authentication Problem, status_code " + str(status_code)
+
+    soc = statusDict['soc']
+    range = statusDict['estimatedrange']
+    kms = statusDict['odometer']
+    vehicle12v = statusDict['vehicle12v']
+
+    soc_ts = statusDict['m_msgtime_s']
+    log.info("soc=" + str(soc) +
+             ", range=" + str(range) +
+             ", soc_ts=" + str(soc_ts) +
+             ", km-stand=" + str(float(kms)/10) +
+             ", soc_12v=" + str(vehicle12v))
+
+    cleanup_token(user_id, password)
+
+    return int(float(soc))
+
+
+if __name__ == '__main__':
+    main()