selfdrive.athena package

Subpackages

• selfdrive.athena.tests package
  • Submodules
  • selfdrive.athena.tests.helpers module
    • EchoSocket
      • EchoSocket.run()
    • HTTPRequestHandler
      • HTTPRequestHandler.do_PUT()
    • MockApi
      • MockApi.get_token()
    • MockResponse
    • MockWebsocket
      • MockWebsocket.recv()
      • MockWebsocket.send()
  • selfdrive.athena.tests.test_athenad module
    • TestAthenadMethods
      • TestAthenadMethods.setUp()
      • TestAthenadMethods.setUpClass()
      • TestAthenadMethods.test_cancelExpiry()
      • TestAthenadMethods.test_cancelUpload()
      • TestAthenadMethods.test_do_upload
      • TestAthenadMethods.test_do_upload_0()
      • TestAthenadMethods.test_do_upload_1()
      • TestAthenadMethods.test_echo()
      • TestAthenadMethods.test_getGithubUsername()
      • TestAthenadMethods.test_getMessage()
      • TestAthenadMethods.test_getSshAuthorizedKeys()
      • TestAthenadMethods.test_getVersion()
      • TestAthenadMethods.test_get_logs_to_send_sorted()
      • TestAthenadMethods.test_jsonrpc_handler()
      • TestAthenadMethods.test_listDataDirectory()
      • TestAthenadMethods.test_listUploadQueue()
      • TestAthenadMethods.test_listUploadQueueCurrent()
      • TestAthenadMethods.test_listUploadQueueEmpty()
      • TestAthenadMethods.test_startLocalProxy()
      • TestAthenadMethods.test_strip_bz2_extension()
      • TestAthenadMethods.test_uploadFileToUrl()
      • TestAthenadMethods.test_uploadFileToUrl_does_not_exist()
      • TestAthenadMethods.test_uploadFileToUrl_duplicate()
      • TestAthenadMethods.test_upload_handler()
      • TestAthenadMethods.test_upload_handler_retry
      • TestAthenadMethods.test_upload_handler_retry_0()
      • TestAthenadMethods.test_upload_handler_retry_1()
      • TestAthenadMethods.test_upload_handler_timeout()
      • TestAthenadMethods.test_upload_queue_persistence()
    • seed_athena_server()
    • with_upload_handler()
  • selfdrive.athena.tests.test_athenad_ping module
    • TestAthenadPing
      • TestAthenadPing.assertTimeout()
      • TestAthenadPing.athenad
      • TestAthenadPing.dongle_id
      • TestAthenadPing.exit_event
      • TestAthenadPing.params
      • TestAthenadPing.setUp()
      • TestAthenadPing.tearDown()
      • TestAthenadPing.tearDownClass()
      • TestAthenadPing.test_offroad()
      • TestAthenadPing.test_onroad()
    • wifi_radio()
  • selfdrive.athena.tests.test_registration module
    • TestRegistration
      • TestRegistration.setUp()
      • TestRegistration.test_missing_cache()
      • TestRegistration.test_no_keys()
      • TestRegistration.test_unregistered()
      • TestRegistration.test_valid_cache()
  • Module contents

Submodules

selfdrive.athena.athenad module

exception selfdrive.athena.athenad.AbortTransferException

Bases: Exception

class selfdrive.athena.athenad.UploadFile(fn: 'str', url: 'str', headers: 'dict[str, str]', allow_cellular: 'bool')

Bases: object

allow_cellular: bool

fn: str

classmethod from_dict(d: dict) → UploadFile

headers: dict[str, str]

url: str

class selfdrive.athena.athenad.UploadItem(path: 'str', url: 'str', headers: 'dict[str, str]', created_at: 'int', id: 'str | None', retry_count: 'int' = 0, current: 'bool' = False, progress: 'float' = 0, allow_cellular: 'bool' = False)

Bases: object

allow_cellular: bool = False

created_at: int

current: bool = False

classmethod from_dict(d: dict) → UploadItem

headers: dict[str, str]

id: str | None

path: str

progress: float = 0

retry_count: int = 0

url: str

class selfdrive.athena.athenad.UploadQueueCache

Bases: object

static cache(upload_queue: Queue[UploadItem]) → None

static initialize(upload_queue: Queue[UploadItem]) → None

selfdrive.athena.athenad.backoff(retries: int) → int

selfdrive.athena.athenad.cancelUpload(upload_id: str | list[str]) → dict[str, int | str]

selfdrive.athena.athenad.cb(sm, item, tid, end_event: Event, sz: int, cur: int) → None

selfdrive.athena.athenad.getGithubUsername() → str

selfdrive.athena.athenad.getMessage(service: str, timeout: int = 1000) → dict

selfdrive.athena.athenad.getNetworkMetered() → bool

selfdrive.athena.athenad.getNetworkType()

selfdrive.athena.athenad.getNetworks()

selfdrive.athena.athenad.getPublicKey() → str | None

selfdrive.athena.athenad.getSimInfo()

selfdrive.athena.athenad.getSshAuthorizedKeys() → str

selfdrive.athena.athenad.getVersion() → dict[str, str]

selfdrive.athena.athenad.get_logs_to_send_sorted() → list[str]

selfdrive.athena.athenad.handle_long_poll(ws: WebSocket, exit_event: Event | None) → None

selfdrive.athena.athenad.jsonrpc_handler(end_event: Event) → None

selfdrive.athena.athenad.listDataDirectory(prefix='') → list[str]

selfdrive.athena.athenad.listUploadQueue() → list[dict[str, str | bool | int | float | dict[str, str]]]

selfdrive.athena.athenad.log_handler(end_event: Event) → None

selfdrive.athena.athenad.main(exit_event: Event = None)

selfdrive.athena.athenad.retry_upload(tid: int, end_event: Event, increase_count: bool = True) → None

selfdrive.athena.athenad.scan_dir(path: str, prefix: str) → list[str]

selfdrive.athena.athenad.setNavDestination(latitude: int = 0, longitude: int = 0, place_name: str = None, place_details: str = None) → dict[str, int]

selfdrive.athena.athenad.setRouteViewed(route: str) → dict[str, int | str]

selfdrive.athena.athenad.startLocalProxy(global_end_event: Event, remote_ws_uri: str, local_port: int) → dict[str, int]

selfdrive.athena.athenad.stat_handler(end_event: Event) → None

selfdrive.athena.athenad.strip_bz2_extension(fn: str) → str

selfdrive.athena.athenad.takeSnapshot() → str | dict[str, str] | None

selfdrive.athena.athenad.uploadFileToUrl(fn: str, url: str, headers: dict[str, str]) → dict[str, int | list[dict[str, str | bool | int | float | dict[str, str]]] | list[str]]

selfdrive.athena.athenad.uploadFilesToUrls(files_data: list[dict[str, str | int | float | bool]]) → dict[str, int | list[dict[str, str | bool | int | float | dict[str, str]]] | list[str]]

selfdrive.athena.athenad.upload_handler(end_event: Event) → None

selfdrive.athena.athenad.ws_manage(ws: WebSocket, end_event: Event) → None

selfdrive.athena.athenad.ws_proxy_recv(ws: WebSocket, local_sock: socket, ssock: socket, end_event: Event, global_end_event: Event) → None

selfdrive.athena.athenad.ws_proxy_send(ws: WebSocket, local_sock: socket, signal_sock: socket, end_event: Event) → None

selfdrive.athena.athenad.ws_recv(ws: WebSocket, end_event: Event) → None

selfdrive.athena.athenad.ws_send(ws: WebSocket, end_event: Event) → None

selfdrive.athena.manage_athenad module

selfdrive.athena.manage_athenad.main()

selfdrive.athena.registration module

selfdrive.athena.registration.is_registered_device() → bool

selfdrive.athena.registration.register(show_spinner=False) → str | None

Module contents