- [Vehicle Status](https://developer.mercedes-benz.com/products/vehicle_status)
Note: the APIs described above do not requires any subscription in case you use them with your own car associated the the Mercedes Me Account.
+Note2: only one car is supported for the moment.
+
+## Home Assistant Custom Component
+The Home Assistant Custom Component is a component to be added in Home Assistant in order to integrate sensors of a Mercedes Benz car.
+This component is still in development.
+### Open Points
+- Fix OAUTH2 Authentication & Get the First Token
+- Log Management
+- Bugfix & Software optimizations
+
+### Installation
+To use this custom component it's necessary to perform the following instructions:
+1) clone the repository
+2) create make a symbolic link from git_repost/custom_components/mercedesmeapi to hass_folder/custom_components
+3) Add in the configuration.yaml the following emelents:
+```yaml
+mercedesmeapi:
+ client_id: <**INSERT_YOUR_CLIENT_ID**>
+ client_secret: <**INSERT_YOUR_CLIENT_SECRET**>
+ vehicle_id: <**INSERT_YOUR_VEHICLE_ID**>
+ scan_interval: <** TIME PERIOD TO REFRESH RESOURCES **>
+```
+4) Actually it's not possible to retrive the Token from scratch. Please use the other script to retrive the first token and copy it into hacs folder (.mercedesme_toke)
+
+## Shell Scripts
+There are two shell script:
+1) Python Version
+2) Bash Version
+The installation is the same, the usage is different.
## Installation
To use this script it's necessary to perform the following instructions:
where CLIENT_ID and CLIENT_SECRET referring to the application information that can be found in [Mercedes Developer Console](https://developer.mercedes-benz.com/console) and VEHICLE_ID is the VIN of your car.
-## Python Usage
+### Python Usage
To execute the script read below:
```bash
usage: mercedes_me_api.py [-h] [-t] [-r] [-s] [-R] [-v]
-v, --version show program's version number and exit
```
-## Bash Usage
+### Bash Usage
To execute the script read below:
```bash
Usage: mercedes_me_api.sh <arguments>
--- /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 datetime import timedelta
+import logging
+
+from .config import MercedesMeConfig
+from .resources import MercedesMeResources
+from .const import *
+
+# Logger
+_LOGGER = logging.getLogger(__name__)
+
+class MercedesMeData:
+
+ ########################
+ # Init
+ ########################
+ def __init__(self, hass, config):
+ # Configuration Data
+ self.mercedesConfig = MercedesMeConfig(hass, config)
+ # Resource Data
+ self.mercedesResources = MercedesMeResources(self.mercedesConfig)
+
+ ########################
+ # Update State
+ ########################
+ def UpdateState(self, event_time):
+ _LOGGER.debug ("Start Update Resources")
+ self.mercedesResources.UpdateResourcesState()
+ #hass.helpers.dispatcher.dispatcher_send(UPDATE_SIGNAL)
+ _LOGGER.debug ("End Update")
+
+ ########################
+ # Update State
+ ########################
+ def UpdateToken(self, event_time):
+ _LOGGER.debug ("Start Refresh Token")
+ self.mercedesConfig.RefreshToken()
+ #hass.helpers.dispatcher.dispatcher_send(UPDATE_SIGNAL)
+ _LOGGER.debug ("End Refresh Token")
+
+########################
+# Setup
+########################
+#async def async_setup(hass, config):
+def setup(hass, config):
+
+ # Creating Data Structure
+ data = MercedesMeData(hass, config)
+ hass.data[DOMAIN] = data
+
+ # Reading Configuration
+ if not data.mercedesConfig.ReadConfig():
+ _LOGGER.error ("Error initializing configuration")
+ return False
+
+ if not data.mercedesResources.ReadResources():
+ _LOGGER.error ("Error initializing resources")
+ return False
+
+ # Create Task to initializate Platform Sensor
+ hass.async_create_task(
+ hass.helpers.discovery.async_load_platform("sensor", DOMAIN, {}, config)
+ )
+
+ # Create Task to Update Status
+ hass.helpers.event.track_time_interval(data.UpdateState, timedelta(seconds=data.mercedesConfig.scan_interval))
+
+ # Create Task to Update Token
+ hass.helpers.event.track_time_interval(data.UpdateToken, timedelta(seconds=data.mercedesConfig.token_expires_in))
+
+ return True
\ No newline at end of file
--- /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 base64
+import json
+import logging
+import os
+import requests
+import voluptuous as vol
+
+from homeassistant.const import (
+ CONF_SCAN_INTERVAL,
+# LENGTH_KILOMETERS,
+# LENGTH_MILES,
+)
+
+from homeassistant.helpers import discovery, config_validation as cv
+
+from .query import (
+ GetResource,
+ GetToken
+)
+from .const import *
+
+# Logger
+_LOGGER = logging.getLogger(__name__)
+
+CONFIG_SCHEMA = vol.Schema (
+ {
+ DOMAIN: vol.Schema (
+ {
+ vol.Required(CONF_CLIENT_ID): cv.string,
+ vol.Required(CONF_CLIENT_SECRET): cv.string,
+ vol.Required(CONF_VEHICLE_ID): cv.string,
+ vol.Optional(CONF_SCAN_INTERVAL, default=30): vol.All(
+ cv.positive_int, vol.Clamp(min=60)
+ ),
+ }
+ )
+ },
+ extra=vol.ALLOW_EXTRA,
+)
+
+class MercedesMeConfig:
+
+ token_file = ""
+ credentials_file = ""
+ resources_file = ""
+ client_id = ""
+ client_secret = ""
+ vin = ""
+ scan_interval = ""
+ base64 = ""
+ access_token = ""
+ refresh_token = ""
+ token_expires_in = ""
+ oauth_url = URL_OAUTH
+ res_url_prefix = URL_RES_PREFIX
+
+ ########################
+ # Init
+ ########################
+ def __init__(self, hass, config):
+ # Home Assistant structures
+ self.hass = hass
+ self.config = config
+ # Files
+ self.token_file = self.hass.config.path(TOKEN_FILE)
+ self.resources_file = self.hass.config.path(RESOURCES_FILE)
+
+ ########################
+ # Read Configuration
+ ########################
+ def ReadConfig(self):
+ needToRefresh = False
+
+ config = self.config[DOMAIN]
+ # Client ID
+ self.client_id = config[CONF_CLIENT_ID]
+ # Client Secret
+ self.client_secret = config[CONF_CLIENT_SECRET]
+ # Vehicle ID
+ self.vin = config[CONF_VEHICLE_ID]
+ # Scan Interval
+ self.scan_interval = config[CONF_SCAN_INTERVAL]
+ # 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
+ if not os.path.isfile(self.token_file):
+ _LOGGER.error ("Token File missing - Creating a new one")
+ #if (not self.CreateToken()): GRGR -> to be fixed
+ if (True):
+ _LOGGER.error ("Error creating token")
+ return False
+ else:
+ with open(self.token_file, 'r') as file:
+ try:
+ token = json.load(file)
+ except ValueError:
+ token = None
+ if self.CheckToken(token):
+ # Save Token
+ self.access_token = token['access_token']
+ self.refresh_token = token['refresh_token']
+ self.token_expires_in = token['expires_in']
+ needToRefresh = True
+ else:
+ _LOGGER.error ("Token File not correct - Creating a new one")
+ #if (not self.CreateToken()):
+ if (True): # GRGR -> to be fixed
+ _LOGGER.error ("Error creating token")
+ return False
+
+ if (needToRefresh):
+ if (not self.RefreshToken()):
+ _LOGGER.error ("Error refreshing token")
+ return False
+
+ return True
+
+ ########################
+ # Write Token
+ ########################
+ def WriteToken(self, token):
+ with open(self.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=" + REDIRECT_URL + "&scope=" + 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 False
+ else:
+ # Save Token
+ self.WriteToken(token)
+ self.access_token = token['access_token']
+ self.refresh_token = token['refresh_token']
+ self.token_expires_in = token['expires_in']
+ return True
+
+ ########################
+ # Refresh Token
+ ########################
+ def RefreshToken(self):
+
+ token = GetToken(self, refresh=True)
+
+ # Check Token
+ if not self.CheckToken(token):
+ return False
+ else:
+ # Save Token
+ self.WriteToken(token)
+ self.access_token = token['access_token']
+ self.refresh_token = token['refresh_token']
+ self.token_expires_in = token['expires_in']
+ return True
--- /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/
+"""
+# Software Parameters
+NAME = "Mercedes Me API"
+DOMAIN = "mercedesmeapi"
+VERSION = "0.3"
+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"
+#UPDATE_SIGNAL = "mercedesmeapi_update"
+
+# File Parameters
+CONF_CLIENT_ID = "client_id"
+CONF_CLIENT_SECRET = "client_secret"
+CONF_VEHICLE_ID = "vehicle_id"
--- /dev/null
+{
+ "domain": "mercedesmeapi",
+ "name": "Mercedes Me Api",
+ "documentation": "https://github.com/xraver/mercedes_me_api",
+ "dependencies": [],
+ "config_flow": false,
+ "codeowners": [
+ "@xraver"
+ ],
+ "requirements": [
+ "requests"
+ ],
+ "homeassistant": "0.100.0"
+ }
--- /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 requests
+
+from .const import *
+
+# Logger
+_LOGGER = logging.getLogger(__name__)
+
+########################
+# GetResource
+########################
+def GetResource(resourceName, resourceURL, config):
+
+ # Set Header
+ headers = {
+ "accept": "application/json;charset=utf-8",
+ "authorization": "Bearer "+ config.access_token
+ }
+
+ # Send Request
+ res = requests.get(resourceURL, headers=headers)
+ try:
+ data = res.json()
+ except ValueError:
+ data = { "reason": "No Data",
+ "code" : res.status_code
+ }
+
+ # 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
+ return data
+
+########################
+# GetToken
+########################
+def GetToken(config, refresh=True, auth_code=""):
+ headers = {
+ "Authorization": "Basic " + config.base64,
+ "content-type": "application/x-www-form-urlencoded"
+ }
+
+ if (not refresh):
+ # New Token
+ data = "grant_type=authorization_code&code=" + auth_code + "&redirect_uri=" + REDIRECT_URL
+ else:
+ # Refresh
+ data = "grant_type=refresh_token&refresh_token=" + config.refresh_token
+
+ res = requests.post(URL_OAUTH, data = data, headers = headers)
+ try:
+ token = res.json()
+ except ValueError:
+ _LOGGER.error ("Error retriving token " + str(res.status_code))
+ return None
+
+ return token
--- /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 datetime import timedelta
+import logging
+import json
+import os
+
+from homeassistant.helpers.entity import Entity
+
+from .config import MercedesMeConfig
+from .query import GetResource
+from .const import *
+
+# Logger
+_LOGGER = logging.getLogger(__name__)
+
+class MercedesMeResource (Entity):
+ def __init__( self, name, vin, version, href, state=None, timestamp=None, valid=False ):
+ self._name = name
+ self._version = version
+ self._href = href
+ self._vin = vin
+ self._vin = "mercedes" # GRGR -> test
+ self._state = state
+ self._timestamp = timestamp
+ self._valid = valid
+
+ def __str__(self):
+ return json.dumps({
+ "name" : self._name,
+ "vin" : self._vin,
+ "version" : self._version,
+ "href" : self._href,
+ "state" : self._state,
+ "timestamp" : self._timestamp,
+ "valid" : self._valid,
+ })
+
+ def getJson(self):
+ return ({
+ "name" : self._name,
+ "vin" : self._vin,
+ "version" : self._version,
+ "href" : self._href,
+ "state" : self._state,
+ "timestamp" : self._timestamp,
+ "valid" : self._valid,
+ })
+
+ @property
+ def name(self):
+ """Return the name of the sensor."""
+ return self._vin + "_" + self._name
+
+ @property
+ def state(self):
+ """Return state for the sensor."""
+ return self._state
+
+ @property
+ def device_state_attributes(self):
+ """Return attributes for the sensor."""
+ return ({
+ "valid": self._valid,
+ "timestamp": self._timestamp,
+ })
+
+# @property
+# def unit_of_measurement(self):
+# """Return the unit of measurement."""
+# return TEMP_CELSIUS
+
+class MercedesMeResources:
+
+ ########################
+ # Init
+ ########################
+ def __init__(self, mercedesConfig):
+
+ self.database = []
+ self.mercedesConfig = mercedesConfig
+
+ ########################
+ # Read Resources
+ ########################
+ def ReadResources(self):
+
+ found = False
+ resources = None
+
+ if not os.path.isfile(self.mercedesConfig.resources_file):
+ # Resources File not present - Retriving new one from server
+ _LOGGER.error ("Resource File missing - Creating a new one.")
+ found = False
+ else:
+ with open(self.mercedesConfig.resources_file, 'r') as file:
+ try:
+ resources = json.load(file)
+ if (not self.CheckResources(resources)):
+ raise ValueError
+ else:
+ found = True
+ except ValueError:
+ _LOGGER.error ("Error reading resource file - Creating a new one.")
+ found = False
+
+ if ( not found ):
+ # Not valid or file missing
+ resources = self.RetriveResourcesList()
+ if( resources == None ):
+ # Not found or wrong
+ _LOGGER.error ("Error retriving resource list.")
+ return False
+ else:
+ # import and write
+ self.ImportResourcesList(resources)
+ self.WriteResourcesFile()
+ return True
+ else:
+ # Valid: just import
+ self.ImportResourcesList(resources)
+ return True
+
+ ########################
+ # Check Resources
+ ########################
+ def CheckResources(self, resources):
+ if "reason" in resources:
+ _LOGGER.error ("Error retriving available resources - " + resources["reason"] + " (" + str(resources["code"]) + ")")
+ return False
+ 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
+
+ ########################
+ # Retrive Resources List
+ ########################
+ def RetriveResourcesList(self):
+ resName = "resources"
+ resURL = URL_RES_PREFIX + "/vehicles/" + self.mercedesConfig.vin + "/" + resName
+ resources = GetResource(resName, resURL, self.mercedesConfig)
+ if not self.CheckResources(resources):
+ _LOGGER.error ("Error retriving available resources")
+ return None
+ else:
+ return resources
+
+ ########################
+ # Import Resources List
+ ########################
+ def ImportResourcesList(self, resources):
+ for res in resources:
+ if("state" in res):
+ self.database.append( MercedesMeResource (res["name"], self.mercedesConfig.vin, res["version"], res["href"], res["state"], res["timestamp"], res["valid"]) )
+ else:
+ self.database.append( MercedesMeResource (res["name"], self.mercedesConfig.vin, res["version"], res["href"]) )
+
+ ########################
+ # Write Resources File
+ ########################
+ def WriteResourcesFile(self):
+ output = []
+ # Extract List
+ for res in self.database:
+ output.append( res.getJson() )
+ # Write File
+ with open(self.mercedesConfig.resources_file, 'w') as file:
+ json.dump(output, file)
+
+ ########################
+ # Print Available Resources
+ ########################
+ def PrintAvailableResources(self):
+ print ("Found %d resources" % len(self.database) + ":")
+ for res in self.database:
+ print (res._name + ": " + URL_RES_PREFIX + res._href)
+
+ ########################
+ # Print Resources State
+ ########################
+ def PrintResourcesState(self, valid = True):
+ for res in self.database:
+ if((not valid) | res._valid):
+ print (res._name + ":")
+ print ("\tvalid: " + str(res._valid))
+ print ("\tstate: " + res._state)
+ print ("\ttimestamp: " + str(res._timestamp))
+
+ ########################
+ # Update Resources State
+ ########################
+ def UpdateResourcesState(self):
+ for res in self.database:
+ resName = res._name
+ resURL = URL_RES_PREFIX + res._href
+ result = GetResource(resName, resURL, self.mercedesConfig)
+ if not "reason" in result:
+ res._valid = True
+ res._timestamp = result[resName]["timestamp"]
+ res._state = result[resName]["value"]
+ # Write Resource File
+ self.WriteResourcesFile()
--- /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
+
+from .config import MercedesMeConfig
+from .config import DOMAIN
+from .resources import MercedesMeResources
+
+#DEPENDENCIES = ['mercedesmeapi']
+
+# Logger
+_LOGGER = logging.getLogger(__name__)
+
+async def async_setup_platform(hass, config, async_add_entities, _discovery_info=None):
+ """Setup sensor platform."""
+
+ mercedesConfig = hass.data[DOMAIN].mercedesConfig
+ mercedesRes = hass.data[DOMAIN].mercedesResources
+
+ async_add_entities(mercedesRes.database, True)
projects / mercedes_me_api.git / commitdiff