Ik heb al een tijdje ook een Python scriptje draaien om de waardes van Stromer te loggen in Traccar.
Hiermee maak ik nog geen gebruik van de refresh tokens, wel logged ie niet elke keer opnieuw in.
Code: Selecteer alles
import requests
import urllib
import datetime
import json
import time
import math
from geopy.geocoders import Nominatim
password = "StromerAccountPassword"
username = "StromerAccountEmail"
client_id = "APP client_id"
client_secret = "APP client_secret"
traccar_url = 'http://10.0.0.6:5055/' #IP and Port of Traccar
traccar_id = 102
unavailableTime = 30 #All locations below this time limit in minutes will be accepted and passed to Traccar
scan_interval = 2*60 #Interval of script in seconds
sleepDuration = 0
refreshInterval = 30*60 #interval in seconds
def get_code(client_id, username, password):
MAX_RETRIES = 20
session = requests.session()
adapter = requests.adapters.HTTPAdapter(max_retries=MAX_RETRIES)
session.mount('https://', adapter)
session.mount('http://', adapter)
url = "https://api3.stromer-portal.ch/users/login/"
s = requests.session()
res = s.get(url)
s.cookies
qs = urllib.parse.urlencode({
"client_id":
client_id,
"response_type":
"code",
"redirect_url":
"stromerauth://auth",
"scope":
"bikeposition bikestatus bikeconfiguration bikelock biketheft bikedata bikepin bikeblink userprofile",
})
data = {
"password": password,
"username": username,
"csrfmiddlewaretoken": s.cookies.get("csrftoken"),
"next": "/o/authorize/?" + qs,
}
res = s.post(url, data=data, headers=dict(Referer=url), allow_redirects=False)
res = s.send(res.next, allow_redirects=False)
qs = urllib.parse.urlparse(res.headers["Location"])
code = qs.query.rsplit("=", 1)[1]
return code
def get_access_token(client_id, client_secret, code):
url = "https://api3.stromer-portal.ch//o/token/"
params = {
"grant_type": "authorization_code",
"client_id": client_id,
"client_secret": client_secret,
"code": code,
"redirect_uri": "stromerauth://auth",
}
res = requests.post(url, params=params)
return res.json()["access_token"]
def call_api(access_token, endpoint, params={}):
url = "https://api3.stromer-portal.ch/rapi/mobile/v2/%s" % endpoint
headers = {"Authorization": "Bearer %s" % access_token}
res = requests.get(url, headers=headers, params={})
data = res.json()["data"]
if isinstance(data, list):
return data[0]
else:
return data
def call_bike(access_token, bike, endpoint, cached="false"):
endpoint = 'bike/%s/%s' % (bike["bikeid"], endpoint)
params = {'cached': '%s' % cached}
state = call_api(access_token, endpoint, params)
return state
def distanceCalc(lat1, lon1, lat2, lon2):
R = 6364.611 # Radius of earth in km @ 52.8*
lat1 = math.radians(lat1)
lon1 = math.radians(lon1)
lat2 = math.radians(lat2)
lon2 = math.radians(lon2)
dlon = lon2 - lon1
dlat = lat2 - lat1
a = math.sin(dlat / 2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2)**2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
distance = R * c * 1000 #distance in meters
return distance
#UTC offset
is_dst = time.daylight and time.localtime().tm_isdst > 0
utc_offset = - (time.altzone if is_dst else time.timezone)
#Variables
prevLat = -1
prevLon = -1
prevTime = -1
prevSoC = -1
prevSpeed = -1
geolocator = Nominatim(user_agent="FindMyStromer")
print("Script for Stromer bike location")
code = get_code(client_id, username, password)
access_token = get_access_token(client_id, client_secret, code)
while True:
starttime = int(time.time())
#Check if it a uncached version needs to be requested
cache = "true"
if ((starttime - prevTime) > (refreshInterval + 1)):
cache = "false"
try:
bike = call_api(access_token, "bike")
state = call_bike(access_token, bike, 'state/', cache)
position = call_bike(access_token, bike, 'position/', cache)
#print(state)
#print(position)
#print(call_bike(access_token, bike, 'service_info/', cache))
#print(call_bike(access_token, bike, 'bike/statistics/all/', cache))
#print(call_bike(access_token, bike, 'bike/statistics/extra_data/', cache))
except:
code = get_code(client_id, username, password)
access_token = get_access_token(client_id, client_secret, code)
bike = call_api(access_token, "bike")
state = call_bike(access_token, bike, 'state/', cache)
position = call_bike(access_token, bike, 'position/', cache)
location_time = state['rcvts']
now = int(time.time())
if (location_time >= now):
location_time = location_time - utc_offset
try:
location_adress = str(geolocator.reverse(str(position['latitude']) + ', ' + str(position['longitude'])))
except:
location_adress = 'Onbekend'
print("Bike is at " + location_adress + " (" + str(position['latitude']) + ', ' + str(position['longitude']) + " at " + str(position['altitude']) + "m)" + " with " + str(state['battery_SOC']) + "% SoC")
#Post to Traccar
validLocation = True
if (prevLat != -1) and (prevLon != -1):
prevDistance = distanceCalc(position['latitude'], position['longitude'], prevLat, prevLon) #in meter
try:
prevSpeed = (prevDistance / (location_time - prevTime)) * 3.6 #in km/h
except:
prevSpeed = 0
if (prevSpeed > 80):
validLocation = False
newData = True
if validLocation:
if (position['latitude'] == prevLat) and (position['longitude'] == prevLon) and (prevSoC == state['battery_SOC']) and (prevSpeed == state['bike_speed']):
newData = False
prevLat = position['latitude']
prevLon = position['longitude']
prevTime = location_time
prevSoC == state['battery_SOC']
prevSpeed = state['bike_speed']
#if True:
if ((now - location_time) < (sleepDuration + 1)) and validLocation and newData:
traccar_post = traccar_url + '?id=' + str(traccar_id) + '&lat=' + str(position['latitude']) + '&lon=' + str(position['longitude']) + '×tamp=' + str(location_time) + '&hdop=2&altitude=' + str(position['altitude']) + '&speed=' + str(state['bike_speed']/1.8) + '&accuracy=' + str(int(123)) + '&batt=' + str(state['battery_SOC']) + '&address=' + location_adress + "&batteryhealth=" + str(state['battery_health']) + "&assist=" + str(state['assistance_level']) + "&totalenergy=" + str(state['total_energy_consumption']) + "&tempbattery=" + str(state['battery_temp']) + "&tempmotor=" + str(state['motor_temp']) + "&locked=" + str(state['lock_flag']) + "&powercycles=" + str(state['power_on_cycles']) + "&totaldistance=" + str(state['total_distance'])
traccar_response = requests.get(traccar_post)
else:
print("Old!")
sleepDuration = scan_interval - (now - starttime)
print(str(int(sleepDuration)) + ' seconden wachten')
time.sleep(sleepDuration)
