Merge pull request #35 from Zegorax/master

[CORE] Add Teslamate integration

.env_sample

@@ -4,6 +4,11 @@ SECRET_KEY=RANDOMLY_GENERATED_HERE
 ADMIN_PASSWORD=HTPASS_PASSWORD_FOR_ADMIN_PAGE
 BASE_URL=/map
 MAPBOX_TOKEN=pk.BLA
-TESLALOGGER_BASEURL=http://raspberry:5010/
-TESLALOGGER_CARID=1
+
+# BACKEND_PROVIDER can be either "teslalogger" or "teslamate"
+# For Teslamate, you need to deploy TeslamateAPI, available here : https://github.com/tobiasehlert/teslamateapi
+BACKEND_PROVIDER=teslamate
+BACKEND_PROVIDER_BASE_URL=http://insert-base-api-here:withport/
+BACKEND_PROVIDER_CAR_ID=1
+
 TZ=Europe/Berlin

.gitignore

@@ -2,3 +2,5 @@
 service.db
 .venv
 .DS_Store
+dev/
+*.pyc

app.py

@@ -9,13 +9,21 @@
 import flask_login
 from geopy.distance import geodesic
 
+from interfaces.backendfactory import BackendProviderFactory
+
 load_dotenv()
 MAPBOX_TOKEN = os.getenv('MAPBOX_TOKEN')
-TESLALOGGER_BASEURL = os.getenv('TESLALOGGER_BASEURL')
-TESLALOGGER_CARID = os.getenv('TESLALOGGER_CARID', 1)
 BASE_URL = os.getenv('BASE_URL')
 PORT = os.getenv('PORT', 5051)
 DATA_DIR = os.getenv('DATA_DIR', '/data/')
+
+BACKEND_PROVIDER = os.getenv('BACKEND_PROVIDER', 'teslalogger')
+BACKEND_PROVIDER_BASE_URL = os.getenv('BACKEND_PROVIDER_BASE_URL')
+BACKEND_PROVIDER_CAR_ID = os.getenv('BACKEND_PROVIDER_CAR_ID', 1)
+
+# Backend provider instanciation
+BackendProviderFactory(BACKEND_PROVIDER, BACKEND_PROVIDER_BASE_URL, BACKEND_PROVIDER_CAR_ID)
+
 app = Flask(__name__)
 app.secret_key = os.getenv('SECRET_KEY')
 
@@ -48,6 +56,7 @@
 users = {'admin': {'password': os.getenv('ADMIN_PASSWORD', 'password')}}
 
 
+
 class User(flask_login.UserMixin):
     pass
 
@@ -151,44 +160,38 @@ def carstate(shortuuid):
     conn.close()
 
     if result:
-        lat = result[0]
-        lng = result[1]
+        db_latitude = result[0]
+        db_longitude = result[1]
         expiry = result[2]
 
         if expiry > time.time():
-            teslalogger = requests.get(TESLALOGGER_BASEURL + 'currentjson/' + TESLALOGGER_CARID + '/')
-
-            carstate = {
-                'latitude': teslalogger.json()['latitude'],
-                'longitude': teslalogger.json()['longitude'],
-                'odometer': teslalogger.json()['odometer'],
-                'driving': teslalogger.json()['driving'],
-                'charging': teslalogger.json()['charging'],
-                'battery_level': teslalogger.json()['battery_level'],
-                }
-
+            provider = BackendProviderFactory.get_instance()
+            provider.refresh_data()
+
+            temp_carstate = vars(provider)
+
             # Check if ETA destination is similar and use Tesla provided destination if it's within 250m
-            destination_db = (lat, lng)
-            if teslalogger.json()['active_route_destination']:
-                destination_tesla = (teslalogger.json()['active_route_latitude'], teslalogger.json()['active_route_longitude'])
-                distance = geodesic(destination_db, destination_tesla).km
-
-                if distance < 0.25:
-                    carstate['eta_destination_lat'] = destination_tesla[0]
-                    carstate['eta_destination_lng'] = destination_tesla[1]
-                    carstate['eta_destination_tesla_seconds'] = teslalogger.json()['active_route_minutes_to_arrival'] * 60
-                    carstate['eta_destination_tesla_battery_level'] = teslalogger.json()['active_route_energy_at_arrival']
+            # destination_db = (lat, lng)
+            if provider.active_route_destination:
+                tesla_destination = (provider.active_route_latitude, provider.active_route_longitude)
+
+                if not db_latitude or not db_longitude:
+                    # If the lat/lon for the destination was not set on the shared link stored in DB, use the destination set in Tesla
+                    temp_carstate['eta_destination_lat'] = provider.active_route_latitude
+                    temp_carstate['eta_destination_lng'] = provider.active_route_longitude
+                    temp_carstate['eta_destination_tesla_seconds'] = provider.active_route_seconds_to_arrival
+                    temp_carstate['eta_destination_tesla_battery_level'] = provider.active_route_energy_at_arrival
                 else:
-                    carstate['eta_destination_lat'] = destination_db[0]
-                    carstate['eta_destination_lng'] = destination_db[1]
-                    carstate['eta_waypoint_lat'] = destination_tesla[0]
-                    carstate['eta_waypoint_lng'] = destination_tesla[1]
+                    temp_carstate['eta_destination_lat'] = db_latitude
+                    temp_carstate['eta_destination_lng'] = db_longitude
+                    temp_carstate['eta_waypoint_lat'] = provider.active_route_latitude
+                    temp_carstate['eta_waypoint_lng'] = provider.active_route_longitude
 
             else:
-                carstate['eta_destination_lat'] = destination_db[0]
-                carstate['eta_destination_lng'] = destination_db[1]
+                temp_carstate['eta_destination_lat'] = provider.latitude
+                temp_carstate['eta_destination_lng'] = provider.longitude
 
-            return carstate
+            return temp_carstate
         else:
             return('Link Expired'), 410
     else:
@@ -226,12 +229,13 @@ def map_admin():
 
     print(result)
 
-    teslalogger = requests.get(TESLALOGGER_BASEURL + 'currentjson/' + TESLALOGGER_CARID + '/')
+    provider = BackendProviderFactory.get_instance()
+    provider.refresh_data()
 
     if 'uuid' in locals():
-        return render_template('map_admin.html.j2', result=result, BASE_URL=BASE_URL, uuid=uuid, mbtoken=MAPBOX_TOKEN, car_location=[teslalogger.json()['longitude'],teslalogger.json()['latitude']])
+        return render_template('map_admin.html.j2', result=result, BASE_URL=BASE_URL, uuid=uuid, mbtoken=MAPBOX_TOKEN, car_location=[provider.longitude, provider.latitude])
     else:
-        return render_template('map_admin.html.j2', result=result, BASE_URL=BASE_URL, mbtoken=MAPBOX_TOKEN, car_location=[teslalogger.json()['longitude'],teslalogger.json()['latitude']])
+        return render_template('map_admin.html.j2', result=result, BASE_URL=BASE_URL, mbtoken=MAPBOX_TOKEN, car_location=[provider.longitude, provider.latitude])
 
 @app.template_filter('fromtimestamp')
 def _jinja2_filter_datetime(date, fmt=None):

backendproviders/teslalogger.py

@@ -0,0 +1,20 @@
+from interfaces.backendinterface import IBackendProvider
+import requests
+
+class TeslaloggerBackendProvider(IBackendProvider):
+    def refresh_data(self):
+        data = requests.get(f"{self.base_url}/currentjson/{self.car_id}/").json()
+
+        self.latitude = data["latitude"]
+        self.longitude = data["longitude"]
+        self.odometer = data["odometer"]
+        self.is_driving = data["driving"]
+        self.is_charging  = data["charging"]
+        self.battery_level = data["battery_level"]
+
+        self.active_route_latitude = data["active_route_latitude"]
+        self.active_route_longitude = data["active_route_longitude"]
+
+        self.active_route_destination = data["active_route_destination"]
+        self.active_route_minutes_to_arrival = data["active_route_minutes_to_arrival"]
+        self.active_route_energy_at_arrival = data["active_route_energy_at_arrival"]

backendproviders/teslamate.py

@@ -0,0 +1,22 @@
+from interfaces.backendinterface import IBackendProvider
+import requests
+
+class TeslamateBackendProvider(IBackendProvider):
+    def refresh_data(self):
+
+        data = requests.get(f"{self.base_url}/api/v1/cars/{self.car_id}/status").json()["data"]["status"]
+
+        self.latitude = data["car_geodata"]["latitude"]
+        self.longitude = data["car_geodata"]["longitude"]
+        self.odometer = data["odometer"]
+        self.is_driving = data["driving_details"]["shift_state"] is not "P"
+        self.is_charging  = data["charging_details"]["time_to_full_charge"] > 0
+        self.battery_level = data["battery_details"]["battery_level"]
+
+
+        self.active_route_latitude = data["driving_details"]["active_route"]["location"]["latitude"]
+        self.active_route_longitude = data["driving_details"]["active_route"]["location"]["longitude"]
+
+        self.active_route_destination = data["driving_details"]["active_route"]["destination"]
+        self.active_route_minutes_to_arrival = data["driving_details"]["active_route"]["minutes_to_arrival"]
+        self.active_route_energy_at_arrival = data["driving_details"]["active_route"]["energy_at_arrival"]

docker-compose.yml

@@ -0,0 +1,40 @@
+services:
+  teslamate:
+    image: teslamate/teslamate:latest
+    restart: always
+    environment:
+      - ENCRYPTION_KEY=secretkey #replace with a secure key to encrypt your Tesla API tokens
+      - DATABASE_USER=teslamate
+      - DATABASE_PASS=password #insert your secure database password!
+      - DATABASE_NAME=teslamate
+      - DATABASE_HOST=database
+      - MQTT_HOST=mosquitto
+    ports:
+      - 4000:4000
+    volumes:
+      - ./dev/import:/opt/app/import
+    cap_drop:
+      - all
+
+  database:
+    image: postgres:16
+    restart: always
+    environment:
+      - POSTGRES_USER=teslamate
+      - POSTGRES_PASSWORD=password #insert your secure database password!
+      - POSTGRES_DB=teslamate
+    volumes:
+      - ./dev/teslamate-db:/var/lib/postgresql/data
+
+  mosquitto:
+    image: eclipse-mosquitto:2
+    restart: always
+    command: mosquitto -c /mosquitto-no-auth.conf
+    # ports:
+    #   - 1883:1883
+    volumes:
+      - ./dev/mosquitto-conf:/mosquitto/config
+      - ./dev/mosquitto-data:/mosquitto/data
+
+networks:
+  internal:

interfaces/backendfactory.py

@@ -0,0 +1,26 @@
+from enum import Enum
+import importlib
+from backendproviders import teslalogger, teslamate
+from interfaces.backendinterface import IBackendProvider
+
+class BackendProviderFactory:
+    # static to have a singleton-like
+    provider: IBackendProvider
+
+    def _load_provider(self, name, base_url, car_id):
+        if name == "teslalogger":
+            return teslalogger.TeslaloggerBackendProvider(base_url, car_id)
+        elif name == "teslamate":
+            return teslamate.TeslamateBackendProvider(base_url, car_id)
+        else:
+            raise Exception(f"Unknown backend provider : {name}. Available choices are 'teslalogger' or 'teslamagte'.")
+
+
+    def __init__(self, provider_name, base_url, car_id):
+        self.provider_name = provider_name
+
+        BackendProviderFactory.provider = self._load_provider(provider_name, base_url, car_id)
+
+    @staticmethod
+    def get_instance() -> IBackendProvider:
+        return BackendProviderFactory.provider

interfaces/backendinterface.py

@@ -0,0 +1,33 @@
+from abc import ABC, abstractmethod
+
+class IBackendProvider(ABC):
+    base_url: str = None
+    car_id: int = None
+
+    latitude: float = None
+    longitude: float = None
+    odometer: float = None
+    is_driving: bool = False
+    is_charging: bool = False
+    battery_level: int = 0
+
+    active_route_destination: str = None
+    active_route_latitude: float = None
+    active_route_longitude: float = None
+    active_route_minutes_to_arrival: float = None
+    active_route_energy_at_arrival: int = None
+
+
+    def __init__(self, base_url, car_id):
+        self.base_url = base_url
+        self.car_id = car_id
+
+        print(f"Starting provider. BASE_URL : {base_url}, CAR_ID : {car_id}")
+
+    @abstractmethod
+    def refresh_data(self):
+        pass
+
+    @property
+    def active_route_seconds_to_arrival(self) -> float:
+        return self.active_route_minutes_to_arrival * 60

requirements.txt

@@ -18,4 +18,4 @@ Werkzeug==3.0.3
 wheel==0.38.4
 zipp==3.19.1
 flask-login==0.6.3
-geopy==2.3.0
+geopy==2.3.0