#!/usr/bin/env python3
# type: ignore

import cereal.messaging as messaging
from cereal.services import service_list

TO_CHECK = ['carState']



            

Reported by Pylint.

              #!/usr/bin/env python3
# type: ignore

import cereal.messaging as messaging
from cereal.services import service_list

TO_CHECK = ['carState']



            

Reported by Pylint.

              # type: ignore

import cereal.messaging as messaging
from cereal.services import service_list

TO_CHECK = ['carState']


if __name__ == "__main__":

            

Reported by Pylint.

              # type: ignore

import cereal.messaging as messaging
from cereal.services import service_list

TO_CHECK = ['carState']


if __name__ == "__main__":

            

Reported by Pylint.

              #!/usr/bin/env python3
import sys
import subprocess
from azure.storage.blob import BlockBlobService

from selfdrive.test.test_routes import routes as test_car_models_routes
from selfdrive.test.process_replay.test_processes import original_segments as replay_segments
from xx.chffr.lib import azureutil  # pylint: disable=import-error
from xx.chffr.lib.storage import _DATA_ACCOUNT_PRODUCTION, _DATA_ACCOUNT_CI, _DATA_BUCKET_PRODUCTION  # pylint: disable=import-error

            

Reported by Pylint.

              #!/usr/bin/env python3
import sys
import subprocess
from azure.storage.blob import BlockBlobService

from selfdrive.test.test_routes import routes as test_car_models_routes
from selfdrive.test.process_replay.test_processes import original_segments as replay_segments
from xx.chffr.lib import azureutil  # pylint: disable=import-error
from xx.chffr.lib.storage import _DATA_ACCOUNT_PRODUCTION, _DATA_ACCOUNT_CI, _DATA_BUCKET_PRODUCTION  # pylint: disable=import-error

            

Reported by Bandit.

                  "--exclude-pattern=*/dcamera.hevc",
    "--exclude-pattern=*.mkv",
  ]
  subprocess.check_call(cmd)

def sync_to_ci_public(route):
  key_prefix = route.replace('|', '/')
  dongle_id = key_prefix.split('/')[0]


            

Reported by Bandit.

                  ]

    try:
      result = subprocess.call(cmd, stdout=subprocess.DEVNULL)
      if result == 0:
        print("Success")
        return True
    except subprocess.CalledProcessError:
      print("Failed")

            

Reported by Bandit.

              #!/usr/bin/env python3
import unittest
from unittest.mock import patch
from parameterized import parameterized

from cereal import log
import cereal.messaging as messaging
from common.params import Params
params = Params()

            

Reported by Pylint.

              from unittest.mock import patch
from parameterized import parameterized

from cereal import log
import cereal.messaging as messaging
from common.params import Params
params = Params()

# Create fake time

            

Reported by Pylint.

              from parameterized import parameterized

from cereal import log
import cereal.messaging as messaging
from common.params import Params
params = Params()

# Create fake time
ssb = 0

            

Reported by Pylint.

              from parameterized import parameterized

from cereal import log
import cereal.messaging as messaging
from common.params import Params
params = Params()

# Create fake time
ssb = 0

            

Reported by Pylint.

              from functools import partial
from typing import Optional

import cereal.messaging as messaging
from selfdrive.swaglog import cloudlog
from selfdrive.boardd.boardd import can_list_to_can_capnp
from panda.python.uds import CanClient, IsoTpMessage, FUNCTIONAL_ADDRS, get_rx_addr_for_tx_addr



            

Reported by Pylint.

              from functools import partial
from typing import Optional

import cereal.messaging as messaging
from selfdrive.swaglog import cloudlog
from selfdrive.boardd.boardd import can_list_to_can_capnp
from panda.python.uds import CanClient, IsoTpMessage, FUNCTIONAL_ADDRS, get_rx_addr_for_tx_addr



            

Reported by Pylint.

              import cereal.messaging as messaging
from selfdrive.swaglog import cloudlog
from selfdrive.boardd.boardd import can_list_to_can_capnp
from panda.python.uds import CanClient, IsoTpMessage, FUNCTIONAL_ADDRS, get_rx_addr_for_tx_addr


class IsoTpParallelQuery:
  def __init__(self, sendcan, logcan, bus, addrs, request, response, response_offset=0x8, functional_addr=False, debug=False):
    self.sendcan = sendcan

            

Reported by Pylint.

              import cereal.messaging as messaging
from selfdrive.swaglog import cloudlog
from selfdrive.boardd.boardd import can_list_to_can_capnp
from panda.python.uds import CanClient, IsoTpMessage, FUNCTIONAL_ADDRS, get_rx_addr_for_tx_addr


class IsoTpParallelQuery:
  def __init__(self, sendcan, logcan, bus, addrs, request, response, response_offset=0x8, functional_addr=False, debug=False):
    self.sendcan = sendcan

            

Reported by Pylint.

              import unittest
import numpy as np

from cereal import log
import cereal.messaging as messaging
from selfdrive.config import Conversions as CV
from selfdrive.controls.lib.lead_mpc import LeadMpc



            

Reported by Pylint.

              import numpy as np

from cereal import log
import cereal.messaging as messaging
from selfdrive.config import Conversions as CV
from selfdrive.controls.lib.lead_mpc import LeadMpc


def RW(v_ego, v_l):

            

Reported by Pylint.

              import numpy as np

from cereal import log
import cereal.messaging as messaging
from selfdrive.config import Conversions as CV
from selfdrive.controls.lib.lead_mpc import LeadMpc


def RW(v_ego, v_l):

            

Reported by Pylint.

              
class FakePubMaster():
  def send(self, s, data):
    assert data


def run_following_distance_simulation(v_lead, t_end=200.0):
  dt = 0.2
  t = 0.

            

Reported by Bandit.

              import random
import unittest

import cereal.messaging as messaging
from common.params import Params
from selfdrive.locationd.calibrationd import Calibrator


class TestCalibrationd(unittest.TestCase):

            

Reported by Pylint.

              import random
import unittest

import cereal.messaging as messaging
from common.params import Params
from selfdrive.locationd.calibrationd import Calibrator


class TestCalibrationd(unittest.TestCase):

            

Reported by Pylint.

              
  def test_read_saved_params(self):
    msg = messaging.new_message('liveCalibration')
    msg.liveCalibration.validBlocks = random.randint(1, 10)
    msg.liveCalibration.rpyCalib = [random.random() for _ in range(3)]
    Params().put("CalibrationParams", msg.to_bytes())
    c = Calibrator(param_put=True)

    self.assertEqual(list(msg.liveCalibration.rpyCalib), c.rpy)

            

Reported by Bandit.

                def test_read_saved_params(self):
    msg = messaging.new_message('liveCalibration')
    msg.liveCalibration.validBlocks = random.randint(1, 10)
    msg.liveCalibration.rpyCalib = [random.random() for _ in range(3)]
    Params().put("CalibrationParams", msg.to_bytes())
    c = Calibrator(param_put=True)

    self.assertEqual(list(msg.liveCalibration.rpyCalib), c.rpy)
    self.assertEqual(msg.liveCalibration.validBlocks, c.valid_blocks)

            

Reported by Bandit.

              def get_apport_stacktrace(fn):
  try:
    cmd = f'apport-retrace -s <(cat <(echo "Package: openpilot") "{fn}")'
    return subprocess.check_output(cmd, shell=True, encoding='utf8', timeout=30, executable='/bin/bash')  # pylint: disable=unexpected-keyword-arg
  except subprocess.CalledProcessError:
    return "Error getting stacktrace"
  except subprocess.TimeoutExpired:
    return "Timeout getting stacktrace"


            

Reported by Bandit.

              import time
import glob

import sentry_sdk

from common.params import Params
from common.file_helpers import mkdirs_exists_ok
from selfdrive.hardware import TICI, HARDWARE
from selfdrive.loggerd.config import ROOT

            

Reported by Pylint.

              import re
import shutil
import signal
import subprocess
import time
import glob

import sentry_sdk


            

Reported by Bandit.

                for f in glob.glob(APPORT_DIR + '*'):
    try:
      os.remove(f)
    except Exception:
      pass


def get_apport_stacktrace(fn):
  try:

            

Reported by Bandit.

              from enum import IntEnum
from typing import Dict, Union, Callable, Any

from cereal import log, car
import cereal.messaging as messaging
from common.realtime import DT_CTRL
from selfdrive.config import Conversions as CV
from selfdrive.locationd.calibrationd import MIN_SPEED_FILTER


            

Reported by Pylint.

              from enum import IntEnum
from typing import Dict, Union, Callable, Any

from cereal import log, car
import cereal.messaging as messaging
from common.realtime import DT_CTRL
from selfdrive.config import Conversions as CV
from selfdrive.locationd.calibrationd import MIN_SPEED_FILTER


            

Reported by Pylint.

              from typing import Dict, Union, Callable, Any

from cereal import log, car
import cereal.messaging as messaging
from common.realtime import DT_CTRL
from selfdrive.config import Conversions as CV
from selfdrive.locationd.calibrationd import MIN_SPEED_FILTER

AlertSize = log.ControlsState.AlertSize

            

Reported by Pylint.

              from typing import Dict, Union, Callable, Any

from cereal import log, car
import cereal.messaging as messaging
from common.realtime import DT_CTRL
from selfdrive.config import Conversions as CV
from selfdrive.locationd.calibrationd import MIN_SPEED_FILTER

AlertSize = log.ControlsState.AlertSize

            

Reported by Pylint.

              import math
from collections import defaultdict, deque

import cereal.messaging as messaging
from cereal import car
from common.numpy_fast import interp
from common.params import Params
from common.realtime import Ratekeeper, Priority, config_realtime_process
from selfdrive.config import RADAR_TO_CAMERA

            

Reported by Pylint.

              import math
from collections import defaultdict, deque

import cereal.messaging as messaging
from cereal import car
from common.numpy_fast import interp
from common.params import Params
from common.realtime import Ratekeeper, Priority, config_realtime_process
from selfdrive.config import RADAR_TO_CAMERA

            

Reported by Pylint.

              from collections import defaultdict, deque

import cereal.messaging as messaging
from cereal import car
from common.numpy_fast import interp
from common.params import Params
from common.realtime import Ratekeeper, Priority, config_realtime_process
from selfdrive.config import RADAR_TO_CAMERA
from selfdrive.controls.lib.cluster.fastcluster_py import cluster_points_centroid

            

Reported by Pylint.

                def __init__(self, dt):
    # Lead Kalman Filter params, calculating K from A, C, Q, R requires the control library.
    # hardcoding a lookup table to compute K for values of radar_ts between 0.1s and 1.0s
    assert dt > .01 and dt < .1, "Radar time step must be between .01s and 0.1s"
    self.A = [[1.0, dt], [0.0, 1.0]]
    self.C = [1.0, 0.0]
    #Q = np.matrix([[10., 0.0], [0.0, 100.]])
    #R = 1e3
    #K = np.matrix([[ 0.05705578], [ 0.03073241]])

            

Reported by Bandit.

                  needs_registration = True
    cloudlog.warning("generating your personal RSA key")
    mkdirs_exists_ok(PERSIST+"/comma")
    assert os.system("openssl genrsa -out "+PERSIST+"/comma/id_rsa.tmp 2048") == 0
    assert os.system("openssl rsa -in "+PERSIST+"/comma/id_rsa.tmp -pubout -out "+PERSIST+"/comma/id_rsa.tmp.pub") == 0
    os.rename(PERSIST+"/comma/id_rsa.tmp", PERSIST+"/comma/id_rsa")
    os.rename(PERSIST+"/comma/id_rsa.tmp.pub", PERSIST+"/comma/id_rsa.pub")

  if needs_registration:

            

Reported by Bandit.

                  cloudlog.warning("generating your personal RSA key")
    mkdirs_exists_ok(PERSIST+"/comma")
    assert os.system("openssl genrsa -out "+PERSIST+"/comma/id_rsa.tmp 2048") == 0
    assert os.system("openssl rsa -in "+PERSIST+"/comma/id_rsa.tmp -pubout -out "+PERSIST+"/comma/id_rsa.tmp.pub") == 0
    os.rename(PERSIST+"/comma/id_rsa.tmp", PERSIST+"/comma/id_rsa")
    os.rename(PERSIST+"/comma/id_rsa.tmp.pub", PERSIST+"/comma/id_rsa.pub")

  if needs_registration:
    if show_spinner:

            

Reported by Bandit.

                  needs_registration = True
    cloudlog.warning("generating your personal RSA key")
    mkdirs_exists_ok(PERSIST+"/comma")
    assert os.system("openssl genrsa -out "+PERSIST+"/comma/id_rsa.tmp 2048") == 0
    assert os.system("openssl rsa -in "+PERSIST+"/comma/id_rsa.tmp -pubout -out "+PERSIST+"/comma/id_rsa.tmp.pub") == 0
    os.rename(PERSIST+"/comma/id_rsa.tmp", PERSIST+"/comma/id_rsa")
    os.rename(PERSIST+"/comma/id_rsa.tmp.pub", PERSIST+"/comma/id_rsa.pub")

  if needs_registration:

            

Reported by Bandit.

                  cloudlog.warning("generating your personal RSA key")
    mkdirs_exists_ok(PERSIST+"/comma")
    assert os.system("openssl genrsa -out "+PERSIST+"/comma/id_rsa.tmp 2048") == 0
    assert os.system("openssl rsa -in "+PERSIST+"/comma/id_rsa.tmp -pubout -out "+PERSIST+"/comma/id_rsa.tmp.pub") == 0
    os.rename(PERSIST+"/comma/id_rsa.tmp", PERSIST+"/comma/id_rsa")
    os.rename(PERSIST+"/comma/id_rsa.tmp.pub", PERSIST+"/comma/id_rsa.pub")

  if needs_registration:
    if show_spinner:

            

Reported by Bandit.