Add more type hinting (#23595)

* Add more type hinting.

* Revert joystick_alert changes.

* Add typing to statsd.

* Update selfdrive/statsd.py

* Update selfdrive/test/test_fingerprints.py

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
pull/23607/head
Ryan 2022-01-21 18:11:16 -05:00 committed by GitHub
parent 0f95e605f5
commit aa9e635311
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 16 additions and 10 deletions

View File

@ -1,13 +1,15 @@
from typing import Dict, Tuple
from common.xattr import getxattr as getattr1 from common.xattr import getxattr as getattr1
from common.xattr import setxattr as setattr1 from common.xattr import setxattr as setattr1
cached_attributes = {} cached_attributes: Dict[Tuple, bytes] = {}
def getxattr(path, attr_name): def getxattr(path: str, attr_name: bytes) -> bytes:
if (path, attr_name) not in cached_attributes: if (path, attr_name) not in cached_attributes:
response = getattr1(path, attr_name) response = getattr1(path, attr_name)
cached_attributes[(path, attr_name)] = response cached_attributes[(path, attr_name)] = response
return cached_attributes[(path, attr_name)] return cached_attributes[(path, attr_name)]
def setxattr(path, attr_name, attr_value): def setxattr(path: str, attr_name: str, attr_value: bytes) -> None:
cached_attributes.pop((path, attr_name), None) cached_attributes.pop((path, attr_name), None)
return setattr1(path, attr_name, attr_value) return setattr1(path, attr_name, attr_value)

View File

@ -120,7 +120,7 @@ def manager_thread() -> None:
params = Params() params = Params()
ignore = [] ignore: List[str] = []
if params.get("DongleId", encoding='utf8') == UNREGISTERED_DONGLE_ID: if params.get("DongleId", encoding='utf8') == UNREGISTERED_DONGLE_ID:
ignore += ["manage_athenad", "uploader"] ignore += ["manage_athenad", "uploader"]
if os.getenv("NOBOARD") is not None: if os.getenv("NOBOARD") is not None:

View File

@ -4,6 +4,8 @@ import zmq
import time import time
from pathlib import Path from pathlib import Path
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import NoReturn
from common.params import Params from common.params import Params
from cereal.messaging import SubMaster from cereal.messaging import SubMaster
from selfdrive.swaglog import cloudlog from selfdrive.swaglog import cloudlog
@ -20,14 +22,14 @@ class StatLog:
def __init__(self): def __init__(self):
self.pid = None self.pid = None
def connect(self): def connect(self) -> None:
self.zctx = zmq.Context() self.zctx = zmq.Context()
self.sock = self.zctx.socket(zmq.PUSH) self.sock = self.zctx.socket(zmq.PUSH)
self.sock.setsockopt(zmq.LINGER, 10) self.sock.setsockopt(zmq.LINGER, 10)
self.sock.connect(STATS_SOCKET) self.sock.connect(STATS_SOCKET)
self.pid = os.getpid() self.pid = os.getpid()
def _send(self, metric: str): def _send(self, metric: str) -> None:
if os.getpid() != self.pid: if os.getpid() != self.pid:
self.connect() self.connect()
@ -37,13 +39,13 @@ class StatLog:
# drop :/ # drop :/
pass pass
def gauge(self, name: str, value: float): def gauge(self, name: str, value: float) -> None:
self._send(f"{name}:{value}|{METRIC_TYPE.GAUGE}") self._send(f"{name}:{value}|{METRIC_TYPE.GAUGE}")
def main(): def main() -> NoReturn:
dongle_id = Params().get("DongleId", encoding='utf-8') dongle_id = Params().get("DongleId", encoding='utf-8')
def get_influxdb_line(measurement: str, value: float, timestamp: datetime, tags: dict): def get_influxdb_line(measurement: str, value: float, timestamp: datetime, tags: dict) -> str:
res = f"{measurement}" res = f"{measurement}"
for k, v in tags.items(): for k, v in tags.items():
res += f",{k}={str(v)}" res += f",{k}={str(v)}"

View File

@ -1,6 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import os import os
import sys import sys
from typing import Dict, List
from common.basedir import BASEDIR from common.basedir import BASEDIR
# messages reserved for CAN based ignition (see can_ignition_hook function in panda/board/drivers/can) # messages reserved for CAN based ignition (see can_ignition_hook function in panda/board/drivers/can)
@ -62,7 +64,7 @@ def check_can_ignition_conflicts(fingerprints, brands):
if __name__ == "__main__": if __name__ == "__main__":
fingerprints = _get_fingerprints() fingerprints = _get_fingerprints()
fingerprints_flat = [] fingerprints_flat: List[Dict] = []
car_names = [] car_names = []
brand_names = [] brand_names = []
for brand in fingerprints: for brand in fingerprints: