--- /dev/null
+"""
+Mercedes Me APIs
+
+Author: G. Ravera
+
+For more details about this component, please refer to the documentation at
+https://github.com/xraver/mercedes_me_api/
+"""
+from configobj import ConfigObj
+import base64
+import json
+import logging
+import os
+import requests
+
+from query import (
+ GetResource,
+ GetToken
+)
+
+# Logger
+logger = logging.getLogger(__name__)
+
+# Software Parameters
+NAME = "Mercedes Me API"
+DOMAIN = "mercedesmeapi"
+VERSION = "0.2"
+TOKEN_FILE = ".mercedesme_token"
+CREDENTIAL_FILE = ".mercedesme_credentials"
+RESOURCES_FILE = ".mercedesme_resources"
+REDIRECT_URL = "https://localhost"
+SCOPE = "mb:vehicle:mbdata:fuelstatus%20mb:vehicle:mbdata:vehiclestatus%20mb:vehicle:mbdata:vehiclelock%20offline_access"
+URL_OAUTH = "https://id.mercedes-benz.com/as/token.oauth2"
+URL_RES_PREFIX = "https://api.mercedes-benz.com/vehicledata/v2"
+
+# File Parameters
+CONF_CLIENT_ID = "CLIENT_ID"
+CONF_CLIENT_SECRET = "CLIENT_SECRET"
+CONF_VEHICLE_ID = "VEHICLE_ID"
+
+class MercedesMeConfig:
+
+ name = NAME
+ domain = DOMAIN
+ version = VERSION
+ token_file = TOKEN_FILE
+ credentials_file = CREDENTIAL_FILE
+ resources_file = RESOURCES_FILE
+ redirect_uri = REDIRECT_URL
+ scope = SCOPE
+ client_id = ""
+ client_secret = ""
+ vin = ""
+ base64 = ""
+ access_token = ""
+ refresh_token = ""
+ oauth_url = URL_OAUTH
+ res_url_prefix = URL_RES_PREFIX
+
+ ########################
+ # Read Configuration
+ ########################
+ def ReadConfig(self):
+ # Read credentials from file
+ if not os.path.isfile(self.credentials_file):
+ logger.error ("Credential File " + self.credentials_file + " not found")
+ return False
+ try:
+ f = ConfigObj(self.credentials_file)
+ except Exception:
+ logger.error ("Wrong "+ self.credentials_file + " file found")
+ return False
+ # Client ID
+ self.client_id = f.get(CONF_CLIENT_ID)
+ if not self.client_id:
+ logger.error ("No "+ CONF_CLIENT_ID + " found in the configuration")
+ return False
+ # Client Secret
+ self.client_secret = f.get(CONF_CLIENT_SECRET)
+ if not self.client_secret:
+ logger.error ("No "+ CONF_CLIENT_SECRET + " found in the configuration")
+ return False
+ # Vehicle ID
+ self.vin = f.get(CONF_VEHICLE_ID)
+ if not self.vin:
+ logger.error ("No "+ CONF_VEHICLE_ID + " found in the configuration")
+ return False
+ # Base64
+ b64_str = self.client_id + ":" + self.client_secret
+ b64_bytes = base64.b64encode( b64_str.encode('ascii') )
+ self.base64 = b64_bytes.decode('ascii')
+
+ # Read Token
+ token_file = self.token_file
+ if not os.path.isfile(token_file):
+ logger.error ("Token File missing - Creating a new one")
+ token = self.CreateToken()
+ if token == None:
+ logger.error ("Error creating token")
+ return False
+ else:
+ with open(token_file, 'r') as file:
+ try:
+ token = json.load(file)
+ except ValueError:
+ token = None
+ if not self.CheckToken(token):
+ logger.error ("Token File not correct - Creating a new one")
+ token = self.CreateToken()
+ if token == None:
+ logger.error ("Error creating token")
+ return False
+ # Save Token
+ self.access_token = token['access_token']
+ self.refresh_token = token['refresh_token']
+ return True
+
+ ########################
+ # Write Token
+ ########################
+ def WriteToken(self, token):
+ token_file = self.token_file
+ with open(token_file, 'w') as file:
+ json.dump(token, file)
+
+ ########################
+ # Check Token
+ ########################
+ def CheckToken(self, token):
+ if "error" in token:
+ if "error_description" in token:
+ logger.error ("Error retriving token: " + token["error_description"])
+ else:
+ logger.error ("Error retriving token: " + token["error"])
+ return False
+ if len(token) == 0:
+ logger.error ("Empty token found.")
+ return False
+ if not 'access_token' in token:
+ logger.error ("Access token not present.")
+ return False
+ if not 'refresh_token' in token:
+ logger.error ("Refresh token not present.")
+ return False
+ return True
+
+ ########################
+ # Create Token
+ ########################
+ def CreateToken(self):
+ print( "Open the browser and insert this link:\n" )
+ print( "https://id.mercedes-benz.com/as/authorization.oauth2?response_type=code&client_id=" + self.client_id + "&redirect_uri=" + self.redirect_uri + "&scope=" + self.scope + "\n")
+ print( "Copy the code in the url:")
+ auth_code = input()
+
+ token = GetToken(self, refresh=False, auth_code=auth_code)
+
+ # Check Token
+ if not self.CheckToken(token):
+ return None
+ else:
+ self.WriteToken(token)
+ return token
+
+ ########################
+ # Refresh Token
+ ########################
+ def RefreshToken(self):
+
+ token = GetToken(self, refresh=True)
+
+ # Check Token
+ if not self.CheckToken(token):
+ return None
+ else:
+ self.WriteToken(token)
+ return token
+ return True
"""
+Mercedes Me APIs
+
Author: G. Ravera
-Version 0.1
-Creation Date: 22/10/2020
-Change log:
- 22/10/2020 - 0.1 - First Issue
+For more details about this component, please refer to the documentation at
+https://github.com/xraver/mercedes_me_api/
"""
import argparse
-import base64
-from configobj import ConfigObj
-import getopt
import logging
-import json
-import os
-import requests
import sys
-# URLs
-oauth_url = "https://id.mercedes-benz.com/as/token.oauth2"
-res_url_prefix = "https://api.mercedes-benz.com/vehicledata/v2"
-# File Parameters
-CONF_CLIENT_ID = "CLIENT_ID"
-CONF_CLIENT_SECRET = "CLIENT_SECRET"
-CONF_VEHICLE_ID = "VEHICLE_ID"
-# Config Parameters
-MercedeMeConfig = {
- "name": "Mercedes Me API",
- "domain": "mercedesmeapi",
- "version": "0.1",
- "token_file": ".mercedesme_token",
- "credentials_file": ".mercedesme_credentials",
- "available_resources_file": ".mercedesme_resources",
- "redirect_uri": "https://localhost",
- "scope": "mb:vehicle:mbdata:fuelstatus%20mb:vehicle:mbdata:vehiclestatus%20mb:vehicle:mbdata:vehiclelock%20offline_access",
- "client_id": "",
- "client_secret": "",
- "vin": "",
- "base64": "",
- "access_token": "",
- "refresh_token": "",
-}
-# Available Resources (Array)
-availableResources = {}
-# Response Validity
-valid_res = False
+from config import MercedesMeConfig
+from resources import MercedesMeResources
+
# Logger
logger = logging.getLogger(__name__)
-########################
-# Read Configuration
-########################
-def ReadCfg():
- global availableResources
-
- # Read Credentials
- if not os.path.isfile(MercedeMeConfig['credentials_file']):
- logger.error ("Credential File " + MercedeMeConfig['credentials_file'] + " not found")
- return False
- f = ConfigObj(MercedeMeConfig['credentials_file'])
- MercedeMeConfig['client_id'] = f.get(CONF_CLIENT_ID)
- if not MercedeMeConfig['client_id']:
- logger.error ("No "+ CONF_CLIENT_ID + " found in the configuration")
- return False
- MercedeMeConfig['client_secret'] = f.get(CONF_CLIENT_SECRET)
- if not MercedeMeConfig['client_secret']:
- logger.error ("No "+ CONF_CLIENT_SECRET + " found in the configuration")
- return False
- MercedeMeConfig['vin'] = f.get(CONF_VEHICLE_ID)
- if not MercedeMeConfig['vin']:
- logger.error ("No "+ CONF_VEHICLE_ID + " found in the configuration")
- return False
- b64_str = MercedeMeConfig['client_id'] + ":" + MercedeMeConfig['client_secret']
- b64_bytes = base64.b64encode( b64_str.encode('ascii') )
- MercedeMeConfig['base64'] = b64_bytes.decode('ascii')
-
- # Read Token
- if not os.path.isfile(MercedeMeConfig['token_file']):
- logger.error ("Token File missing - Creating a new one")
- token = CreateToken()
- if token == None:
- logger.error ("Error creating token")
- return False
- else:
- with open(MercedeMeConfig['token_file'], 'r') as file:
- try:
- token = json.load(file)
- except ValueError:
- token = {}
- if not CheckToken(token):
- logger.error ("Token File not correct - Creating a new one")
- token = CreateToken()
- if token == None:
- return False
- # Save Token
- MercedeMeConfig['access_token'] = token['access_token']
- MercedeMeConfig['refresh_token'] = token['refresh_token']
-
- # Get Resources List
- if not os.path.isfile(MercedeMeConfig['available_resources_file']):
- logger.error ("Resource File missing - Creating a new one")
- return CreateResourcesFile()
- else:
- with open(MercedeMeConfig['available_resources_file'], 'r') as file:
- try:
- availableResources = json.load(file)
- except ValueError:
- availableResources = {}
- if not CheckResources(availableResources):
- logger.error ("Resource File not correct - Creating a new one")
- return CreateResourcesFile()
-
- return True
-
-########################
-# Write Token
-########################
-def WriteToken(token):
- with open(MercedeMeConfig['token_file'], 'w') as file:
- json.dump(token, file)
-
-########################
-# Check Token
-########################
-def CheckToken(token):
- if "error" in token:
- if "error_description" in token:
- logger.error ("Error retriving token: " + token["error_description"])
- else:
- logger.error ("Error retriving token: " + token["error"])
- return False
- if len(token) == 0:
- logger.error ("Empty token found.")
- return False
- if not 'access_token' in token:
- logger.error ("Access token not present.")
- return False
- if not 'refresh_token' in token:
- logger.error ("Refresh token not present.")
- return False
-
- return True
-
-########################
-# Create Token
-########################
-def CreateToken():
- print( "Open the browser and insert this link:\n" )
- print( "https://id.mercedes-benz.com/as/authorization.oauth2?response_type=code&client_id=" + MercedeMeConfig["client_id"] + "&redirect_uri=" + MercedeMeConfig["redirect_uri"] + "&scope=" + MercedeMeConfig["scope"] + "\n")
- print( "Copy the code in the url:")
- auth_code = input()
-
- headers = {
- "Authorization": "Basic " + MercedeMeConfig['base64'],
- "content-type": "application/x-www-form-urlencoded"
- }
- data = "grant_type=authorization_code&code=" + auth_code + "&redirect_uri=" + MercedeMeConfig["redirect_uri"]
-
- res = requests.post(oauth_url, data = data, headers = headers)
- try:
- token = res.json()
- except ValueError:
- logger.error ("Error retriving token " + str(res.status_code))
- return None
-
- # Check Token
- if not CheckToken(token):
- return None
- else:
- WriteToken(token)
- return token
-
-########################
-# Refresh Token
-########################
-def RefreshToken():
- headers = {
- "Authorization": "Basic " + MercedeMeConfig['base64'],
- "content-type": "application/x-www-form-urlencoded"
- }
- data = "grant_type=refresh_token&refresh_token=" + MercedeMeConfig['refresh_token']
-
- res = requests.post(oauth_url, data = data, headers = headers)
- try:
- token = res.json()
- except ValueError:
- logger.error ("Error refreshing token: " + str(res.status_code))
- return False
-
- MercedeMeConfig['access_token'] = token['access_token']
- MercedeMeConfig['refresh_token'] = token['refresh_token']
-
- WriteToken(token)
-
- return True
-
-########################
-# Init Resources
-########################
-def InitResources():
- # Create Empty Status
- for res in availableResources:
- res["status"] = ""
- res["timestamp"] = 0
- res["valid"] = False
-
-########################
-# Write Resources File
-########################
-def WriteResourcesFile():
- with open(MercedeMeConfig['available_resources_file'], 'w') as file:
- json.dump(availableResources, file)
-
-########################
-# Check Resources
-########################
-def CheckResources(resources):
- if "error" in resources:
- if "error_description" in resources:
- logger.error ("Error retriving resources: " + resources["error_description"])
- else:
- logger.error ("Error retriving resources: " + resources["error"])
- return False
- if len(resources) == 0:
- logger.error ("Empty resources found.")
- return False
-
- return True
-
-########################
-# Create Resources File
-########################
-def CreateResourcesFile():
- global availableResources
- resName = "resources"
- resURL = res_url_prefix + "/vehicles/" + MercedeMeConfig['vin'] + "/" + resName
- availableResources = GetResource(resName, resURL)
- if valid_res:
- InitResources()
- WriteResourcesFile()
- return True
- else:
- logger.error ("Error retriving available resources")
- logger.error ("-> " + availableResources["reason"] + " (" + str(availableResources["code"]) + ")")
- return False
-
-########################
-# GetResource
-########################
-def GetResource(resourceName, resourceURL):
- global valid_res
-
- # Set Header
- headers = {
- "accept": "application/json;charset=utf-8",
- "authorization": "Bearer "+MercedeMeConfig['access_token']
- }
-
- # Send Request
- res = requests.get(resourceURL, headers=headers)
- try:
- data = res.json()
- valid_res = True
- except ValueError:
- data = { "reason": "No Data",
- "code" : res.status_code
- }
- valid_res = False
-
- # Check Error
- if not res.ok:
- if ("reason" in data):
- reason = data["reason"]
- else:
- if res.status_code == 400:
- reason = "Bad Request"
- elif res.status_code == 401:
- reason = "Invalid or missing authorization in header"
- elif res.status_code == 402:
- reason = "Payment required"
- elif res.status_code == 403:
- reason = "Forbidden"
- elif res.status_code == 404:
- reason = "Page not found"
- elif res.status_code == 429:
- reason = "The service received too many requests in a given amount of time"
- elif res.status_code == 500:
- reason = "An error occurred on the server side"
- elif res.status_code == 503:
- reason = "The server is unable to service the request due to a temporary unavailability condition"
- else:
- reason = "Generic Error"
- data["reason"] = reason
- data["code"] = res.status_code
- valid_res = False
- return data
-
-########################
-# Print Available Resources
-########################
-def PrintAvailableResources():
- print ("Found %d resources" % len(availableResources) + ":")
- for res in availableResources:
- print (res["name"] + ": " + res_url_prefix + res["href"])
-
-########################
-# Print Resources Status
-########################
-def PrintResourcesStatus(valid = True):
- for res in availableResources:
- if((not valid) | res["valid"]):
- print (res["name"] + ":")
- print ("\tvalid: " + str(res["valid"]))
- print ("\tstatus: " + res["status"])
- print ("\ttimestamp: " + str(res["timestamp"]))
-
-########################
-# Update Resources Status
-########################
-def UpdateResourcesStatus():
- for res in availableResources:
- resName = res["name"]
- resURL = res_url_prefix + res["href"]
- result = GetResource(resName, resURL)
- if valid_res:
- res["valid"] = True
- res["timestamp"] = result[resName]["timestamp"]
- res["status"] = result[resName]["value"]
- # Write Resource File
- WriteResourcesFile()
+class MercedesMeData:
+ def __init__(self):
+ # Configuration Data
+ self.config = MercedesMeConfig()
+ # Resource Data
+ self.resources = MercedesMeResources(self.config)
########################
# Parse Input
parser.add_argument('-r', '--refresh', action='store_true', help="Procedure to refresh the Access Token")
parser.add_argument('-s', '--status', action='store_true', help="Retrive the Status of your Vehicle")
parser.add_argument('-R', '--resources', action='store_true', help="Retrive the list of available resources of your Vehicle")
- parser.add_argument('-v', '--version', action='version', version='%(prog)s ' + MercedeMeConfig['version'])
+ parser.add_argument('-v', '--version', action='version', version='%(prog)s ' + data.config.version)
if len(sys.argv)==1:
parser.print_help(sys.stderr)
# Main
########################
if __name__ == "__main__":
+
+ # Creating Data Structure
+ data = MercedesMeData()
+
args = ParseInput()
# Reading Configuration
- if not ReadCfg():
- logger.error ("Error reading configuration")
+ if not data.config.ReadConfig():
+ logger.error ("Error initializing configuration")
+ exit (1)
+
+ if not data.resources.ReadResources():
+ logger.error ("Error initializing resources")
exit (1)
- if ( args.token == True):
- if not CreateToken():
+ if (args.token == True):
+ if not data.config.CreateToken():
logger.error ("Error creating token")
exit (1)
if (args.refresh == True):
- if not RefreshToken():
+ if not data.config.RefreshToken():
logger.error ("Error refreshing token")
exit (1)
if (args.resources):
- PrintAvailableResources()
+ data.resources.PrintAvailableResources()
if (args.status == True):
- UpdateResourcesStatus()
- PrintResourcesStatus()
+ data.resources.UpdateResourcesStatus()
+ data.resources.PrintResourcesStatus()
--- /dev/null
+"""
+Mercedes Me APIs
+
+Author: G. Ravera
+
+For more details about this component, please refer to the documentation at
+https://github.com/xraver/mercedes_me_api/
+"""
+import logging
+import json
+import os
+
+from config import MercedesMeConfig
+from query import GetResource
+
+# Logger
+logger = logging.getLogger(__name__)
+
+class MercedesMeResources:
+
+ resources = None
+ config = None
+
+ ########################
+ # Init
+ ########################
+ def __init__(self, config):
+ self.config = config
+
+ ########################
+ # Read Resources
+ ########################
+ def ReadResources(self):
+ # Get Resources List
+ if not os.path.isfile(self.config.resources_file):
+ logger.error ("Resource File missing - Creating a new one.")
+ if (self.CreateResourcesFile() == False):
+ logger.error ("Error creating resources file.")
+ return None
+ else:
+ with open(self.config.resources_file, 'r') as file:
+ try:
+ self.resources = json.load(file)
+ except ValueError:
+ resources = None
+ if not self.CheckResources():
+ logger.error ("Resource File not correct - Creating a new one.")
+ if (self.CreateResourcesFile() == False):
+ logger.error ("Error creating resources file.")
+ return None
+ return True
+
+ ########################
+ # Init Resources
+ ########################
+ def InitResources(self):
+ # Create Empty Status
+ for res in self.resources:
+ res["status"] = ""
+ res["timestamp"] = 0
+ res["valid"] = False
+
+ ########################
+ # Write Resources File
+ ########################
+ def WriteResourcesFile(self):
+ with open(self.config.resources_file, 'w') as file:
+ json.dump(self.resources, file)
+
+ ########################
+ # Check Resources
+ ########################
+ def CheckResources(self):
+ if "error" in self.resources:
+ if "error_description" in self.resources:
+ logger.error ("Error retriving resources: " + self.resources["error_description"])
+ else:
+ logger.error ("Error retriving resources: " + self.resources["error"])
+ return False
+ if len(self.resources) == 0:
+ logger.error ("Empty resources found.")
+ return False
+ return True
+
+ ########################
+ # Create Resources File
+ ########################
+ def CreateResourcesFile(self):
+ resName = "resources"
+ resURL = self.config.res_url_prefix + "/vehicles/" + self.config.vin + "/" + resName
+ self.resources = GetResource(resName, resURL, self.config)
+ if "reason" in self.resources:
+ logger.error ("Error retriving available resources - " + self.resources["reason"] + " (" + str(self.resources["code"]) + ")")
+ return False
+ else:
+ self.InitResources()
+ self.WriteResourcesFile()
+ return True
+
+ ########################
+ # Print Available Resources
+ ########################
+ def PrintAvailableResources(self):
+ print ("Found %d resources" % len(self.resources) + ":")
+ for res in self.resources:
+ print (res["name"] + ": " + self.config.res_url_prefix + res["href"])
+
+ ########################
+ # Print Resources Status
+ ########################
+ def PrintResourcesStatus(self, valid = True):
+ for res in self.resources:
+ if((not valid) | res["valid"]):
+ print (res["name"] + ":")
+ print ("\tvalid: " + str(res["valid"]))
+ print ("\tstatus: " + res["status"])
+ print ("\ttimestamp: " + str(res["timestamp"]))
+
+ ########################
+ # Update Resources Status
+ ########################
+ def UpdateResourcesStatus(self):
+ for res in self.resources:
+ resName = res["name"]
+ resURL = self.config.res_url_prefix + res["href"]
+ result = GetResource(resName, resURL, self.config)
+ if not "reason" in result:
+ res["valid"] = True
+ res["timestamp"] = result[resName]["timestamp"]
+ res["status"] = result[resName]["value"]
+ # Write Resource File
+ self.WriteResourcesFile()
\ No newline at end of file
projects / mercedes_me_api.git / commitdiff