diff --git a/selfdrive/loggerd/xattr_cache.py b/selfdrive/loggerd/xattr_cache.py index aa97f0c77..e72169250 100644 --- a/selfdrive/loggerd/xattr_cache.py +++ b/selfdrive/loggerd/xattr_cache.py @@ -1,13 +1,15 @@ +from typing import Dict, Tuple + from common.xattr import getxattr as getattr1 from common.xattr import setxattr as setattr1 -cached_attributes = {} -def getxattr(path, attr_name): +cached_attributes: Dict[Tuple, bytes] = {} +def getxattr(path: str, attr_name: bytes) -> bytes: if (path, attr_name) not in cached_attributes: response = getattr1(path, attr_name) cached_attributes[(path, attr_name)] = response 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) return setattr1(path, attr_name, attr_value) diff --git a/selfdrive/manager/manager.py b/selfdrive/manager/manager.py index 77fb9543d..3208b1b81 100755 --- a/selfdrive/manager/manager.py +++ b/selfdrive/manager/manager.py @@ -120,7 +120,7 @@ def manager_thread() -> None: params = Params() - ignore = [] + ignore: List[str] = [] if params.get("DongleId", encoding='utf8') == UNREGISTERED_DONGLE_ID: ignore += ["manage_athenad", "uploader"] if os.getenv("NOBOARD") is not None: diff --git a/selfdrive/statsd.py b/selfdrive/statsd.py index 0967c6ef9..2e62e3253 100755 --- a/selfdrive/statsd.py +++ b/selfdrive/statsd.py @@ -4,6 +4,8 @@ import zmq import time from pathlib import Path from datetime import datetime, timezone +from typing import NoReturn + from common.params import Params from cereal.messaging import SubMaster from selfdrive.swaglog import cloudlog @@ -20,14 +22,14 @@ class StatLog: def __init__(self): self.pid = None - def connect(self): + def connect(self) -> None: self.zctx = zmq.Context() self.sock = self.zctx.socket(zmq.PUSH) self.sock.setsockopt(zmq.LINGER, 10) self.sock.connect(STATS_SOCKET) self.pid = os.getpid() - def _send(self, metric: str): + def _send(self, metric: str) -> None: if os.getpid() != self.pid: self.connect() @@ -37,13 +39,13 @@ class StatLog: # drop :/ pass - def gauge(self, name: str, value: float): + def gauge(self, name: str, value: float) -> None: self._send(f"{name}:{value}|{METRIC_TYPE.GAUGE}") -def main(): +def main() -> NoReturn: 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}" for k, v in tags.items(): res += f",{k}={str(v)}" diff --git a/selfdrive/test/test_fingerprints.py b/selfdrive/test/test_fingerprints.py index 4ea60a76a..01c7d753c 100755 --- a/selfdrive/test/test_fingerprints.py +++ b/selfdrive/test/test_fingerprints.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 import os import sys +from typing import Dict, List + from common.basedir import BASEDIR # 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__": fingerprints = _get_fingerprints() - fingerprints_flat = [] + fingerprints_flat: List[Dict] = [] car_names = [] brand_names = [] for brand in fingerprints: