psa_car_controller/MyPSACC.py

import json
import re
import threading
import traceback
import uuid
from copy import copy
from datetime import datetime
from http import HTTPStatus
from json import JSONEncoder
from hashlib import md5
from time import sleep

from oauth2_client.credentials_manager import CredentialManager, ServiceInformation
import paho.mqtt.client as mqtt
from requests import Response
from typing import Tuple

import psa_connectedcar as psac
from Car import Cars, Car
from ecomix import Ecomix
from otp.Otp import load_otp, new_otp_session, save_otp, ConfigException, Otp
from psa_connectedcar import ApiClient
from psa_connectedcar.rest import ApiException
from MyLogger import logger

from utils import get_temp, rate_limit
from web.db import get_db, clean_position
from geojson import Feature, Point, FeatureCollection
from geojson import dumps as geo_dumps

PSA_CORRELATION_DATE_FORMAT = "%Y%m%d%H%M%S%f"
PSA_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"

oauhth_url = {"clientsB2CPeugeot": "https://idpcvs.peugeot.com/am/oauth2/access_token",
              "clientsB2CCitroen": "https://idpcvs.citroen.com/am/oauth2/access_token",
              "clientsB2CDS": "https://idpcvs.driveds.com/am/oauth2/access_token",
              "clientsB2COpel": "https://idpcvs.opel.com/am/oauth2/access_token",
              "clientsB2CVauxhall": "https://idpcvs.vauxhall.co.uk/am/oauth2/access_token"}

authorize_service = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize"
remote_url = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id="
scopes = ['openid profile']
MQTT_SERVER = "mwa.mpsa.com"
MQTT_REQ_TOPIC = "psa/RemoteServices/from/cid/"
MQTT_RESP_TOPIC = "psa/RemoteServices/to/cid/"
MQTT_EVENT_TOPIC = "psa/RemoteServices/events/MPHRTServices/"
MQTT_TOKEN_TTL = 890
CARS_FILE = "cars.json"


# add method to class Energy
def get_energy(self, energy_type):
    for energy in self._energy:
        if energy.type == energy_type:
            return energy
    return psac.models.energy.Energy(charging=psac.models.energy_charging.EnergyCharging())


psac.models.status.Status.get_energy = get_energy


class OpenIdCredentialManager(CredentialManager):
    def _grant_password_request(self, login: str, password: str, realm: str) -> dict:
        return dict(grant_type='password',
                    username=login,
                    scope=' '.join(self.service_information.scopes),
                    password=password, realm=realm)

    def init_with_user_credentials(self, login: str, password: str, realm: str):
        self._token_request(self._grant_password_request(login, password, realm), True)

    @staticmethod
    def _is_token_expired(response: Response) -> bool:
        if response.status_code == HTTPStatus.UNAUTHORIZED.value:
            logger.info("token expired, renew")
            try:
                json_data = response.json()
                return json_data.get('moreInformation') == 'Token is invalid'
            except ValueError:
                return False
        else:
            return False


class Oauth2PSACCApiConfig(psac.Configuration):
    def set_refresh_callback(self, callback):
        self.refresh_callback = callback


class OauthAPIClient(ApiClient):
    def call_api(self, resource_path, method,
                 path_params=None, query_params=None, header_params=None,
                 body=None, post_params=None, files=None,
                 response_type=None, auth_settings=None, async_req=None,
                 _return_http_data_only=None, collection_formats=None,
                 _preload_content=True, _request_timeout=None):
        for _ in range(0, 2):
            try:
                if not async_req:
                    return self._ApiClient__call_api(resource_path, method,
                                                     path_params, query_params, header_params,
                                                     body, post_params, files,
                                                     response_type, auth_settings,
                                                     _return_http_data_only, collection_formats,
                                                     _preload_content, _request_timeout)
                return self.pool.apply_async(self.__call_api, (resource_path,
                                                               method, path_params, query_params,
                                                               header_params, body,
                                                               post_params, files,
                                                               response_type, auth_settings,
                                                               _return_http_data_only,
                                                               collection_formats,
                                                               _preload_content, _request_timeout))
            except ApiException as e:
                if e.reason == 'Unauthorized':
                    self.configuration.refresh_callback()
                else:
                    raise e


def gen_correlation_id(date):
    date_str = date.strftime(PSA_CORRELATION_DATE_FORMAT)[:-3]
    uuid_str = str(uuid.uuid4()).replace("-", "")
    correlation_id = uuid_str + date_str
    return correlation_id


class MyPSACC:
    vehicles_url = "https://idpcvs.peugeot.com/api/connectedcar/v2/oauth/authorize"

    def connect(self, user, password):
        self.manager.init_with_user_credentials(user, password, self.realm)

    def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code,
                 proxies=None, weather_api=None):
        self.realm = realm
        self.service_information = ServiceInformation(authorize_service,
                                                      oauhth_url[self.realm],
                                                      client_id,
                                                      client_secret,
                                                      scopes, False)
        self.client_id = client_id
        self.manager = OpenIdCredentialManager(self.service_information)
        self.api_config = Oauth2PSACCApiConfig()
        self.api_config.set_refresh_callback(self.manager._refresh_token)
        self.manager.refresh_token = refresh_token
        self.remote_refresh_token = remote_refresh_token
        self.remote_access_token = None
        self.vehicles_list = Cars.load_cars(CARS_FILE)
        self.set_proxies(proxies)
        self.customer_id = customer_id
        self._config_hash = None
        self.api_config.verify_ssl = False
        self.api_config.api_key['client_id'] = self.client_id
        self.api_config.api_key['x-introspect-realm'] = self.realm
        self.headers = {
            "x-introspect-realm": realm,
            "accept": "application/hal+json",
            "User-Agent": "okhttp/4.8.0",
        }
        self.remote_token_last_update = None
        self._record_enabled = False
        self.otp = None
        self.weather_api = weather_api
        self.country_code = country_code
        self.mqtt_client = None
        self.precond_programs = {}
        self.info_callback = []
        self.info_refresh_rate = 120

    def refresh_token(self):
        self.manager._refresh_token()

    def api(self) -> psac.VehiclesApi:
        self.api_config.access_token = self.manager._access_token
        api_instance = psac.VehiclesApi(OauthAPIClient(self.api_config))
        return api_instance

    def set_proxies(self, proxies):
        if proxies is None:
            self._proxies = dict(http='', https='')
            self.api_config.proxy = None
        else:
            self._proxies = proxies
            self.api_config.proxy = proxies['http']
        self.manager.proxies = self._proxies
        Otp.set_proxies(proxies)

    def get_vehicle_info(self, vin):
        res = None
        car = self.vehicles_list.get_car_by_vin(vin)
        for _ in range(0, 2):
            try:
                res = self.api().get_vehicle_status(car.vehicle_id, extension=["odometer"])
                if res is not None:
                    if self._record_enabled:
                        self.record_info(vin, res)
                    break
            except ApiException:
                logger.error(traceback.format_exc())
        car.status = res
        return res

    def refresh_vehicle_info(self):
        if self.info_refresh_rate is not None:
            while True:
                sleep(self.info_refresh_rate)
                logger.info("refresh_vehicle_info")
                for car in self.vehicles_list:
                    self.get_vehicle_info(car.vin)
                for callback in self.info_callback:
                    callback()

    # monitor doesn't seem to work
    def new_monitor(self, vin, body):
        res = self.manager.post("https://api.groupe-psa.com/connectedcar/v4/user/vehicles/" +
                                self.vehicles_list.get_car_by_vin(vin).id + "/status?client_id=" + self.client_id,
                                headers=self.headers, data=body)
        data = res.json()
        return data

    def get_vehicles(self):
        try:
            res = self.api().get_vehicles_by_device()
            for vehicle in res.embedded.vehicles:
                self.vehicles_list.add(Car(vehicle.vin, vehicle.id, vehicle.brand, vehicle.label))
            self.vehicles_list.save_cars()
        except ApiException:
            logger.error(traceback.format_exc())
        return self.vehicles_list

    def load_otp(self, force_new=False):
        otp_session = load_otp()
        if otp_session is None or force_new:
            self.get_sms_otp_code()
            otp_session = new_otp_session(otp_session)
        self.otp = otp_session

    def get_sms_otp_code(self):
        res = self.manager.post(
            "https://api.groupe-psa.com/applications/cvs/v4/mobile/smsCode?client_id=" + self.client_id,
            headers={
                "Connection": "Keep-Alive",
                "User-Agent": "okhttp/4.8.0",
                "x-introspect-realm": self.realm
            })
        return res

    # 6 otp by day
    @rate_limit(6, 3600 * 24)
    def get_otp_code(self):
        try:
            otp_code = self.otp.get_otp_code()
        except ConfigException:
            self.load_otp(force_new=True)
            otp_code = self.otp.get_otp_code()
        save_otp(self.otp)
        return otp_code

    def get_remote_access_token(self, password):
        res = self.manager.post(remote_url + self.client_id,
                                json={"grant_type": "password", "password": password},
                                headers=self.headers)
        data = res.json()
        self.remote_access_token = data["access_token"]
        self.remote_refresh_token = data["refresh_token"]
        return res

    def refresh_remote_token(self, force=False):
        if not force and self.remote_token_last_update is not None:
            last_update: datetime = self.remote_token_last_update
            if (datetime.now() - last_update).total_seconds() < MQTT_TOKEN_TTL:
                return None
        self.manager._refresh_token()
        if self.remote_refresh_token is None:
            logger.error("remote_refresh_token isn't defined")
            self.load_otp(force_new=True)
        res = self.manager.post(remote_url + self.client_id,
                                json={"grant_type": "refresh_token", "refresh_token": self.remote_refresh_token},
                                headers=self.headers)
        data = res.json()
        logger.debug("refresh_remote_token: %s", data)
        if "access_token" in data:
            self.remote_access_token = data["access_token"]
            self.remote_refresh_token = data["refresh_token"]
            self.remote_token_last_update = datetime.now()
        else:
            logger.error("can't refresh_remote_token: %s\n Create a new one", data)
            self.remote_token_last_update = datetime.now()
            otp_code = self.get_otp_code()
            res = self.get_remote_access_token(otp_code)
        self.mqtt_client.username_pw_set("IMA_OAUTH_ACCESS_TOKEN", self.remote_access_token)
        return res

    def on_mqtt_connect(self, client, userdata, rc, a):
        logger.info("Connected with result code %s", rc)
        topics = [MQTT_RESP_TOPIC + self.customer_id + "/#"]
        for car in self.vehicles_list:
            topics.append(MQTT_EVENT_TOPIC + car.vin)
        for topic in topics:
            client.subscribe(topic)
            logger.info("subscribe to %s", topic)

    def on_mqtt_disconnect(self, client, userdata, rc):
        logger.warning("Disconnected with result code %d", rc)
        if rc == 1:
            self.refresh_remote_token(force=True)
        else:
            logger.warning(mqtt.error_string(rc))

    def on_mqtt_message(self, client, userdata, msg):
        try:
            logger.info("mqtt msg %s %s", msg.topic, msg.payload)
            data = json.loads(msg.payload)
            charge_info = None
            if msg.topic.startswith(MQTT_RESP_TOPIC):
                if "return_code" not in data:
                    logger.debug("mqtt msg hasn't return code")
                elif data["return_code"] == "400":
                    self.refresh_remote_token(force=True)
                    logger.error("retry last request, token was expired")
                elif data["return_code"] == "300":
                    logger.error('%d', data["return_code"])
                elif data["return_code"] != "0":
                    logger.error('%s : %s', data["return_code"], data["reason"])
                    if msg.topic.endswith("/VehicleState"):
                        charge_info = data["resp_data"]["charging_state"]
                        self.precond_programs[data["vin"]] = data["resp_data"]["precond_state"]["programs"]
            elif msg.topic.startswith(MQTT_EVENT_TOPIC):
                charge_info = data["charging_state"]
            if charge_info is not None and charge_info['remaining_time'] != 0 and charge_info['rate'] == 0:
                # fix a psa server bug where charge beginning without status api being properly updated
                logger.warning("charge begin but API isn't updated")
                sleep(60)
                self.wakeup(data["vin"])
        except KeyError:
            logger.error(traceback.format_exc())

    def start_mqtt(self):
        self.load_otp()
        self.mqtt_client = mqtt.Client(clean_session=True, protocol=mqtt.MQTTv311)
        self.refresh_remote_token()
        self.mqtt_client.tls_set_context()
        self.mqtt_client.on_connect = self.on_mqtt_connect
        self.mqtt_client.on_message = self.on_mqtt_message
        self.mqtt_client.on_disconnect = self.on_mqtt_disconnect
        self.mqtt_client.connect(MQTT_SERVER, 8885, 60)
        self.mqtt_client.loop_start()
        self.__keep_mqtt()
        return self.mqtt_client.is_connected()

    def __keep_mqtt(self):  # avoid token expiration
        timeout = 3600 * 24  # 1 day
        if len(self.vehicles_list) > 0:
            self.get_state(self.vehicles_list[0].vin)
        t = threading.Timer(timeout, self.__keep_mqtt)
        t.setDaemon(True)
        t.start()

    def mqtt_request(self, vin, req_parameters):
        self.refresh_token()
        date = datetime.utcnow()
        date_str = date.strftime(PSA_DATE_FORMAT)
        data = {"access_token": self.remote_access_token, "customer_id": self.customer_id,
                "correlation_id": gen_correlation_id(date), "req_date": date_str, "vin": vin,
                "req_parameters": req_parameters}

        return json.dumps(data)

    def get_charge_hour(self, vin):
        reg = r"PT([0-9]{1,2})H([0-9]{1,2})?"
        data = self.get_vehicle_info(vin)
        hour_str = data.get_energy('Electric').charging.next_delayed_time
        try:
            hour = re.findall(reg, hour_str)[0]
            h = int(hour[0])
            if hour[1] == '':
                m = 0
            else:
                m = hour[1]
            return h, m
        except IndexError:
            logger.error(traceback.format_exc())
            logger.error("Can't get charge hour: %s", hour_str)

    def get_charge_status(self, vin):
        data = self.get_vehicle_info(vin)
        status = data.get_energy('Electric').charging.status
        return status

    def veh_charge_request(self, vin, hour, miinute, charge_type):
        # todo consider actual state before change the hour
        msg = self.mqtt_request(vin, {"program": {"hour": hour, "minute": miinute}, "type": charge_type})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehCharge", msg)

    def change_charge_hour(self, vin, hour, miinute):
        # todo consider actual state before change the hour
        self.veh_charge_request(vin, hour, miinute, "delayed")
        return True

    def charge_now(self, vin, now):
        if now:
            charge_type = "immediate"
        else:
            charge_type = "delayed"
        hour, minute = self.get_charge_hour(vin)
        self.veh_charge_request(vin, hour, minute, charge_type)
        return True

    def horn(self, vin, count):
        msg = self.mqtt_request(vin, {"nb_horn": count, "action": "activate"})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/Horn", msg)

    def lights(self, vin, duration: int):
        msg = self.mqtt_request(vin, {"action": "activate", "duration": duration})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/Lights", msg)

    @rate_limit(3, 60 * 20)
    def wakeup(self, vin):
        logger.info("ask wakeup to %s", vin)
        msg = self.mqtt_request(vin, {"action": "state"})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehCharge/state", msg)
        return True

    # get state from server by mqtt
    def get_state(self, vin):
        logger.info("ask state to %s", vin)
        msg = self.mqtt_request(vin, {"action": "state"})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehicleState", msg)
        return True

    def lock_door(self, vin, lock: bool):
        if lock:
            value = "lock"
        else:
            value = "unlock"

        msg = self.mqtt_request(vin, {"action": value})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/Doors", msg)
        return True

    def preconditioning(self, vin, activate: bool):
        if activate:
            value = "activate"
        else:
            value = "deactivate"
        self.get_state(vin)
        sleep(2)  # wait for rep
        if vin in self.precond_programs:
            programs = self.precond_programs[vin]
        else:
            programs = {
                "program1": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0},
                "program2": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0},
                "program3": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0},
                "program4": {"day": [0, 0, 0, 0, 0, 0, 0], "hour": 34, "minute": 7, "on": 0}
            }
        msg = self.mqtt_request(vin, {"asap": value, "programs": programs})
        logger.info(msg)
        self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/ThermalPrecond", msg)
        return True

    def save_config(self, name="config.json", force=False):
        config_str = json.dumps(self, cls=MyPeugeotEncoder, sort_keys=True, indent=4).encode("utf8")
        new_hash = md5(config_str).hexdigest()
        if force or self._config_hash != new_hash:
            with open(name, "wb") as f:
                f.write(config_str)
            self._config_hash = new_hash
            logger.info("save config change")

    @staticmethod
    def load_config(name="config.json"):
        with open(name, "r") as f:
            config_str = f.read()
            config = dict(**json.loads(config_str))
            if "country_code" not in config:
                config["country_code"] = input("What is your country code ? (ex: FR, GB, DE, ES...)\n")
            return MyPSACC(**config)

    def set_record(self, value: bool):
        self._record_enabled = value

    def record_info(self, vin, status: psac.models.status.Status):
        mileage = status.timed_odometer.mileage
        level = status.get_energy('Electric').level
        charging_status = status.get_energy('Electric').charging.status
        charge_date = status.get_energy('Electric').updated_at
        level_fuel = status.get_energy('Fuel').level
        try:
            longitude = status.last_position.geometry.coordinates[0]
            latitude = status.last_position.geometry.coordinates[1]
            date = status.last_position.properties.updated_at
        except AttributeError:
            logger.error("last_position not available from api")
            longitude = latitude = None
            date = charge_date
        try:
            moving = status.kinetic.moving
        except AttributeError:
            logger.error("kinetic not available from api")
            moving = None
        logger.debug(
            "vin:%s longitude:%s latitude:%s date:%s mileage:%s level:%s charging_status:%s charge_date:%s level_fuel:"
            "%s moving:%s", vin, longitude, latitude, date, mileage, level, charging_status, charge_date, level_fuel,
            moving)
        conn = get_db()
        if mileage == 0:  # fix a bug of the api
            logger.error("The api return a wrong mileage for %s : %f", vin, mileage)
        else:
            if conn.execute("SELECT Timestamp from position where Timestamp=?", (date,)).fetchone() is None:
                temp = get_temp(latitude,longitude,self.weather_api)
                if level_fuel == 0:  # fix fuel level not provided when car is off
                    try:
                        level_fuel = conn.execute(
                            "SELECT level_fuel FROM position WHERE level_fuel>0 AND VIN=? ORDER BY Timestamp DESC LIMIT 1",
                            (vin,)).fetchone()[0]
                        logger.info("level_fuel fixed with last real value %f for %s", level_fuel, vin)
                    except TypeError:
                        level_fuel = None
                        logger.info("level_fuel unfixed for %s", vin)

                conn.execute(
                    "INSERT INTO position(Timestamp,VIN,longitude,latitude,mileage,level,level_fuel,moving,temperature) "
                    "VALUES(?,?,?,?,?,?,?,?,?)",
                    (date, vin, longitude, latitude, mileage, level, level_fuel, moving, temp))

                conn.commit()
                logger.info("new position recorded for %s", vin)
                clean_position(conn)
            else:
                logger.debug("position already saved")

        if charging_status == "InProgress":
            try:
                in_progress = conn.execute("SELECT stop_at FROM battery WHERE VIN=? ORDER BY start_at DESC limit 1",
                                           (vin,)).fetchone()[0] is None
            except TypeError:
                in_progress = False
            if not in_progress:
                conn.execute("INSERT INTO battery(start_at,start_level,VIN) VALUES(?,?,?)", (charge_date, level, vin))
                conn.commit()
        else:
            try:
                start_at, stop_at, start_level = conn.execute(
                    "SELECT start_at, stop_at, start_level from battery WHERE VIN=? ORDER BY start_at "
                    "DESC limit 1", (vin,)).fetchone()
                in_progress = stop_at is None
                if in_progress:
                    co2_per_kw = Ecomix.get_co2_per_kw(start_at, charge_date, latitude, longitude)
                    kw = (level - start_level) / 100 * self.vehicles_list.get_car_by_vin(vin).battery_power
                    conn.execute(
                        "UPDATE battery set stop_at=?, end_level=?, co2=?, kw=? WHERE start_at=? and VIN=?",
                        (charge_date, level, co2_per_kw, kw, start_at, vin))
                    conn.commit()
            except TypeError:
                logger.debug("battery table is empty")
        conn.close()

    @staticmethod
    def get_recorded_position():
        conn = get_db()
        res = conn.execute('SELECT * FROM position ORDER BY Timestamp')
        features_list = []
        for row in res:
            if row["longitude"] is None or row["latitude"] is None:
                continue
            feature = Feature(geometry=Point((row["longitude"], row["latitude"])),
                              properties={"vin": row["vin"], "date": row["Timestamp"].strftime("%x %X"),
                                          "mileage": row["mileage"],
                                          "level": row["level"], "level_fuel": row["level_fuel"]})
            features_list.append(feature)
        feature_collection = FeatureCollection(features_list)
        conn.close()
        return geo_dumps(feature_collection, sort_keys=True)

    @staticmethod
    def get_chargings(mini=None, maxi=None) -> Tuple[dict]:
        conn = get_db()
        if mini is not None:
            if maxi is not None:
                res = conn.execute("select * from battery WHERE start_at>=? and start_at<=?", (mini, maxi)).fetchall()
            else:
                res = conn.execute("select * from battery WHERE start_at>=?", (mini,)).fetchall()
        elif maxi is not None:
            res = conn.execute("select * from battery WHERE start_at<=?", (maxi,)).fetchall()
        else:
            res = conn.execute("select * from battery").fetchall()
        return tuple(map(dict, res))


class MyPeugeotEncoder(JSONEncoder):
    def default(self, mp: MyPSACC):
        data = copy(mp.__dict__)
        mpd = {"proxies": data["_proxies"], "refresh_token": mp.manager.refresh_token,
               "client_secret": mp.service_information.client_secret}
        for el in ["client_id", "realm", "remote_refresh_token", "customer_id", "weather_api", "country_code"]:
            mpd[el] = data[el]
        return mpd