mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-02-19 05:24:06 +08:00
148eaf8fa6
* 3.12 * pprofile is broken * use modified metadrivepy3-12 * 0.3.0 metadrive * add metadrive/commaai git dependency * metadrive git set * pin sounddevice 0.4.6 * datetime.utcnow() deprecation * poetry lock * make datetime not aware * poetry lock * pin pytools * google_crc32c wheel * unpin sounddevice * clean metadrive * use python crc * mypy * 3.12.4 * allow python3.11 * test * no pip * poetry * better * better * merge * remove * try * test * try this * snok * python * simpler * setuptools * lower * try * try * work? * ubuntu deps * ubuntu * try * remove * move * remove this * names * Update .github/workflows/tools_tests.yaml Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com> * python<4 * <3.13 --------- Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com> Co-authored-by: Maxime Desroches <desroches.maxime@gmail.com>
47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
import jwt
|
|
import os
|
|
import requests
|
|
from datetime import datetime, timedelta, UTC
|
|
from openpilot.system.hardware.hw import Paths
|
|
from openpilot.system.version import get_version
|
|
|
|
API_HOST = os.getenv('API_HOST', 'https://api.commadotai.com')
|
|
|
|
class Api:
|
|
def __init__(self, dongle_id):
|
|
self.dongle_id = dongle_id
|
|
with open(Paths.persist_root()+'/comma/id_rsa') as f:
|
|
self.private_key = f.read()
|
|
|
|
def get(self, *args, **kwargs):
|
|
return self.request('GET', *args, **kwargs)
|
|
|
|
def post(self, *args, **kwargs):
|
|
return self.request('POST', *args, **kwargs)
|
|
|
|
def request(self, method, endpoint, timeout=None, access_token=None, **params):
|
|
return api_get(endpoint, method=method, timeout=timeout, access_token=access_token, **params)
|
|
|
|
def get_token(self, expiry_hours=1):
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
payload = {
|
|
'identity': self.dongle_id,
|
|
'nbf': now,
|
|
'iat': now,
|
|
'exp': now + timedelta(hours=expiry_hours)
|
|
}
|
|
token = jwt.encode(payload, self.private_key, algorithm='RS256')
|
|
if isinstance(token, bytes):
|
|
token = token.decode('utf8')
|
|
return token
|
|
|
|
|
|
def api_get(endpoint, method='GET', timeout=None, access_token=None, **params):
|
|
headers = {}
|
|
if access_token is not None:
|
|
headers['Authorization'] = "JWT " + access_token
|
|
|
|
headers['User-Agent'] = "openpilot-" + get_version()
|
|
|
|
return requests.request(method, API_HOST + "/" + endpoint, timeout=timeout, headers=headers, params=params)
|