Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

stejk · Příspěvekod **stejk** » ned 05. led 2020 14:02:56

Bohužel pokračování našeho občasníku nyní ukončuji. Mezi virtualboxem a hostujícím systémem se mi nedaří rozchodit přímé TCP spojení, a nemám proto jak odladit čtvrtou část návodu (rozchození webového serveru a skriptů pro zobrazení dat z databáze). Čtenářům proto navrhuji použít řešení Havrlovo, kterému tímto děkuji. A všem vám přeji klidný rok.

Havrla · Příspěvekod **Havrla** » pon 06. led 2020 21:28:09

Hezke povanocni.
Pripravim image linuxoveho virtualu s BMW skryptem a grafanou ke stazeni.

Havrla · Příspěvekod **Havrla** » pon 17. úno 2020 13:45:12

#################################################################################
17.2.2020
BMW zmenilo servery, nutno upravit:

# API Gateway
# North America: b2vapi.bmwgroup.us
# Rest of World: b2vapi.bmwgroup.com
# China: b2vapi.bmwgroup.cn:8592
BASE_URL = 'customer.bmwgroup.com'

a v sekci getSocData

BASE_URLL = 'b2vapi.bmwgroup.com'
url = 'https://' + BASE_URLL + '/api/vehicle/navigation/v1/' + vin

Havrla · Příspěvekod **Havrla** » pát 21. úno 2020 15:13:25

BMW dalsi znema prihlaseni

Kód: Vybrat vše


BASE_URL = 'customer.bmwgroup.com'
#BASE_URL = 'b2vapi.bmwgroup.com'

def getToken(BASE_URL, username, password, timeout=(20, 20)):
    try:
        headers = {
                "Content-Type": "application/x-www-form-urlencoded",
                "Content-Length": "124",
                "Connection": "Keep-Alive",
                "Host": BASE_URL,
                "Accept-Encoding": "gzip",
                "Authorization": "Basic blF2NkNxdHhKdVhXUDc0eGYzQ0p3VUVQOjF6REh4NnVuNGNEanli"
                                 "TEVOTjNreWZ1bVgya0VZaWdXUGNRcGR2RFJwSUJrN3JPSg==",
                "Credentials": "nQv6CqtxJuXWP74xf3CJwUEP:1zDHx6un4cDjybLENN3kyfumX2kEYigWPcQpdvDRpIBk7rOJ",
                "User-Agent": "okhttp/2.60"}

        data = {
            'client_id': 'dbf0a542-ebd1-4ff0-a9a7-55172fbfce35',
            'response_type': 'token',
            'redirect_uri': 'https://www.bmw-connecteddrive.com/app/static/external-dispatch.html',
            'scope': 'authenticate_user vehicle_data remote_services',
            'username': username,
            'password': password}

        data = urllib.parse.urlencode(data)#       url = 'https://' + BASE_URL + '/webapi/oauth/token'
#        url = 'https://' + BASE_URL + '/webapi/oauth/token'
        url = 'https://' + BASE_URL + '/gcdm/oauth/authenticate'
        #print (url)
        r = requests.post(url, data=data, headers=headers, allow_redirects=False)
        #print (r)
        if r.status_code == 302:
            logging.info('Access token acquired')
            response_json = dict(
                    urllib.parse.parse_qsl(urllib.parse.urlparse(r.headers['Location']).fragment)
                )
            #print(response_json)
            #print(response_json['access_token'])
            return response_json['access_token']
        else:
            raise ValueError('Unable to login')

    except Exception as msg:
        logging.error(msg)

Havrla · Příspěvekod **Havrla** » stř 10. lis 2021 14:37:23

Tak nam BMW na nekterych serverech vraci :

########################################
HTTP Status 410 - Gone
type Status report
messageGone
descriptionThe requested resource is no longer available, and no forwarding address is known.
Payara Micro #badassfish
########################################

url s soc jeste funguje :-)

, pozice lze vycitat odtud.

stejk · Příspěvekod **stejk** » čtv 11. lis 2021 22:57:39

das prosim URL ?

Havrla · Příspěvekod **Havrla** » pát 12. lis 2021 15:40:28

Kód: Vybrat vše



[root@CL-ITNK-BMW ~]# cat /usr/local/bin/bmw-uni.py 
#! /usr/bin/env python3

import requests
import json
import urllib
import logging
import sys, os, time
import pickle
import pymysql
# Import smtplib for the actual sending function
import smtplib
# Import the email modules we'll need
from email.mime.text import MIMEText
from getopt import getopt, GetoptError



#   API Gateway
#   North America: b2vapi.bmwgroup.us
#   Rest of World: b2vapi.bmwgroup.com
#   China: b2vapi.bmwgroup.cn:8592
BASE_URL = 'customer.bmwgroup.com'
#BASE_URL = 'b2vapi.bmwgroup.com'

def usage():
    print("Usage : %s -v VINnumber" % sys.argv[0])
    print("")
    print("     -v VIN_number")
    print("     -u user")
    print("     -p password")
    print("     -t token")
    print("     -w apiweatherkey")
    print("     -h help") 
#enddef

apikey=""                                                                                                                                                                                                                                    
credentials={}                                                                                                                                                                                                                               
credentials['access_token']="TOKEN"                                                                                                                                                                                                          
                                                                                                                                                                                                                                             
                                                                                                                                                                                                                                             
if len(sys.argv) > 1:                                                                                                                                                                                                                        
    try:                                                                                                                                                                                                                                     
        opts,args = getopt(sys.argv[1:], "v:u:p:t:w:h")                                                                                                                                                                                      
    except GetoptError as err:                                                                                                                                                                                                               
        print( "error:",  err)                                                                                                                                                                                                               
        sys.exit(1)                                                                                                                                                                                                                          
    #endtry

    for (arg,value) in opts:
        if (arg == "-v"):
            try:
                credentials['vin'] = str(value)
            except:
                print("ERROR parametru! -v vin_number")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-u"):
            try:
                credentials['username'] = str(value)
            except: 
                print("ERROR parametru! -u username")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-p"):
            try:
                credentials['password'] = str(value)
            except: 
                print("ERROR parametru! -p password")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-t"):
            try:
                credentials['access_token'] = str(value)
            except:
                print("ERROR parametru! -t TOKEN")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-w"):
            try:
                apikey = str(value)
            except:
                print("ERROR parametru! -w api_key_weather")
                sys.exit(1)
            #endtry
        #endif




        if (arg == "-h"):
            usage()
            sys.exit(1)
        #endif
    #endfor
else:
    usage()
    sys.exit(1)
#endif

# 7 difit from VIN
vin=credentials['vin']
vin7=vin[-7:]

logging.basicConfig(filename="/var/log/bmw/bmw_%s.log" % vin7,
                            filemode='a',
                            format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
                            datefmt='%H:%M:%S',
                            level=logging.INFO)


def getToken(BASE_URL, username, password, timeout=(20, 20)):
    try:
        headers = {
                "Content-Type": "application/x-www-form-urlencoded",
                "Content-Length": "124",
                "Connection": "Keep-Alive",
                "Host": BASE_URL,
                "Accept-Encoding": "gzip",
                "Authorization": "Basic blF2NkNxdHhKdVhXUDc0eGYzQ0p3VUVQOjF6REh4NnVuNGNEanli"
                                 "TEVOTjNreWZ1bVgya0VZaWdXUGNRcGR2RFJwSUJrN3JPSg==",
                "Credentials": "nQv6CqtxJuXWP74xf3CJwUEP:1zDHx6un4cDjybLENN3kyfumX2kEYigWPcQpdvDRpIBk7rOJ",
                "User-Agent": "okhttp/2.60"}

        data = {
            'client_id': 'dbf0a542-ebd1-4ff0-a9a7-55172fbfce35',
            'response_type': 'token',
            'redirect_uri': 'https://www.bmw-connecteddrive.com/app/static/external-dispatch.html',
            'scope': 'authenticate_user vehicle_data remote_services',
            'username': username,
            'password': password}

        data = urllib.parse.urlencode(data)#       url = 'https://' + BASE_URL + '/webapi/oauth/token'
#        url = 'https://' + BASE_URL + '/webapi/oauth/token'
        url = 'https://' + BASE_URL + '/gcdm/oauth/authenticate'
        #print (url)
        r = requests.post(url, data=data, headers=headers, allow_redirects=False)
        #print (r)
        if r.status_code == 302:
            logging.info('Access token acquired')
            response_json = dict(
                    urllib.parse.parse_qsl(urllib.parse.urlparse(r.headers['Location']).fragment)
                )
            #print(response_json)
            #print(response_json['access_token'])
            return response_json['access_token']
        else:
            raise ValueError('Unable to login')

    except Exception as msg:
        logging.error(msg)


def getSocData(BASE_URL, token, vin):
    try:
        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1',
                    'Authorization': 'Bearer ' + token}

        BASE_URLL = 'b2vapi.bmwgroup.com'
        url = 'https://' + BASE_URLL + '/api/vehicle/navigation/v1/' + vin

#        print(url)
        r = requests.get(url, headers=headers, timeout=(20, 20))
#        print(r)
#        print(r.json())
        if r.status_code == 200:
            status = r.json()
            if status.get('socmax'):
                 socMax = status['socmax']
            if status.get('socMax'):
                 socMax = status['socMax']
            positionlat = status['latitude']
            positionlon = status['longitude']
            soc = status['soc']
            logging.info('SoC Data retreived successfully')
            return socMax, positionlat, positionlon, soc
        else:
            raise ValueError('Unable to retreive SoC data from vehicle')

    except Exception as msg:
        logging.error(msg)


def getBattery(BASE_URL, token, vin):
    try:
        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1',
                    'Authorization': 'Bearer ' + token}

        url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/status'
        r = requests.get(url, headers=headers, timeout=(20, 20))
        #print(r.json())

        if r.status_code == 200:
            status = r.json()
            chargingLevelHv = status['vehicleStatus']['chargingLevelHv']
            remainingRangeElectric = status['vehicleStatus']['remainingRangeElectric']
            connectionStatus = status['vehicleStatus']['connectionStatus']
            chargingStatus = status['vehicleStatus']['chargingStatus']
            mileage = status['vehicleStatus']['mileage']
            positionlat = status['vehicleStatus']['position']['lat']
            positionlon = status['vehicleStatus']['position']['lon']
            try:
                remainingFuel = status['vehicleStatus']['remainingFuel']
            except:
                remainingFuel = 0
            try:
                remainingRangeFuel = status['vehicleStatus']['remainingRangeFuel']
            except:
                remainingRangeFuel = 0
            updateReason = status['vehicleStatus']['updateReason']
            logging.info('Battery data retreived successfully')
            return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason

        else:
#            raise ValueError('Unable to retreive battery data from vehicle')
            chargingLevelHv = 0
            remainingRangeElectric = 0
            connectionStatus = "unknown"
            chargingStatus = "unknown"
            mileage = 0
            positionlat = 49.83122600
            positionlon = 14.69111800
            remainingFuel = 0
            remainingRangeFuel = 0
            updateReason = 0
            logging.info('Battery data retreived NOT successfully')
            return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason

    except Exception as msg:
        logging.error(msg)


def getTrip(BASE_URL, token, vin):
    try:
        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1',
                    'Authorization': 'Bearer ' + token}

        url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/statistics/lastTrip'

        r = requests.get(url, headers=headers, timeout=(20, 20))
        #print(r.json())

        if r.status_code == 200:
            status = r.json()
            avgElectricConsumption = status['lastTrip']['avgElectricConsumption']
            avgRecuperation = status['lastTrip']['avgRecuperation']
            totalDistance = status['lastTrip']['totalDistance']
            electricDistance = status['lastTrip']['electricDistance']
            duration = status['lastTrip']['duration']
            logging.info('Trip data retreived successfully') 
            return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration

        else:
#            raise ValueError('Unable to retreive Trip data from vehicle')
            avgElectricConsumption = 0
            avgRecuperation = 0
            totalDistance = 0
            electricDistance = 0
            duration = 0
            logging.info('Trip data retreived  NOT  successfully')
            return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration

    except Exception as msg:
        logging.error(msg)



#  https://api.openweathermap.org/data/2.5/onecall?lat=49.84843&lon=14.704501&appid=07ee063f9e1248912cd522104a802244&exclude=minutely,hourly,daily,alerts&units=metric

def getTeplota(positionlat, positionlon):
    try:
        url="https://api.openweathermap.org/data/2.5/onecall?lat="+str(positionlat)+"&lon="+str(positionlon)+"&appid="+apikey+"&exclude=minutely,hourly,daily,alerts&units=metric"
        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1'}
        r = requests.get(url, headers=headers, timeout=(20, 20))
        if r.status_code == 200:
            status = r.json()
            teplota = status['current']['temp']
            logging.info('Teplota data retreived successfully')
            return teplota

        else:
            raise ValueError('Unable to retreive Teplota data from internet at car position.')

    except Exception as msg:
        logging.error(msg)


def getPower():
    try:
        soubor = open("/tmp/bmw_power_%s.txt" % vin7, mode='rt')
        vykon = soubor.read()
        soubor.close()
    except:
        vykon=0
    return vykon

## TODO
#def sendEMAIL(message_chargingLevelHv,message_connectionStatus,message_chargingStatus,message_subjectEmoji):
#    try:
#        logging.info(message_chargingLevelHv)
#        logging.info(message_connectionStatus)
#        logging.info(message_chargingStatus)
#        SMTP_SERVER = '10.27.23.12'
#        SMTP_PORT = 587
#        GMAIL_USERNAME = 'xxx@xxx.cz'
#        GMAIL_PASSWORD = 'xxx #CAUTION: This is stored in plain text!
#        recipient = 'xxxx@xxx.cz'
#        Message_chargingLevelHv=str(message_chargingLevelHv)
##        message_subjectEmoji = '=?utf-8?Q? =F0=9F=9A=97 ?='
#        subject = message_subjectEmoji + ' BMW ' + Message_chargingLevelHv + '% '
#        emailText = 'pripojeni:' + message_connectionStatus + '<br>nabijeni:' + message_chargingStatus
#        headers = ["From: " + GMAIL_USERNAME,
#                "Subject: " + subject,
#                "To: " + recipient,
#                "MIME-Version: 1.0",
#                "Content-Type: text/html"]
#        headers = "\r\n".join(headers)
#        session = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
#        session.ehlo()
#        session.starttls()
#        session.ehlo
#        session.login(GMAIL_USERNAME, GMAIL_PASSWORD)
#        session.sendmail(GMAIL_USERNAME, recipient, headers + "\r\n\r\n" + emailText)
#        session.quit()
#        logging.info('EMAIL message sent successfully')
#    except Exception as msg:
#        logging.error(msg)
#        sys.exit(1)
#

def getLastPercent():
    try:
        with open ("/var/spool/bmw/last_percent_%s" % vin7  , 'rb') as lp:
            last_percent = pickle.load(lp)
        logging.info(f'last_percent loaded with a value of {last_percent}')
        return last_percent
    except FileNotFoundError:
        logging.error('last_percent file not found')
        with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp:
            pickle.dump(0, lp)
        logging.info('last_percent file created with value of 0')
        with open ("/var/spool/bmw/last_percent_%s" % vin7 , 'rb') as lp:
            last_percent = pickle.load(lp)
        return last_percent

def setLastPercent(chargingLevelHv):
    try:
        with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp:
            pickle.dump(chargingLevelHv, lp)
        logging.info(f'last_percent file updated with {chargingLevelHv}')
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)


try:
    last_percent = int(getLastPercent())

    try:
        token = getToken(BASE_URL, credentials['username'], credentials['password'])
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)

    try:
         chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason = getBattery(BASE_URL, token, credentials['vin'])
    except Exception as msg:
         logging.error(msg) 
         sys.exit(1)   

    try:        
        socMax, positionlat, positionlon, chargingLevelHv  = getSocData(BASE_URL, token, credentials['vin'])
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)

    try:
        avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration = getTrip(BASE_URL, token, credentials['vin'])
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)


    try:    
        teplota = getTeplota(positionlat, positionlon)
    except Exception as msg:
        logging.error(msg)
        teplota = 1

    try:
        CharegPower = getPower()
    except Exception as msg:
        logging.error(msg)
        CharegPower = 0


    message = 'BMW i3 Status:\n' + f'Battery: {chargingLevelHv}% ({remainingRangeElectric} km) - {connectionStatus}/{chargingStatus}\nMileage: {mileage} km\nPosition http://maps.apple.com/?ll={positionlat},{positionlon}\n' + f'socMax: {socMax} kWh\n'






    conn = pymysql.connect(host='127.0.0.1', unix_socket='/var/lib/mysql/mysql.sock', user='bmw', passwd='xxxxxxxx', db="bmw_%s" % vin7)
    cur = conn.cursor()

    try:
        cur.execute('''SELECT mileage FROM i3  WHERE chargingStatus = "CHARGING" ORDER BY datum DESC LIMIT 1''')
        mileageCH= cur.fetchone()
        try:
            DistanceSinceLastCharge = mileage - mileageCH[0]
        except:
            DistanceSinceLastCharge = 0
        cur.execute('''SELECT mileage FROM i3  WHERE chargingStatus = "FINISHED_FULLY_CHARGED" ORDER BY datum DESC LIMIT 1''')
        mileageCH= cur.fetchone()
        try:
            DistanceSinceLastFCharg = mileage - mileageCH[0]
        except:
            DistanceSinceLastFCharg = 0 

    except pymysql.IntegrityError:
        logging.warn("failed to select values")



    try:
        affected_count = cur.execute('''INSERT INTO i3 (mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)''',(mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg))
        conn.commit()
        logging.info("inserted values to DB")
    except pymysql.IntegrityError:
        logging.warn("failed to insert values")
    finally:
        cur.close()
        conn.close()


#   nabijeni poprve dosalo 100 procent (mame plne nabito)
    if chargingLevelHv > last_percent and chargingLevelHv == 100:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = ''
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   blizime se uz nabitemu stavu
    elif chargingLevelHv > last_percent and chargingLevelHv > 83:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = ''
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)

# chargingStatus
# INVALID
# ERROR
# NOT_CHARGING
# CHARGING
# FINISHED_FULLY_CHARGED

# connectionStatus
# CONNECTED
# DISCONNECTED

# DISCONNECTED INVALID      vuz neni pripojen a nenabiji se - nejcasteji
# CONNECTED    NOT_CHARGING vuz je   pripojen a nenabiji se - vypadek napajeni napr. BILLA


#   kabel pripojen a charging INVALID
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'INVALID':
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =E2=9D=8C ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   kabel pripojen a charging ERROR
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'ERROR':
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   kabel pripojen a charging NOT_CHARGING
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'NOT_CHARGING' and chargingLevelHv > last_percent:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =E2=9A=A0=F0=9F=94=8C ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   kabel pripojen a charging FINISHED_FULLY_CHARGED
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'FINISHED_FULLY_CHARGED' and chargingLevelHv > last_percent:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =F0=9F=92=AF ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
    else:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = ''
        message_subjectEmoji = '=?utf-8?Q?  =F0=9F=94=B4=E2=9A=A0=F0=9F=94=8C ?='
        message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?='
##        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)

except Exception as msg:
    logging.error(msg)
    sys.exit(1)

pouziti:
[root@CL-ITNK-BMW ~]# /usr/local/bin/bmw-uni.py
Usage : /usr/local/bin/bmw-uni.py -v VINnumber
-v VIN_number
-u user
-p password
-t token
-w apiweatherkey
-h help
[root@CL-ITNK-BMW ~]#

Vydrz tyden vyvojari od bimmer_connct uz maji nove api rozhrani .... ja to pak upravim aby to fungovalo.
https://github.com/bimmerconnected/bimmer_connected/issues/325
https://github.com/bimmerconnected/bimmer_connected/discussions/327

Havrla · Příspěvekod **Havrla** » pát 12. lis 2021 15:44:22

Havrla píše:
Kód: Vybrat vše
[root@CL-ITNK-BMW ~]# cat /usr/local/bin/bmw-uni.py #! /usr/bin/env python3 import requests import json import urllib import logging import sys, os, time import pickle import pymysql # Import smtplib for the actual sending function import smtplib # Import the email modules we'll need from email.mime.text import MIMEText from getopt import getopt, GetoptError # API Gateway # North America: b2vapi.bmwgroup.us # Rest of World: b2vapi.bmwgroup.com # China: b2vapi.bmwgroup.cn:8592 BASE_URL = 'customer.bmwgroup.com' #BASE_URL = 'b2vapi.bmwgroup.com' def usage(): print("Usage : %s -v VINnumber" % sys.argv[0]) print("") print(" -v VIN_number") print(" -u user") print(" -p password") print(" -t token") print(" -w apiweatherkey") print(" -h help") #enddef apikey="" credentials={} credentials['access_token']="TOKEN" if len(sys.argv) > 1: try: opts,args = getopt(sys.argv[1:], "v:u:p:t:w:h") except GetoptError as err: print( "error:", err) sys.exit(1) #endtry for (arg,value) in opts: if (arg == "-v"): try: credentials['vin'] = str(value) except: print("ERROR parametru! -v vin_number") sys.exit(1) #endtry #endif if (arg == "-u"): try: credentials['username'] = str(value) except: print("ERROR parametru! -u username") sys.exit(1) #endtry #endif if (arg == "-p"): try: credentials['password'] = str(value) except: print("ERROR parametru! -p password") sys.exit(1) #endtry #endif if (arg == "-t"): try: credentials['access_token'] = str(value) except: print("ERROR parametru! -t TOKEN") sys.exit(1) #endtry #endif if (arg == "-w"): try: apikey = str(value) except: print("ERROR parametru! -w api_key_weather") sys.exit(1) #endtry #endif if (arg == "-h"): usage() sys.exit(1) #endif #endfor else: usage() sys.exit(1) #endif # 7 difit from VIN vin=credentials['vin'] vin7=vin[-7:] logging.basicConfig(filename="/var/log/bmw/bmw_%s.log" % vin7, filemode='a', format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s', datefmt='%H:%M:%S', level=logging.INFO) def getToken(BASE_URL, username, password, timeout=(20, 20)): try: headers = { "Content-Type": "application/x-www-form-urlencoded", "Content-Length": "124", "Connection": "Keep-Alive", "Host": BASE_URL, "Accept-Encoding": "gzip", "Authorization": "Basic blF2NkNxdHhKdVhXUDc0eGYzQ0p3VUVQOjF6REh4NnVuNGNEanli" "TEVOTjNreWZ1bVgya0VZaWdXUGNRcGR2RFJwSUJrN3JPSg==", "Credentials": "nQv6CqtxJuXWP74xf3CJwUEP:1zDHx6un4cDjybLENN3kyfumX2kEYigWPcQpdvDRpIBk7rOJ", "User-Agent": "okhttp/2.60"} data = { 'client_id': 'dbf0a542-ebd1-4ff0-a9a7-55172fbfce35', 'response_type': 'token', 'redirect_uri': 'https://www.bmw-connecteddrive.com/app/static/external-dispatch.html', 'scope': 'authenticate_user vehicle_data remote_services', 'username': username, 'password': password} data = urllib.parse.urlencode(data)# url = 'https://' + BASE_URL + '/webapi/oauth/token' # url = 'https://' + BASE_URL + '/webapi/oauth/token' url = 'https://' + BASE_URL + '/gcdm/oauth/authenticate' #print (url) r = requests.post(url, data=data, headers=headers, allow_redirects=False) #print (r) if r.status_code == 302: logging.info('Access token acquired') response_json = dict( urllib.parse.parse_qsl(urllib.parse.urlparse(r.headers['Location']).fragment) ) #print(response_json) #print(response_json['access_token']) return response_json['access_token'] else: raise ValueError('Unable to login') except Exception as msg: logging.error(msg) def getSocData(BASE_URL, token, vin): try: headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1', 'Authorization': 'Bearer ' + token} BASE_URLL = 'b2vapi.bmwgroup.com' url = 'https://' + BASE_URLL + '/api/vehicle/navigation/v1/' + vin # print(url) r = requests.get(url, headers=headers, timeout=(20, 20)) # print(r) # print(r.json()) if r.status_code == 200: status = r.json() if status.get('socmax'): socMax = status['socmax'] if status.get('socMax'): socMax = status['socMax'] positionlat = status['latitude'] positionlon = status['longitude'] soc = status['soc'] logging.info('SoC Data retreived successfully') return socMax, positionlat, positionlon, soc else: raise ValueError('Unable to retreive SoC data from vehicle') except Exception as msg: logging.error(msg) def getBattery(BASE_URL, token, vin): try: headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1', 'Authorization': 'Bearer ' + token} url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/status' r = requests.get(url, headers=headers, timeout=(20, 20)) #print(r.json()) if r.status_code == 200: status = r.json() chargingLevelHv = status['vehicleStatus']['chargingLevelHv'] remainingRangeElectric = status['vehicleStatus']['remainingRangeElectric'] connectionStatus = status['vehicleStatus']['connectionStatus'] chargingStatus = status['vehicleStatus']['chargingStatus'] mileage = status['vehicleStatus']['mileage'] positionlat = status['vehicleStatus']['position']['lat'] positionlon = status['vehicleStatus']['position']['lon'] try: remainingFuel = status['vehicleStatus']['remainingFuel'] except: remainingFuel = 0 try: remainingRangeFuel = status['vehicleStatus']['remainingRangeFuel'] except: remainingRangeFuel = 0 updateReason = status['vehicleStatus']['updateReason'] logging.info('Battery data retreived successfully') return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason else: # raise ValueError('Unable to retreive battery data from vehicle') chargingLevelHv = 0 remainingRangeElectric = 0 connectionStatus = "unknown" chargingStatus = "unknown" mileage = 0 positionlat = 49.83122600 positionlon = 14.69111800 remainingFuel = 0 remainingRangeFuel = 0 updateReason = 0 logging.info('Battery data retreived NOT successfully') return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason except Exception as msg: logging.error(msg) def getTrip(BASE_URL, token, vin): try: headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1', 'Authorization': 'Bearer ' + token} url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/statistics/lastTrip' r = requests.get(url, headers=headers, timeout=(20, 20)) #print(r.json()) if r.status_code == 200: status = r.json() avgElectricConsumption = status['lastTrip']['avgElectricConsumption'] avgRecuperation = status['lastTrip']['avgRecuperation'] totalDistance = status['lastTrip']['totalDistance'] electricDistance = status['lastTrip']['electricDistance'] duration = status['lastTrip']['duration'] logging.info('Trip data retreived successfully') return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration else: # raise ValueError('Unable to retreive Trip data from vehicle') avgElectricConsumption = 0 avgRecuperation = 0 totalDistance = 0 electricDistance = 0 duration = 0 logging.info('Trip data retreived NOT successfully') return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration except Exception as msg: logging.error(msg) # https://api.openweathermap.org/data/2.5/onecall?lat=49.84843&lon=14.704501&appid=07ee063f9e1248912cd522104a802244&exclude=minutely,hourly,daily,alerts&units=metric def getTeplota(positionlat, positionlon): try: url="https://api.openweathermap.org/data/2.5/onecall?lat="+str(positionlat)+"&lon="+str(positionlon)+"&appid="+apikey+"&exclude=minutely,hourly,daily,alerts&units=metric" headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1'} r = requests.get(url, headers=headers, timeout=(20, 20)) if r.status_code == 200: status = r.json() teplota = status['current']['temp'] logging.info('Teplota data retreived successfully') return teplota else: raise ValueError('Unable to retreive Teplota data from internet at car position.') except Exception as msg: logging.error(msg) def getPower(): try: soubor = open("/tmp/bmw_power_%s.txt" % vin7, mode='rt') vykon = soubor.read() soubor.close() except: vykon=0 return vykon ## TODO #def sendEMAIL(message_chargingLevelHv,message_connectionStatus,message_chargingStatus,message_subjectEmoji): # try: # logging.info(message_chargingLevelHv) # logging.info(message_connectionStatus) # logging.info(message_chargingStatus) # SMTP_SERVER = '10.27.23.12' # SMTP_PORT = 587 # GMAIL_USERNAME = 'xxx@xxx.cz' # GMAIL_PASSWORD = 'xxxx!' #CAUTION: This is stored in plain text! # recipient = 'xxx@xxxx.cz' # Message_chargingLevelHv=str(message_chargingLevelHv) ## message_subjectEmoji = '=?utf-8?Q? =F0=9F=9A=97 ?=' # subject = message_subjectEmoji + ' BMW ' + Message_chargingLevelHv + '% ' # emailText = 'pripojeni:' + message_connectionStatus + '<br>nabijeni:' + message_chargingStatus # headers = ["From: " + GMAIL_USERNAME, # "Subject: " + subject, # "To: " + recipient, # "MIME-Version: 1.0", # "Content-Type: text/html"] # headers = "\r\n".join(headers) # session = smtplib.SMTP(SMTP_SERVER, SMTP_PORT) # session.ehlo() # session.starttls() # session.ehlo # session.login(GMAIL_USERNAME, GMAIL_PASSWORD) # session.sendmail(GMAIL_USERNAME, recipient, headers + "\r\n\r\n" + emailText) # session.quit() # logging.info('EMAIL message sent successfully') # except Exception as msg: # logging.error(msg) # sys.exit(1) # def getLastPercent(): try: with open ("/var/spool/bmw/last_percent_%s" % vin7 , 'rb') as lp: last_percent = pickle.load(lp) logging.info(f'last_percent loaded with a value of {last_percent}') return last_percent except FileNotFoundError: logging.error('last_percent file not found') with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp: pickle.dump(0, lp) logging.info('last_percent file created with value of 0') with open ("/var/spool/bmw/last_percent_%s" % vin7 , 'rb') as lp: last_percent = pickle.load(lp) return last_percent def setLastPercent(chargingLevelHv): try: with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp: pickle.dump(chargingLevelHv, lp) logging.info(f'last_percent file updated with {chargingLevelHv}') except Exception as msg: logging.error(msg) sys.exit(1) try: last_percent = int(getLastPercent()) try: token = getToken(BASE_URL, credentials['username'], credentials['password']) except Exception as msg: logging.error(msg) sys.exit(1) try: chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason = getBattery(BASE_URL, token, credentials['vin']) except Exception as msg: logging.error(msg) sys.exit(1) try: socMax, positionlat, positionlon, chargingLevelHv = getSocData(BASE_URL, token, credentials['vin']) except Exception as msg: logging.error(msg) sys.exit(1) try: avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration = getTrip(BASE_URL, token, credentials['vin']) except Exception as msg: logging.error(msg) sys.exit(1) try: teplota = getTeplota(positionlat, positionlon) except Exception as msg: logging.error(msg) teplota = 1 try: CharegPower = getPower() except Exception as msg: logging.error(msg) CharegPower = 0 message = 'BMW i3 Status:\n' + f'Battery: {chargingLevelHv}% ({remainingRangeElectric} km) - {connectionStatus}/{chargingStatus}\nMileage: {mileage} km\nPosition http://maps.apple.com/?ll={positionlat},{positionlon}\n' + f'socMax: {socMax} kWh\n' conn = pymysql.connect(host='127.0.0.1', unix_socket='/var/lib/mysql/mysql.sock', user='bmw', passwd='xxxxxxxx', db="bmw_%s" % vin7) cur = conn.cursor() try: cur.execute('''SELECT mileage FROM i3 WHERE chargingStatus = "CHARGING" ORDER BY datum DESC LIMIT 1''') mileageCH= cur.fetchone() try: DistanceSinceLastCharge = mileage - mileageCH[0] except: DistanceSinceLastCharge = 0 cur.execute('''SELECT mileage FROM i3 WHERE chargingStatus = "FINISHED_FULLY_CHARGED" ORDER BY datum DESC LIMIT 1''') mileageCH= cur.fetchone() try: DistanceSinceLastFCharg = mileage - mileageCH[0] except: DistanceSinceLastFCharg = 0 except pymysql.IntegrityError: logging.warn("failed to select values") try: affected_count = cur.execute('''INSERT INTO i3 (mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)''',(mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg)) conn.commit() logging.info("inserted values to DB") except pymysql.IntegrityError: logging.warn("failed to insert values") finally: cur.close() conn.close() # nabijeni poprve dosalo 100 procent (mame plne nabito) if chargingLevelHv > last_percent and chargingLevelHv == 100: setLastPercent(chargingLevelHv) message_subjectEmoji = '' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # blizime se uz nabitemu stavu elif chargingLevelHv > last_percent and chargingLevelHv > 83: setLastPercent(chargingLevelHv) message_subjectEmoji = '' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # chargingStatus # INVALID # ERROR # NOT_CHARGING # CHARGING # FINISHED_FULLY_CHARGED # connectionStatus # CONNECTED # DISCONNECTED # DISCONNECTED INVALID vuz neni pripojen a nenabiji se - nejcasteji # CONNECTED NOT_CHARGING vuz je pripojen a nenabiji se - vypadek napajeni napr. BILLA # kabel pripojen a charging INVALID elif connectionStatus == 'CONNECTED' and chargingStatus == 'INVALID': setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =E2=9D=8C ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # kabel pripojen a charging ERROR elif connectionStatus == 'CONNECTED' and chargingStatus == 'ERROR': setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # kabel pripojen a charging NOT_CHARGING elif connectionStatus == 'CONNECTED' and chargingStatus == 'NOT_CHARGING' and chargingLevelHv > last_percent: setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =E2=9A=A0=F0=9F=94=8C ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # kabel pripojen a charging FINISHED_FULLY_CHARGED elif connectionStatus == 'CONNECTED' and chargingStatus == 'FINISHED_FULLY_CHARGED' and chargingLevelHv > last_percent: setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =F0=9F=92=AF ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) else: setLastPercent(chargingLevelHv) message_subjectEmoji = '' message_subjectEmoji = '=?utf-8?Q? =F0=9F=94=B4=E2=9A=A0=F0=9F=94=8C ?=' message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?=' ## sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) except Exception as msg: logging.error(msg) sys.exit(1)

pouziti:
[root@CL-ITNK-BMW ~]# /usr/local/bin/bmw-uni.py
Usage : /usr/local/bin/bmw-uni.py -v VINnumber
-v VIN_number
-u user
-p password
-t token
-w apiweatherkey
-h help
[root@CL-ITNK-BMW ~]#
/usr/local/bin/bmw-uni.py -v WBY1Z81050VXXXXX -u "userlogin" -p "tajneheslo" -w "07ee063f912348912cd556704a802299"

Vydrz tyden vyvojari od bimmer_connct uz maji nove api rozhrani .... ja to pak upravim aby to fungovalo.
https://github.com/bimmerconnected/bimmer_connected/issues/325
https://github.com/bimmerconnected/bimmer_connected/discussions/327

Havrla · Příspěvekod **Havrla** » pát 12. lis 2021 15:47:11

Havrla píše:
Havrla píše:
Kód: Vybrat vše
[root@CL-ITNK-BMW ~]# cat /usr/local/bin/bmw-uni.py #! /usr/bin/env python3 import requests import json import urllib import logging import sys, os, time import pickle import pymysql # Import smtplib for the actual sending function import smtplib # Import the email modules we'll need from email.mime.text import MIMEText from getopt import getopt, GetoptError # API Gateway # North America: b2vapi.bmwgroup.us # Rest of World: b2vapi.bmwgroup.com # China: b2vapi.bmwgroup.cn:8592 BASE_URL = 'customer.bmwgroup.com' #BASE_URL = 'b2vapi.bmwgroup.com' def usage(): print("Usage : %s -v VINnumber" % sys.argv[0]) print("") print(" -v VIN_number") print(" -u user") print(" -p password") print(" -t token") print(" -w apiweatherkey") print(" -h help") #enddef apikey="" credentials={} credentials['access_token']="TOKEN" if len(sys.argv) > 1: try: opts,args = getopt(sys.argv[1:], "v:u:p:t:w:h") except GetoptError as err: print( "error:", err) sys.exit(1) #endtry for (arg,value) in opts: if (arg == "-v"): try: credentials['vin'] = str(value) except: print("ERROR parametru! -v vin_number") sys.exit(1) #endtry #endif if (arg == "-u"): try: credentials['username'] = str(value) except: print("ERROR parametru! -u username") sys.exit(1) #endtry #endif if (arg == "-p"): try: credentials['password'] = str(value) except: print("ERROR parametru! -p password") sys.exit(1) #endtry #endif if (arg == "-t"): try: credentials['access_token'] = str(value) except: print("ERROR parametru! -t TOKEN") sys.exit(1) #endtry #endif if (arg == "-w"): try: apikey = str(value) except: print("ERROR parametru! -w api_key_weather") sys.exit(1) #endtry #endif if (arg == "-h"): usage() sys.exit(1) #endif #endfor else: usage() sys.exit(1) #endif # 7 difit from VIN vin=credentials['vin'] vin7=vin[-7:] logging.basicConfig(filename="/var/log/bmw/bmw_%s.log" % vin7, filemode='a', format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s', datefmt='%H:%M:%S', level=logging.INFO) def getToken(BASE_URL, username, password, timeout=(20, 20)): try: headers = { "Content-Type": "application/x-www-form-urlencoded", "Content-Length": "124", "Connection": "Keep-Alive", "Host": BASE_URL, "Accept-Encoding": "gzip", "Authorization": "Basic blF2NkNxdHhKdVhXUDc0eGYzQ0p3VUVQOjF6REh4NnVuNGNEanli" "TEVOTjNreWZ1bVgya0VZaWdXUGNRcGR2RFJwSUJrN3JPSg==", "Credentials": "nQv6CqtxJuXWP74xf3CJwUEP:1zDHx6un4cDjybLENN3kyfumX2kEYigWPcQpdvDRpIBk7rOJ", "User-Agent": "okhttp/2.60"} data = { 'client_id': 'dbf0a542-ebd1-4ff0-a9a7-55172fbfce35', 'response_type': 'token', 'redirect_uri': 'https://www.bmw-connecteddrive.com/app/static/external-dispatch.html', 'scope': 'authenticate_user vehicle_data remote_services', 'username': username, 'password': password} data = urllib.parse.urlencode(data)# url = 'https://' + BASE_URL + '/webapi/oauth/token' # url = 'https://' + BASE_URL + '/webapi/oauth/token' url = 'https://' + BASE_URL + '/gcdm/oauth/authenticate' #print (url) r = requests.post(url, data=data, headers=headers, allow_redirects=False) #print (r) if r.status_code == 302: logging.info('Access token acquired') response_json = dict( urllib.parse.parse_qsl(urllib.parse.urlparse(r.headers['Location']).fragment) ) #print(response_json) #print(response_json['access_token']) return response_json['access_token'] else: raise ValueError('Unable to login') except Exception as msg: logging.error(msg) def getSocData(BASE_URL, token, vin): try: headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1', 'Authorization': 'Bearer ' + token} BASE_URLL = 'b2vapi.bmwgroup.com' url = 'https://' + BASE_URLL + '/api/vehicle/navigation/v1/' + vin # print(url) r = requests.get(url, headers=headers, timeout=(20, 20)) # print(r) # print(r.json()) if r.status_code == 200: status = r.json() if status.get('socmax'): socMax = status['socmax'] if status.get('socMax'): socMax = status['socMax'] positionlat = status['latitude'] positionlon = status['longitude'] soc = status['soc'] logging.info('SoC Data retreived successfully') return socMax, positionlat, positionlon, soc else: raise ValueError('Unable to retreive SoC data from vehicle') except Exception as msg: logging.error(msg) def getBattery(BASE_URL, token, vin): try: headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1', 'Authorization': 'Bearer ' + token} url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/status' r = requests.get(url, headers=headers, timeout=(20, 20)) #print(r.json()) if r.status_code == 200: status = r.json() chargingLevelHv = status['vehicleStatus']['chargingLevelHv'] remainingRangeElectric = status['vehicleStatus']['remainingRangeElectric'] connectionStatus = status['vehicleStatus']['connectionStatus'] chargingStatus = status['vehicleStatus']['chargingStatus'] mileage = status['vehicleStatus']['mileage'] positionlat = status['vehicleStatus']['position']['lat'] positionlon = status['vehicleStatus']['position']['lon'] try: remainingFuel = status['vehicleStatus']['remainingFuel'] except: remainingFuel = 0 try: remainingRangeFuel = status['vehicleStatus']['remainingRangeFuel'] except: remainingRangeFuel = 0 updateReason = status['vehicleStatus']['updateReason'] logging.info('Battery data retreived successfully') return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason else: # raise ValueError('Unable to retreive battery data from vehicle') chargingLevelHv = 0 remainingRangeElectric = 0 connectionStatus = "unknown" chargingStatus = "unknown" mileage = 0 positionlat = 49.83122600 positionlon = 14.69111800 remainingFuel = 0 remainingRangeFuel = 0 updateReason = 0 logging.info('Battery data retreived NOT successfully') return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason except Exception as msg: logging.error(msg) def getTrip(BASE_URL, token, vin): try: headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1', 'Authorization': 'Bearer ' + token} url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/statistics/lastTrip' r = requests.get(url, headers=headers, timeout=(20, 20)) #print(r.json()) if r.status_code == 200: status = r.json() avgElectricConsumption = status['lastTrip']['avgElectricConsumption'] avgRecuperation = status['lastTrip']['avgRecuperation'] totalDistance = status['lastTrip']['totalDistance'] electricDistance = status['lastTrip']['electricDistance'] duration = status['lastTrip']['duration'] logging.info('Trip data retreived successfully') return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration else: # raise ValueError('Unable to retreive Trip data from vehicle') avgElectricConsumption = 0 avgRecuperation = 0 totalDistance = 0 electricDistance = 0 duration = 0 logging.info('Trip data retreived NOT successfully') return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration except Exception as msg: logging.error(msg) # https://api.openweathermap.org/data/2.5/onecall?lat=49.84843&lon=14.704501&appid=07ee063f9e1248912cd522104a802244&exclude=minutely,hourly,daily,alerts&units=metric def getTeplota(positionlat, positionlon): try: url="https://api.openweathermap.org/data/2.5/onecall?lat="+str(positionlat)+"&lon="+str(positionlon)+"&appid="+apikey+"&exclude=minutely,hourly,daily,alerts&units=metric" headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1'} r = requests.get(url, headers=headers, timeout=(20, 20)) if r.status_code == 200: status = r.json() teplota = status['current']['temp'] logging.info('Teplota data retreived successfully') return teplota else: raise ValueError('Unable to retreive Teplota data from internet at car position.') except Exception as msg: logging.error(msg) def getPower(): try: soubor = open("/tmp/bmw_power_%s.txt" % vin7, mode='rt') vykon = soubor.read() soubor.close() except: vykon=0 return vykon ## TODO #def sendEMAIL(message_chargingLevelHv,message_connectionStatus,message_chargingStatus,message_subjectEmoji): # try: # logging.info(message_chargingLevelHv) # logging.info(message_connectionStatus) # logging.info(message_chargingStatus) # SMTP_SERVER = '10.27.23.12' # SMTP_PORT = 587 # GMAIL_USERNAME = 'xxxxx@xxxxt.cz' # GMAIL_PASSWORD = 'XXXXX!' #CAUTION: This is stored in plain text! # recipient = 'xxxx@xxxx.cz' # Message_chargingLevelHv=str(message_chargingLevelHv) ## message_subjectEmoji = '=?utf-8?Q? =F0=9F=9A=97 ?=' # subject = message_subjectEmoji + ' BMW ' + Message_chargingLevelHv + '% ' # emailText = 'pripojeni:' + message_connectionStatus + '<br>nabijeni:' + message_chargingStatus # headers = ["From: " + GMAIL_USERNAME, # "Subject: " + subject, # "To: " + recipient, # "MIME-Version: 1.0", # "Content-Type: text/html"] # headers = "\r\n".join(headers) # session = smtplib.SMTP(SMTP_SERVER, SMTP_PORT) # session.ehlo() # session.starttls() # session.ehlo # session.login(GMAIL_USERNAME, GMAIL_PASSWORD) # session.sendmail(GMAIL_USERNAME, recipient, headers + "\r\n\r\n" + emailText) # session.quit() # logging.info('EMAIL message sent successfully') # except Exception as msg: # logging.error(msg) # sys.exit(1) # def getLastPercent(): try: with open ("/var/spool/bmw/last_percent_%s" % vin7 , 'rb') as lp: last_percent = pickle.load(lp) logging.info(f'last_percent loaded with a value of {last_percent}') return last_percent except FileNotFoundError: logging.error('last_percent file not found') with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp: pickle.dump(0, lp) logging.info('last_percent file created with value of 0') with open ("/var/spool/bmw/last_percent_%s" % vin7 , 'rb') as lp: last_percent = pickle.load(lp) return last_percent def setLastPercent(chargingLevelHv): try: with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp: pickle.dump(chargingLevelHv, lp) logging.info(f'last_percent file updated with {chargingLevelHv}') except Exception as msg: logging.error(msg) sys.exit(1) try: last_percent = int(getLastPercent()) try: token = getToken(BASE_URL, credentials['username'], credentials['password']) except Exception as msg: logging.error(msg) sys.exit(1) try: chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason = getBattery(BASE_URL, token, credentials['vin']) except Exception as msg: logging.error(msg) sys.exit(1) try: socMax, positionlat, positionlon, chargingLevelHv = getSocData(BASE_URL, token, credentials['vin']) except Exception as msg: logging.error(msg) sys.exit(1) try: avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration = getTrip(BASE_URL, token, credentials['vin']) except Exception as msg: logging.error(msg) sys.exit(1) try: teplota = getTeplota(positionlat, positionlon) except Exception as msg: logging.error(msg) teplota = 1 try: CharegPower = getPower() except Exception as msg: logging.error(msg) CharegPower = 0 message = 'BMW i3 Status:\n' + f'Battery: {chargingLevelHv}% ({remainingRangeElectric} km) - {connectionStatus}/{chargingStatus}\nMileage: {mileage} km\nPosition http://maps.apple.com/?ll={positionlat},{positionlon}\n' + f'socMax: {socMax} kWh\n' conn = pymysql.connect(host='127.0.0.1', unix_socket='/var/lib/mysql/mysql.sock', user='bmw', passwd='xxxxxxxx', db="bmw_%s" % vin7) cur = conn.cursor() try: cur.execute('''SELECT mileage FROM i3 WHERE chargingStatus = "CHARGING" ORDER BY datum DESC LIMIT 1''') mileageCH= cur.fetchone() try: DistanceSinceLastCharge = mileage - mileageCH[0] except: DistanceSinceLastCharge = 0 cur.execute('''SELECT mileage FROM i3 WHERE chargingStatus = "FINISHED_FULLY_CHARGED" ORDER BY datum DESC LIMIT 1''') mileageCH= cur.fetchone() try: DistanceSinceLastFCharg = mileage - mileageCH[0] except: DistanceSinceLastFCharg = 0 except pymysql.IntegrityError: logging.warn("failed to select values") try: affected_count = cur.execute('''INSERT INTO i3 (mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)''',(mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg)) conn.commit() logging.info("inserted values to DB") except pymysql.IntegrityError: logging.warn("failed to insert values") finally: cur.close() conn.close() # nabijeni poprve dosalo 100 procent (mame plne nabito) if chargingLevelHv > last_percent and chargingLevelHv == 100: setLastPercent(chargingLevelHv) message_subjectEmoji = '' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # blizime se uz nabitemu stavu elif chargingLevelHv > last_percent and chargingLevelHv > 83: setLastPercent(chargingLevelHv) message_subjectEmoji = '' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # chargingStatus # INVALID # ERROR # NOT_CHARGING # CHARGING # FINISHED_FULLY_CHARGED # connectionStatus # CONNECTED # DISCONNECTED # DISCONNECTED INVALID vuz neni pripojen a nenabiji se - nejcasteji # CONNECTED NOT_CHARGING vuz je pripojen a nenabiji se - vypadek napajeni napr. BILLA # kabel pripojen a charging INVALID elif connectionStatus == 'CONNECTED' and chargingStatus == 'INVALID': setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =E2=9D=8C ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # kabel pripojen a charging ERROR elif connectionStatus == 'CONNECTED' and chargingStatus == 'ERROR': setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # kabel pripojen a charging NOT_CHARGING elif connectionStatus == 'CONNECTED' and chargingStatus == 'NOT_CHARGING' and chargingLevelHv > last_percent: setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =E2=9A=A0=F0=9F=94=8C ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) # kabel pripojen a charging FINISHED_FULLY_CHARGED elif connectionStatus == 'CONNECTED' and chargingStatus == 'FINISHED_FULLY_CHARGED' and chargingLevelHv > last_percent: setLastPercent(chargingLevelHv) message_subjectEmoji = '=?utf-8?Q? =F0=9F=92=AF ?=' # sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) else: setLastPercent(chargingLevelHv) message_subjectEmoji = '' message_subjectEmoji = '=?utf-8?Q? =F0=9F=94=B4=E2=9A=A0=F0=9F=94=8C ?=' message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?=' ## sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji) logging.info(message) except Exception as msg: logging.error(msg) sys.exit(1)

pouziti:
[root@CL-ITNK-BMW ~]# /usr/local/bin/bmw-uni.py
Usage : /usr/local/bin/bmw-uni.py -v VINnumber
-v VIN_number
-u user
-p password
-t token
-w apiweatherkey
-h help
[root@CL-ITNK-BMW ~]#

[root@CL-ITNK-BMW ~]# /usr/local/bin/bmw-uni-test.py -v WBY1Z21020V607314 -u "mujlogin" -p "mojeheslo" -w "mujpocasovyticket"
BATT:
SOC:
{'auxPowerEcoPro': 1.0, 'auxPowerEcoProPlus': 1.0, 'auxPowerRegular': 1.0, 'latitude': 49.231226, 'longitude': 14.791136, 'pendingUpdate': False, 'soc': 93.0, 'socmax': 38.52, 'vehicleTracking': True}
Trip:
BMW i3 Status:
Battery: 93.0% (0 km) - unknown/unknown
Mileage: 0 km
Position http://maps.apple.com/?ll=49.231226,14.791136
socMax: 38.52 kWh

[root@CL-ITNK-BMW ~]#

Vydrz tyden vyvojari od bimmer_connct uz maji nove api rozhrani .... ja to pak upravim aby to fungovalo.
https://github.com/bimmerconnected/bimmer_connected/issues/325
https://github.com/bimmerconnected/bimmer_connected/discussions/327

stejk · Příspěvekod **stejk** » sob 13. lis 2021 7:19:24

Díky. Počkáme s čím přijdou vývojáři

stejk · Příspěvekod **stejk** » úte 23. lis 2021 7:57:07

Prý pujde vyčítat socMax, ale priorita to není
https://github.com/bimmerconnected/bimmer_connected/discussions/361

Havrla · Příspěvekod **Havrla** » úte 23. lis 2021 10:50:34

V klidu....

k aktualimu skryptu co mame si priinstalni :
pip3 install --upgrade bimmer_connected

muj novy vzhledem k moznostem jakstaks skrypt

Kód: Vybrat vše

[root@CL-ITNK-BMW ~]# cat /usr/local/bin/bmw-uni.py
#! /usr/bin/env python3

import requests
import json
import urllib
import logging
import sys, os, time
import pickle
import pymysql
# Import smtplib for the actual sending function
import smtplib
# Import the email modules we'll need
from email.mime.text import MIMEText
from getopt import getopt, GetoptError

import bimmer_connected
from bimmer_connected import country_selector
from bimmer_connected import account
from bimmer_connected.utils import to_json



#   API Gateway
#   North America: b2vapi.bmwgroup.us
#   Rest of World: b2vapi.bmwgroup.com
#   China: b2vapi.bmwgroup.cn:8592
BASE_URL = 'customer.bmwgroup.com'
#BASE_URL = 'b2vapi.bmwgroup.com'

def usage():
    print("Usage : %s -v VINnumber" % sys.argv[0])
    print("")
    print("     -v VIN_number")
    print("     -u user")                                                                                                                                                                                                                    
    print("     -p password")                                                                                                                                                                                                                
    print("     -t token")                                                                                                                                                                                                                   
    print("     -w apiweatherkey")                                                                                                                                                                                                           
    print("     -h help")                                                                                                                                                                                                                    
#enddef                                                                                                                                                                                                                                      
                                                                                                                                                                                                                                             
apikey=""                                                                                                                                                                                                                                    
credentials={}                                                                                                                                                                                                                               
credentials['access_token']="TOKEN"                                                                                                                                                                                                          
                                                                                                                                                                                                                                             

if len(sys.argv) > 1:
    try:
        opts,args = getopt(sys.argv[1:], "v:u:p:t:w:h")
    except GetoptError as err:
        print( "error:",  err)
        sys.exit(1)
    #endtry

    for (arg,value) in opts:
        if (arg == "-v"):
            try:
                credentials['vin'] = str(value)
            except:
                print("ERROR parametru! -v vin_number")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-u"):
            try:
                credentials['username'] = str(value)
            except: 
                print("ERROR parametru! -u username")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-p"):
            try:
                credentials['password'] = str(value)
            except: 
                print("ERROR parametru! -p password")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-t"):
            try:
                credentials['access_token'] = str(value)
            except:
                print("ERROR parametru! -t TOKEN")
                sys.exit(1)
            #endtry
        #endif
        if (arg == "-w"):
            try:
                apikey = str(value)
            except:
                print("ERROR parametru! -w api_key_weather")
                sys.exit(1)
            #endtry
        #endif




        if (arg == "-h"):
            usage()
            sys.exit(1)
        #endif
    #endfor
else:
    usage()
    sys.exit(1)
#endif

# 7 difit from VIN
vin=credentials['vin']
vin7=vin[-7:]

logging.basicConfig(filename="/var/log/bmw/bmw_%s.log" % vin7,
                            filemode='a',
                            format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
                            datefmt='%H:%M:%S',
                            level=logging.INFO)


def getToken(BASE_URL, username, password, timeout=(20, 20)):
    try:
        headers = {
                "Content-Type": "application/x-www-form-urlencoded",
                "Content-Length": "124",
                "Connection": "Keep-Alive",
                "Host": BASE_URL,
                "Accept-Encoding": "gzip",
                "Authorization": "Basic blF2NkNxdHhKdVhXUDc0eGYzQ0p3VUVQOjF6REh4NnVuNGNEanli"
                                 "TEVOTjNreWZ1bVgya0VZaWdXUGNRcGR2RFJwSUJrN3JPSg==",
                "Credentials": "nQv6CqtxJuXWP74xf3CJwUEP:1zDHx6un4cDjybLENN3kyfumX2kEYigWPcQpdvDRpIBk7rOJ",
                "User-Agent": "okhttp/2.60"}

        data = {
            'client_id': 'dbf0a542-ebd1-4ff0-a9a7-55172fbfce35',
            'response_type': 'token',
            'redirect_uri': 'https://www.bmw-connecteddrive.com/app/static/external-dispatch.html',
            'scope': 'authenticate_user vehicle_data remote_services',
            'username': username,
            'password': password}

        data = urllib.parse.urlencode(data)#       url = 'https://' + BASE_URL + '/webapi/oauth/token'
#        url = 'https://' + BASE_URL + '/webapi/oauth/token'
        url = 'https://' + BASE_URL + '/gcdm/oauth/authenticate'
        #print (url)
        r = requests.post(url, data=data, headers=headers, allow_redirects=False)
        #print (r)
        if r.status_code == 302:
            logging.info('Access token acquired')
            response_json = dict(
                    urllib.parse.parse_qsl(urllib.parse.urlparse(r.headers['Location']).fragment)
                )
            #print(response_json)
            #print(response_json['access_token'])
            return response_json['access_token']
        else:
            raise ValueError('Unable to login')

    except Exception as msg:
        logging.error(msg)


def getSocData(BASE_URL, token, vin):
    try:
        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1',
                    'Authorization': 'Bearer ' + token}

        BASE_URLL = 'b2vapi.bmwgroup.com'
        url = 'https://' + BASE_URLL + '/api/vehicle/navigation/v1/' + vin

#        print(url)
        r = requests.get(url, headers=headers, timeout=(20, 20))
#        print(r)
#        print(r.json())
        if r.status_code == 200:
            status = r.json()
            if status.get('socmax'):
                 socMax = status['socmax']
            if status.get('socMax'):
                 socMax = status['socMax']
            logging.info('SoC Data retreived successfully')
            return socMax
        else:
            raise ValueError('Unable to retreive SoC data from vehicle')

    except Exception as msg:
        logging.error(msg)


def getBattery(username, password, vin):
    try:
        czech=bimmer_connected.country_selector.Regions(2)
        account=bimmer_connected.account.ConnectedDriveAccount(username,password,czech,None,5)
        vehicle=account.get_vehicle(vin)

        chargingLevelHv=vehicle.status.charging_level_hv
        remainingRangeElectric=vehicle.status.remaining_range_electric[0]
        connectionStatus=vehicle.status.connection_status
        chargingStatus=vehicle.status.charging_status.split('.')[0]
        if chargingStatus == "COMPLETE" : 
            chargingStatus = "FINISHED_FULLY_CHARGED" 
        elif chargingStatus == "NOT_CHARGING" :
            chargingStatus = "INVALID" 

        mileage=vehicle.status.mileage[0]
        positionlat, positionlon = vehicle.status.gps_position
        try:
            remainingFuel=vehicle.status.remaining_fuel[0]
        except:
            remainingFuel=0
        try:
            remainingRangeFuel=vehicle.status.remaining_range_fuel[0]
        except:
            remainingRangeFuel=0
        updateReason=0





        logging.info('Battery data retreived successfully')
        return chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason


    except Exception as msg:
        logging.error(msg)

#def getTrip(BASE_URL, token, vin):
#    try:
#        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1',
#                    'Authorization': 'Bearer ' + token}
#
#        url = 'https://' + BASE_URL + '/webapi/v1/user/vehicles/' + vin + '/statistics/lastTrip'
#
#        r = requests.get(url, headers=headers, timeout=(20, 20))
#        print("Trip:")
#        try:
#            print(r.json())
#        except:
#            pass
#        if r.status_code == 200:
#            status = r.json()
#            avgElectricConsumption = status['lastTrip']['avgElectricConsumption']
#            avgRecuperation = status['lastTrip']['avgRecuperation']
#            totalDistance = status['lastTrip']['totalDistance']
#            electricDistance = status['lastTrip']['electricDistance']
#            duration = status['lastTrip']['duration']
#            logging.info('Trip data retreived successfully') 
#            return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration
#
#        else:
##            raise ValueError('Unable to retreive Trip data from vehicle')
#            avgElectricConsumption = 0
#            avgRecuperation = 0
#            totalDistance = 0
#            electricDistance = 0
#            duration = 0
#            logging.info('Trip data retreived  NOT  successfully')
#            return avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration
#
#    except Exception as msg:
#        logging.error(msg)
#


#  https://api.openweathermap.org/data/2.5/onecall?lat=49.84843&lon=14.704501&appid=07ee063f9e1248912cd522104a802244&exclude=minutely,hourly,daily,alerts&units=metric

def getTeplota(positionlat, positionlon):
    try:
        url="https://api.openweathermap.org/data/2.5/onecall?lat="+str(positionlat)+"&lon="+str(positionlon)+"&appid="+apikey+"&exclude=minutely,hourly,daily,alerts&units=metric"
        headers = {'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_1_1 like Mac OS X) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0 Mobile/15B150 Safari/604.1'}
        r = requests.get(url, headers=headers, timeout=(20, 20))
        if r.status_code == 200:
            status = r.json()
            teplota = status['current']['temp']
            logging.info('Teplota data retreived successfully')
            return teplota

        else:
            raise ValueError('Unable to retreive Teplota data from internet at car position.')

    except Exception as msg:
        logging.error(msg)


def getPower():
    try:
        soubor = open("/tmp/bmw_power_%s.txt" % vin7, mode='rt')
        vykon = soubor.read()
        soubor.close()
    except:
        vykon=0
    return vykon

## TODO
#def sendEMAIL(message_chargingLevelHv,message_connectionStatus,message_chargingStatus,message_subjectEmoji):
#    try:
#        logging.info(message_chargingLevelHv)
#        logging.info(message_connectionStatus)
#        logging.info(message_chargingStatus)
#        SMTP_SERVER = '10.27.23.12'
#        SMTP_PORT = 587
#        GMAIL_USERNAME = 'xxx@xxx.cz'
#        GMAIL_PASSWORD = 'xxxxxxxxx!' #CAUTION: This is stored in plain text!
#        recipient = 'xxx@xxxx.cz'
#        Message_chargingLevelHv=str(message_chargingLevelHv)
##        message_subjectEmoji = '=?utf-8?Q? =F0=9F=9A=97 ?='
#        subject = message_subjectEmoji + ' BMW ' + Message_chargingLevelHv + '% '
#        emailText = 'pripojeni:' + message_connectionStatus + '<br>nabijeni:' + message_chargingStatus
#        headers = ["From: " + GMAIL_USERNAME,
#                "Subject: " + subject,
#                "To: " + recipient,
#                "MIME-Version: 1.0",
#                "Content-Type: text/html"]
#        headers = "\r\n".join(headers)
#        session = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
#        session.ehlo()
#        session.starttls()
#        session.ehlo
#        session.login(GMAIL_USERNAME, GMAIL_PASSWORD)
#        session.sendmail(GMAIL_USERNAME, recipient, headers + "\r\n\r\n" + emailText)
#        session.quit()
#        logging.info('EMAIL message sent successfully')
#    except Exception as msg:
#        logging.error(msg)
#        sys.exit(1)
#

def getLastPercent():
    try:
        with open ("/var/spool/bmw/last_percent_%s" % vin7  , 'rb') as lp:
            last_percent = pickle.load(lp)
        logging.info(f'last_percent loaded with a value of {last_percent}')
        return last_percent
    except FileNotFoundError:
        logging.error('last_percent file not found')
        with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp:
            pickle.dump(0, lp)
        logging.info('last_percent file created with value of 0')
        with open ("/var/spool/bmw/last_percent_%s" % vin7 , 'rb') as lp:
            last_percent = pickle.load(lp)
        return last_percent

def setLastPercent(chargingLevelHv):
    try:
        with open("/var/spool/bmw/last_percent_%s" % vin7 , 'wb') as lp:
            pickle.dump(chargingLevelHv, lp)
        logging.info(f'last_percent file updated with {chargingLevelHv}')
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)


try:
    last_percent = int(getLastPercent())

    try:
        token = getToken(BASE_URL, credentials['username'], credentials['password'])
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)

    try:
         chargingLevelHv, remainingRangeElectric, connectionStatus, chargingStatus, mileage, positionlat, positionlon, remainingFuel, remainingRangeFuel, updateReason = getBattery(credentials['username'], credentials['password'], credentials['vin'])
    except Exception as msg:
         logging.error(msg) 
         sys.exit(1)   

    try:        
        socMax  = getSocData(BASE_URL, token, credentials['vin'])
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)

    try:
#        avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration = getTrip(BASE_URL, token, credentials['vin'])
         avgElectricConsumption, avgRecuperation, totalDistance, electricDistance, duration = (0,0,0,0,0) 
    except Exception as msg:
        logging.error(msg)
        sys.exit(1)


    try:    
        teplota = getTeplota(positionlat, positionlon)
    except Exception as msg:
        logging.error(msg)
        teplota = 1

    try:
        CharegPower = getPower()
    except Exception as msg:
        logging.error(msg)
        CharegPower = 0


    message = 'BMW i3 Status:\n' + f'Battery: {chargingLevelHv}% ({remainingRangeElectric} km) - {connectionStatus}/{chargingStatus}\nMileage: {mileage} km\nPosition http://maps.apple.com/?ll={positionlat},{positionlon}\n' + f'socMax: {socMax} kWh\n'






    conn = pymysql.connect(host='127.0.0.1', unix_socket='/var/lib/mysql/mysql.sock', user='bmw', passwd='xxxxxxxxx', db="bmw_%s" % vin7)
    cur = conn.cursor()

    try:
        cur.execute('''SELECT mileage FROM i3  WHERE chargingStatus = "CHARGING" ORDER BY datum DESC LIMIT 1''')
        mileageCH= cur.fetchone()
        try:
            DistanceSinceLastCharge = mileage - mileageCH[0]
        except:
            DistanceSinceLastCharge = 0
        cur.execute('''SELECT mileage FROM i3  WHERE chargingStatus = "FINISHED_FULLY_CHARGED" ORDER BY datum DESC LIMIT 1''')
        mileageCH= cur.fetchone()
        try:
            DistanceSinceLastFCharg = mileage - mileageCH[0]
        except:
            DistanceSinceLastFCharg = 0 

    except pymysql.IntegrityError:
        logging.warn("failed to select values")



    try:
        affected_count = cur.execute('''INSERT INTO i3 (mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)''',(mileage,positionlat,positionlon,chargingLevelHv,remainingRangeElectric,socMax,connectionStatus,chargingStatus,avgElectricConsumption,avgRecuperation,totalDistance,electricDistance,duration,teplota,remainingFuel,remainingRangeFuel,CharegPower,updateReason,DistanceSinceLastCharge,DistanceSinceLastFCharg))
        conn.commit()
        logging.info("inserted values to DB")
    except pymysql.IntegrityError:
        logging.warn("failed to insert values")
    finally:
        cur.close()
        conn.close()


#   nabijeni poprve dosalo 100 procent (mame plne nabito)
    if chargingLevelHv > last_percent and chargingLevelHv == 100:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = ''
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   blizime se uz nabitemu stavu
    elif chargingLevelHv > last_percent and chargingLevelHv > 83:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = ''
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)

# chargingStatus
# INVALID
# ERROR
# NOT_CHARGING
# CHARGING
# FINISHED_FULLY_CHARGED

# connectionStatus
# CONNECTED
# DISCONNECTED

# DISCONNECTED INVALID      vuz neni pripojen a nenabiji se - nejcasteji
# CONNECTED    NOT_CHARGING vuz je   pripojen a nenabiji se - vypadek napajeni napr. BILLA


#   kabel pripojen a charging INVALID
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'INVALID':
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =E2=9D=8C ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   kabel pripojen a charging ERROR
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'ERROR':
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   kabel pripojen a charging NOT_CHARGING
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'NOT_CHARGING' and chargingLevelHv > last_percent:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =E2=9A=A0=F0=9F=94=8C ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
#   kabel pripojen a charging FINISHED_FULLY_CHARGED
    elif connectionStatus == 'CONNECTED' and chargingStatus == 'FINISHED_FULLY_CHARGED' and chargingLevelHv > last_percent:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = '=?utf-8?Q? =F0=9F=92=AF ?='
#        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)
    else:
        setLastPercent(chargingLevelHv)
        message_subjectEmoji = ''
        message_subjectEmoji = '=?utf-8?Q?  =F0=9F=94=B4=E2=9A=A0=F0=9F=94=8C ?='
        message_subjectEmoji = '=?utf-8?Q? =E2=9D=97=E2=9D=97 ?='
##        sendEMAIL(chargingLevelHv,connectionStatus,chargingStatus,message_subjectEmoji)
        logging.info(message)

except Exception as msg:
    logging.error(msg)
    sys.exit(1)
[root@CL-ITNK-BMW ~]#

Mishaczech · Příspěvekod **Mishaczech** » úte 23. lis 2021 12:43:19

Havrla píše:Hezke povanocni.
Pripravim image linuxoveho virtualu s BMW skryptem a grafanou ke stazeni.

Ja furt čekám na ten IMG :-)

Havrla · Příspěvekod **Havrla** » úte 23. lis 2021 13:14:30

Mishaczech píše:
Havrla píše:Hezke povanocni.
Pripravim image linuxoveho virtualu s BMW skryptem a grafanou ke stazeni.

Ja furt čekám na ten IMG

Vsak je pred vanocena teprve :-)

Mozna mam lepci model, posli mi login do connectdrive/portalu a ja ti to zagrafuju a mas to bez prace :-)

Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení

Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení