FordConnect API

auth.py

import logging
from dataclasses import dataclass
from typing import Dict

import requests


logger = logging.getLogger(__name__)


@dataclass
class FordAuth:
    client_id: str
    client_secret: str
    url: str = (
        "https://dah2vb2cprod.b2clogin.com/914d88b1-3523-4bf6-9be4-1b96b4f6f919/oauth2/v2.0/token?p=B2C_1A_signup_signin_common"
    )

    def from_code(
        self,
        code: str,
    ) -> Dict[str, str]:
        """Get the access token & refresh token from a code."""
        return self._fetch(
            grant_type="authorization_code",
            code=code,
            redirect_uri="https%3A%2F%2Flocalhost%3A3000",
        )

    def from_refresh_token(self, refresh_token: str) -> Dict[str, str]:
        """Refresh the access token & refresh token."""
        return self._fetch(
            grant_type="refresh_token",
            refresh_token=refresh_token,
        )

    def _fetch(
        self,
        *,
        grant_type: str,
        **payload,
    ) -> Dict[str, str]:
        response = requests.post(
            self.url,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            data={
                "grant_type": grant_type,
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                **payload,
            },
        )
        response.raise_for_status()
        return response.json()

connect.py

import logging
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Dict, Any, Optional, ByteString, Literal
from time import sleep, time

import requests

from . import responses

logger = logging.getLogger(__name__)


@dataclass
class FordConnect:
    access_token: str
    api_url: str = "https://api.mps.ford.com/api/fordconnect"
    max_request_interval: float = 3
    last_request: Optional[time] = None

    @property
    def _headers(self):
        return {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Application-Id": "AFDC085B-377A-4351-B23E-5E1D35FB3700",
            "Authorization": f"Bearer {self.access_token}",
        }

    @contextmanager
    def _throttle(self):
        if self.last_request and time() - self.last_request < self.max_request_interval:
            wait_time = self.max_request_interval - (time() - self.last_request)
            logger.debug(f"Waiting {wait_time} seconds before sending request.")
            sleep(wait_time)
        try:
            yield
        finally:
            self.last_request = time()

    def _req(
        self, url: str, method: str = "GET", data: Dict[str, Any] = None, as_bytes=False
    ) -> Any:
        logger.info(f"Sending {method} request to {url}")
        with self._throttle():
            response = requests.request(method, url, headers=self._headers, json=data)
        logger.debug(
            f"Response: {response.status_code} - {'BYTES' if as_bytes else response.text}"
        )
        response.raise_for_status()
        if as_bytes:
            return response.content
        return response.json()

    def get_vehicle_list(self) -> responses.VehicleListResponse:
        return self._req(
            url=f"{self.api_url}/v2/vehicles",
        )

    def get_vehicle_info(self, vehicle_id: str) -> responses.VehicleInfoResponse:
        return self._req(
            url=f"{self.api_url}/v3/vehicles/{vehicle_id}",
        )

    def get_vehicle_capabilities(
        self, vehicle_id: str
    ) -> responses.VehicleCapabilitiesResponse:
        return self._req(
            url=f"{self.api_url}/v3/vehicles/{vehicle_id}/capabilities",
        )

    def get_vehicle_location(
        self, vehicle_id: str
    ) -> responses.VehicleLocationResponse:
        return self._req(
            url=f"{self.api_url}/v2/vehicles/{vehicle_id}/location",
        )

    def get_vehicle_image(
        self,
        vehicle_id: str,
        make: str,
        model: str,
        year: str,
        image_type: Literal["thumbnail", "full"] = "thumbnail",
    ) -> ByteString:
        return self._req(
            url=f"{self.api_url}/v1/vehicles/{vehicle_id}/images/{image_type}?make={make}&model={model}&year={year}",
            as_bytes=True,
        )

    def get_vehicle_image_url(
        self,
        vehicle_id: str,
    ) -> ByteString:
        return self._req(
            url=f"{self.api_url}/v1/vehicles/{vehicle_id}/images/imageUrl",
        )

    def get_health(self) -> responses.HealthStatusInfoResponse:
        return self._req(
            url=f"{self.api_url}/v1/health",
        )

    def _send_vehicle_command(
        self,
        vehicle_id: str,
        command: str,
        data: Dict[str, Any] = None,
        poll: bool = True,
        timeout: int = 30,
    ) -> responses.CommandResponse:
        response: responses.CommandResponse = self._req(
            url=f"{self.api_url}/v1/vehicles/{vehicle_id}/{command}",
            method="POST",
            data=data,
        )
        start = time()
        while poll and timeout > time() - start:
            if response["commandStatus"] not in ["QUEUED", "INPROGRESS"]:
                break
            response = self.get_command_status(
                vehicle_id=vehicle_id,
                command=command,
                command_id=response["commandId"],
            )
        return response

    def get_command_status(
        self, vehicle_id: str, command: str, command_id: str
    ) -> responses.CommandResponse:
        return self._req(
            url=f"{self.api_url}/v1/vehicles/{vehicle_id}/{command}/{command_id}",
        )

    def start_charge(
        self, vehicle_id: str, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="startCharge",
            poll=poll,
        )

    def stop_charge(
        self, vehicle_id: str, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="stopCharge",
            poll=poll,
        )

    def unlock_vehicle(
        self, vehicle_id: str, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="unlock",
            poll=poll,
        )

    def lock_vehicle(
        self, vehicle_id: str, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="lock",
            poll=poll,
        )

    def start_engine(
        self, vehicle_id: str, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="startEngine",
            poll=poll,
        )

    def stop_engine(
        self, vehicle_id: str, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="stopEngine",
            poll=poll,
        )

    def send_signal(
        self, vehicle_id: str, duration: int, poll: bool = True
    ) -> responses.CommandResponse:
        return self._send_vehicle_command(
            vehicle_id=vehicle_id,
            command="signal",
            data={"duration": duration},
            poll=poll,
        )

    def cancel_signal(self, vehicle_id: str, signal_id: str) -> Any:
        return self._req(
            url=f"{self.api_url}/v1/vehicles/{vehicle_id}/signal/{signal_id}",
            method="DELETE",
        )

example.py

import logging
import os

from .auth import FordAuth
from .connect import FordConnect


logger = logging.getLogger("fordconnect")
logger.setLevel(logging.INFO)
console_handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)


if __name__ == "__main__":
    FORD_CLIENT_ID = os.environ["FORD_CLIENT_ID"]
    FORD_CLIENT_SECRET = os.environ["FORD_CLIENT_SECRET"]
    FORD_REFRESH_TOKEN = os.environ["FORD_REFRESH_TOKEN"]

    # Get auth token from valid refresh token.
    logger.info("Getting access token...")
    auth = FordAuth(
        client_id=FORD_CLIENT_ID,
        client_secret=FORD_CLIENT_SECRET,
    ).from_refresh_token(FORD_REFRESH_TOKEN)

    # Now interact with FordConnect API.
    logger.info("Locking vehicles...")
    connect = FordConnect(auth["access_token"])
    for vehicle in connect.get_vehicle_list()["vehicles"]:
        status = connect.lock_vehicle(vehicle["vehicleId"])
        if status["commandStatus"] == "COMPLETED":
            logger.info("✅ Vehicle locked!")
        else:
            logger.error(status)
            logger.error("❌ Failed to lock vehicle.")

responses.py

from typing import TypedDict, List, Literal


class FuelLevel(TypedDict):
    value: float
    distanceToEmpty: float
    timestamp: str


class VehicleDetails(TypedDict):
    fuelLevel: FuelLevel
    mileage: float
    odometer: float


class RemoteStartStatus(TypedDict):
    status: str
    duration: int
    timeStamp: str


class IgnitionStatus(TypedDict):
    value: str
    timeStamp: str


class DoorStatus(TypedDict):
    vehicleDoor: str
    value: str
    vehicleOccupantRole: str
    timeStamp: str


class LockStatus(TypedDict):
    value: str
    timeStamp: str


class AlarmStatus(TypedDict):
    value: str
    timeStamp: str


class VehicleStatus(TypedDict):
    tirePressureWarning: bool
    deepSleepInProgress: bool
    firmwareUpgradeInProgress: bool
    remoteStartStatus: RemoteStartStatus
    ignitionStatus: IgnitionStatus
    doorStatus: List[DoorStatus]
    lockStatus: LockStatus
    alarmStatus: AlarmStatus


class VehicleLocation(TypedDict):
    speed: float
    direction: str
    timeStamp: str
    longitude: str
    latitude: str


class VehicleCapabilities(TypedDict):
    remoteLock: str
    remoteUnlock: str
    remoteStart: str
    remoteStop: str
    boundaryAlerts: str


class Vehicle(TypedDict):
    vehicleId: str
    make: str
    modelName: str
    modelYear: str
    color: str
    nickName: str
    modemEnabled: bool
    lastUpdated: str
    vehicleAuthorizationIndicator: int
    serviceCompatible: bool
    engineType: str
    vehicleDetails: VehicleDetails
    vehicleStatus: VehicleStatus
    vehicleLocation: VehicleLocation
    vehicleCapabilities: VehicleCapabilities


class VehicleInfoResponse(TypedDict):
    status: Literal["SUCCESS"]
    vehicle: Vehicle


class VehicleCapabilities(TypedDict):
    remoteLock: str
    remoteUnlock: str
    remoteStart: str
    remoteStop: str
    boundaryAlerts: str
    remoteChirpHonk: str
    remotePanicAlarm: str
    displayPreferredChargeTimes: str
    departureTimes: str
    globalStartStopCharge: str


class VehicleCapabilitiesResponse(TypedDict):
    status: Literal["SUCCESS"]
    vehicleCapabilities: VehicleCapabilities


class VehicleLocation(TypedDict):
    speed: float
    direction: str
    timeStamp: str
    longitude: str
    latitude: str


class VehicleLocationResponse(TypedDict):
    status: Literal["SUCCESS"]
    vehicleLocation: VehicleLocation


class HealthStatus(TypedDict):
    Command_Control: str
    Vehicle_Information: str
    Signal: str
    Media: str
    Revoke: str


class HealthStatusInfoResponse(TypedDict):
    healthStatus: HealthStatus


class CommandResponse(TypedDict):
    commandId: str
    status: Literal["SUCCESS", "FAILED"]
    commandStatus: Literal[
        "QUEUED",
        "INPROGRESS",
        "COMPLETED",
        "FAILED",
        "EMPTY",
        "TIMEOUT",
        "UNKNOWN",
        "COMMUNICATIONFAILED",
        "PENDINGRESPONSE",
        "MODEMINDEEPSLEEPMODE",
        "FIRMWAREUPGRADEINPROGRESS",
        "FAILEDDUETOINVEHICLESETTINGS",
    ]


class VehicleSummary(TypedDict):
    vehicleId: str
    make: str
    modelName: str
    modelYear: str
    color: str
    nickName: str
    modemEnabled: bool
    vehicleAuthorizationIndicator: int
    serviceCompatible: bool
    engineType: str


class VehicleListResponse(TypedDict):
    vehicles: List[VehicleSummary]


__all__ = ["VehicleListResponse"]