github/psa_car_controller
1
0
Fork 0
mirror of https://github.com/flobz/psa_car_controller.git synced 2026-08-21 08:56:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'flobz:master' into patch-1

Browse Source
This commit is contained in:
kireffoh
2024-02-26 09:30:13 +01:00
committed by GitHub
parent cdf65fa999 b477e4a34c
commit 48cbf2bfd6
9 changed files with 78 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -33,7 +33,7 @@ The official API is documented [here](https://developer.groupe-psa.io/webapi/b2c
- [Installation on Raspberry Pi with docker-compose (external Tutorial)](https://return2.net/opel-peugeot-electric-vehicle-set-charging-threshold-limit/)
## II. Use the API
Look at [API documentation](./docs/psacc_api)
Look at [API documentation](./docs/psacc_api.md)
## III. Use the dashboard
psa_car_controller/psa/connected_car_api/configuration.py
+1 -1
View File
@@ -72,7 +72,7 @@ class Configuration(object):
# Debug file location
self.logger_file = None
# Debug switch
self.debug = False
self.debug = True
# SSL/TLS verification
# Set this to false to skip verifying SSL certificate when calling API
psa_car_controller/psa/constants.py
+19 -6
View File
@@ -3,12 +3,20 @@ IMMEDIATE_CHARGE = "immediate"
PSA_CORRELATION_DATE_FORMAT = "%Y%m%d%H%M%S%f"
PSA_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
realm_info = {
"clientsB2CPeugeot": {"oauth_url": "https://idpcvs.peugeot.com/am/oauth2/access_token", "app_name": "MyPeugeot"},
"clientsB2CCitroen": {"oauth_url": "https://idpcvs.citroen.com/am/oauth2/access_token", "app_name": "MyCitroen"},
"clientsB2CDS": {"oauth_url": "https://idpcvs.driveds.com/am/oauth2/access_token", "app_name": "MyDS"},
"clientsB2COpel": {"oauth_url": "https://idpcvs.opel.com/am/oauth2/access_token", "app_name": "MyOpel"},
"clientsB2CPeugeot": {"oauth_url": "https://idpcvs.peugeot.com/am/oauth2/access_token", "app_name": "MyPeugeot",
"scheme": "mymap"},
"clientsB2CCitroen": {"oauth_url": "https://idpcvs.citroen.com/am/oauth2/access_token", "app_name": "MyCitroen",
"scheme": "mymacsdk"},
"clientsB2CDS": {"oauth_url": "https://idpcvs.driveds.com/am/oauth2/access_token", "app_name": "MyDS",
"scheme": "mymdssdk"},
"clientsB2COpel": {"oauth_url": "https://idpcvs.opel.com/am/oauth2/access_token", "app_name": "MyOpel",
"scheme": "mymopsdk"},
"clientsB2CVauxhall": {"oauth_url": "https://idpcvs.vauxhall.co.uk/am/oauth2/access_token",
"app_name": "MyVauxhall"}
"app_name": "MyVauxhall",
"scheme": "mymvxsdk",
}
}
MQTT_BRANDCODE = {"AP": "AP",
"AC": "AC",
@@ -28,7 +36,12 @@ DEFAULT_PRECONDITIONING_PROGRAM = {
"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}
}
AUTHORIZE_SERVICE = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize"
AUTHORIZE_SERVICE = {"clientsB2COpel": "https://idpcvs.opel.com/am/oauth2/authorize",
"clientsB2CPeugeot": "https://idpcvs.peugeot.com/am/oauth2/authorize",
"clientsB2CCitroen": "https://idpcvs.citroen.com/am/oauth2/authorize",
"clientsB2CDS": "https://idpcvs.driveds.com/am/oauth2/authorize",
"clientsB2CVauxhall": "https://idpcvs.vauxhall.co.uk/am/oauth2/authorize"
}
REMOTE_URL = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id="
BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", "app_name": "MyOpel"},
"com.psa.mym.mypeugeot": {"realm": "clientsB2CPeugeot", "brand_code": "AP", "app_name": "MyPeugeot"},
psa_car_controller/psa/oauth.py
+36 -3
View File
@@ -1,8 +1,13 @@
import logging
import hashlib
import secrets
import base64
from typing import Tuple
from http import HTTPStatus
from typing import Optional
from oauth2_client.credentials_manager import CredentialManager, ServiceInformation
from oauth2_client.credentials_manager import CredentialManager, ServiceInformation, OAuthError
from requests import Response, RequestException
from psa_car_controller.common.utils import rate_limit
@@ -22,8 +27,36 @@ class OpenIdCredentialManager(CredentialManager):
return {"grant_type": 'password', "username": login, "scope": ' '.join(self.service_information.scopes),
"password": password, "realm": realm}
def init_with_user_credentials_realm(self, login: str, password: str, realm: str):
self._token_request(self._grant_password_request_realm(login, password, realm), True)
def generate_sha256_pkce(self, length: int) -> Tuple[str, str]:
if not (43 <= length <= 128):
raise ValueError("Invalid length: %d" % length)
verifier = secrets.token_urlsafe(length)
encoded = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode('ascii')).digest())
challenge = encoded.decode('ascii')[:-1]
return verifier, challenge
def init_with_oauth2_redirect(self, scheme: str, country_code: str):
ret = ""
while True:
redir_uri = scheme + "://oauth2redirect/" + country_code.lower()
code_verifier, code_challenge = self.generate_sha256_pkce(64)
url = self.generate_authorize_url(redir_uri, secrets.token_urlsafe(16),
code_challenge=code_challenge, code_challenge_method="S256")
logger.info("Now login to this URL in a browser: %s", url)
try:
ret = input("\nCopy+paste the resulting mymXX-code (in F12 > Network, "
"when you hit the final OK button, 36 chars, UUID format): ")
logger.info("Try getting a token with code %s", ret)
assert len(ret) == 36, "Invalid code length"
self._token_request({"grant_type": 'authorization_code', "code": ret,
"redirect_uri": redir_uri, "code_verifier": code_verifier}, False)
except (OAuthError, AssertionError):
logger.exception("Failed to get a token")
if input("Retry ? yes/NO") == "yes":
continue
break
@staticmethod
def _is_token_expired(response: Response) -> bool:
psa_car_controller/psa/setup/app_decoder.py
+2 -2
View File
@@ -91,8 +91,8 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code,
# Psacc
psacc = PSAClient(None, apk_parser.client_id, apk_parser.client_secret,
None, customer_id, BRAND[package_name]["realm"],
country_code)
psacc.connect(client_email, client_password)
country_code, BRAND[package_name]["brand_code"])
psacc.connect()
psacc.save_config(name=config_prefix + "config.json")
res = psacc.get_vehicles()
psa_car_controller/psacc/application/psa_client.py
+5 -4
View File
@@ -31,14 +31,14 @@ logger = CustomLogger.getLogger(__name__)
class PSAClient:
def connect(self, user, password):
self.manager.init_with_user_credentials_realm(user, password, self.realm)
def connect(self):
self.manager.init_with_oauth2_redirect(realm_info[self.realm]["scheme"], self.country_code)
# pylint: disable=too-many-arguments
def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code,
proxies=None, weather_api=None, abrp=None, co2_signal_api=None):
brand=None, proxies=None, weather_api=None, abrp=None, co2_signal_api=None):
self.realm = realm
self.service_information = ServiceInformation(AUTHORIZE_SERVICE,
self.service_information = ServiceInformation(AUTHORIZE_SERVICE[self.realm],
realm_info[self.realm]['oauth_url'],
client_id,
client_secret,
@@ -60,6 +60,7 @@ class PSAClient:
self._record_enabled = False
self.weather_api = weather_api
self.country_code = country_code
self.brand = brand
self.info_callback = []
self.info_refresh_rate = 120
if abrp is None:
psa_car_controller/psacc/utils/utils.py
+2 -2
View File
@@ -9,7 +9,7 @@ TIMEOUT_IN_S = 10
def get_temp(latitude: str, longitude: str, api_key: str) -> float:
try:
if not (latitude is None or longitude is None or api_key is None):
if latitude and longitude and api_key:
weather_rep = requests.get("https://api.openweathermap.org/data/2.5/onecall",
params={"lat": latitude, "lon": longitude,
"exclude": "minutely,hourly,daily,alerts",
@@ -17,7 +17,7 @@ def get_temp(latitude: str, longitude: str, api_key: str) -> float:
"units": "metric"},
timeout=TIMEOUT_IN_S)
weather_res_json = weather_rep.json()
temp = weather_res_json.get("current", weather_res_json.get("main"))["temp"]
temp = weather_res_json.get("current", weather_res_json["main"])["temp"]
logger.debug("Temperature :%fc", temp)
return temp
except ConnectionError:
psa_car_controller/web/view/config_views.py
+10 -4
View File
@@ -61,10 +61,16 @@ login_config_layout = dbc.Row(dbc.Col(md=12, lg=2, className="m-3", children=[
color="secondary",
)]),
dbc.Row(dbc.Button("Submit", color="primary", id="submit-form")),
dbc.Row(
dbc.FormText(
"After submit be patient it can take some time...",
color="secondary")),
html.Div(html.P([
"1. After submit switch back to the CLI and wait for the login URL to be shown", html.Br(),
"2. Open that URL in a browser", html.Br(),
"3. Complete the login procedure there too", html.Br(),
"4. Open your browser's DevTools (F12) and then the 'Network' tab", html.Br(),
"5. Hit the final 'OK' button, under 'LOGIN SUCCESSFUL'", html.Br(),
"6. Find xxxx://oauth2redirect....?code=<copy this part>&scope=openid... "
"in the DevTools and paste it into the command line interface", html.Br(),
"7. Switch back to the browser and do the OTP config", html.Br()]
)),
dcc.Loading(
id="loading-2",
children=[html.Div([html.Div(id="form_result")])],
pyproject.toml
+2 -2
View File
@@ -12,7 +12,7 @@ include = [
[tool.poetry.dependencies]
python = ">=3.7.2, <4.0.0"
paho-mqtt = ">=1.5.0"
paho-mqtt = ">=1.5.0, <2.0.0"
dash = ">=2.9.0, <3.0.0"
dash-daq = "^0.5.0"
plotly = ">=5"
@@ -21,7 +21,7 @@ Werkzeug = ">=1.0.0"
Flask = ">=1.0.4"
dash-bootstrap-components = ">=1"
ConfigUpdater = ">=3.0"
oauth2-client = "^1.2.1"
oauth2-client = "^1.3.0"
requests = "^2.27.1"
pytz = "^2021.0"
argparse = "^1.4.0"