Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
Moderátor: Josef
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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.
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
Hezke povanocni.
Pripravim image linuxoveho virtualu s BMW skryptem a grafanou ke stazeni.
Pripravim image linuxoveho virtualu s BMW skryptem a grafanou ke stazeni.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
#################################################################################
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
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
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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)
Naposledy upravil(a) Havrla dne sob 18. pro 2021 15:19:52, celkem upraveno 1 x.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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.
########################################
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.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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
Naposledy upravil(a) Havrla dne sob 18. pro 2021 15:19:31, celkem upraveno 1 x.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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
Naposledy upravil(a) Havrla dne sob 18. pro 2021 15:18:40, celkem upraveno 1 x.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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
Naposledy upravil(a) Havrla dne sob 18. pro 2021 15:17:28, celkem upraveno 1 x.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
Díky. Počkáme s čím přijdou vývojáři
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
Prý pujde vyčítat socMax, ale priorita to není
https://github.com/bimmerconnected/bimmer_connected/discussions/361
https://github.com/bimmerconnected/bimmer_connected/discussions/361
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
V klidu....
k aktualimu skryptu co mame si priinstalni :
pip3 install --upgrade bimmer_connected
muj novy vzhledem k moznostem jakstaks skrypt
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 ~]#
Naposledy upravil(a) Havrla dne sob 18. pro 2021 15:18:02, celkem upraveno 3 x.
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
-
- Příspěvky: 141
- Registrován: stř 10. črc 2019 8:18:43
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
Havrla píše:Hezke povanocni.
Pripravim image linuxoveho virtualu s BMW skryptem a grafanou ke stazeni.
Ja furt čekám na ten IMG
Re: Vyčítání dat z BMW telematiky, uložení do vlastní databáze a následné zobrazení
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
Havrla
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L
Jezdive: BMW i3 REx; BMW I3, Volvo V70 T6; TM3, Tatra T700-II.
Nejezdive: Tatra 613-4; Skoda 105L