mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-02-18 22:23:56 +08:00
5e857427ba
* enable some more flake8 checks
* some more quick ones
* bump opendbc
* e401
* e711 e712
* e115 e116
* e222
* e301
* remove that
* e129
* e701 e702
* e125 e131
* e227
* e306
* e262
* W503
* e713
* e704
* e731
* bump opendbc
* fix some e722
old-commit-hash: d9bf9f0a40
42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
import jwt
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from common.basedir import PERSIST
|
|
from selfdrive.version import version
|
|
|
|
class Api():
|
|
def __init__(self, dongle_id):
|
|
self.dongle_id = dongle_id
|
|
with open(PERSIST+'/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):
|
|
now = datetime.utcnow()
|
|
payload = {
|
|
'identity': self.dongle_id,
|
|
'nbf': now,
|
|
'iat': now,
|
|
'exp': now + timedelta(hours=1)
|
|
}
|
|
return jwt.encode(payload, self.private_key, algorithm='RS256').decode('utf8')
|
|
|
|
def api_get(endpoint, method='GET', timeout=None, access_token=None, **params):
|
|
backend = "https://api.commadotai.com/"
|
|
|
|
headers = {}
|
|
if access_token is not None:
|
|
headers['Authorization'] = "JWT "+access_token
|
|
|
|
headers['User-Agent'] = "openpilot-" + version
|
|
|
|
return requests.request(method, backend+endpoint, timeout=timeout, headers = headers, params=params)
|