import os
import subprocess
from cffi import FFI
from common.basedir import BASEDIR

# Initialize visiontest. Ignore output.
_visiond_dir = os.path.dirname(os.path.abspath(__file__))
_libvisiontest = "libvisiontest.so"
try:  # because this crashes sometimes when running pipeline

            

Reported by Bandit.

              _visiond_dir = os.path.dirname(os.path.abspath(__file__))
_libvisiontest = "libvisiontest.so"
try:  # because this crashes sometimes when running pipeline
  subprocess.check_output(["make", "-C", _visiond_dir, "-f",
                           os.path.join(_visiond_dir, "visiontest.mk"),
                           _libvisiontest])
except Exception:
  pass


            

Reported by Bandit.

              _visiond_dir = os.path.dirname(os.path.abspath(__file__))
_libvisiontest = "libvisiontest.so"
try:  # because this crashes sometimes when running pipeline
  subprocess.check_output(["make", "-C", _visiond_dir, "-f",
                           os.path.join(_visiond_dir, "visiontest.mk"),
                           _libvisiontest])
except Exception:
  pass


            

Reported by Bandit.

                subprocess.check_output(["make", "-C", _visiond_dir, "-f",
                           os.path.join(_visiond_dir, "visiontest.mk"),
                           _libvisiontest])
except Exception:
  pass


class VisionTest():
  """A version of the vision model that can be run on a desktop.

            

Reported by Bandit.

              
  def transform_output_buffer(self, yuv_data, y_out, u_out, v_out,
                              transform):
    assert len(yuv_data) == self.input_size[0] * self.input_size[1] * 3 / 2

    cast = self.ffi.cast
    from_buffer = self.ffi.from_buffer
    yuv_ptr = cast("unsigned char*", from_buffer(yuv_data))
    transform_ptr = self.ffi.new("float[]", transform)

            

Reported by Bandit.

              from cereal import car
from selfdrive.config import Conversions as CV
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.mazda.values import DBC, LKAS_LIMITS, GEN1

class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              from cereal import car
from selfdrive.config import Conversions as CV
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.mazda.values import DBC, LKAS_LIMITS, GEN1

class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              from cereal import car
from selfdrive.config import Conversions as CV
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.mazda.values import DBC, LKAS_LIMITS, GEN1

class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              from cereal import car
from selfdrive.config import Conversions as CV
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.mazda.values import DBC, LKAS_LIMITS, GEN1

class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              from cereal import car
from selfdrive.config import Conversions as CV
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.mazda.values import DBC, LKAS_LIMITS, GEN1

class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              #!/usr/bin/env python3
import unittest
import importlib
from parameterized import parameterized

from cereal import car
from selfdrive.car.fingerprints import all_known_cars
from selfdrive.car.car_helpers import interfaces
from selfdrive.car.fingerprints import _FINGERPRINTS as FINGERPRINTS

            

Reported by Pylint.

              import importlib
from parameterized import parameterized

from cereal import car
from selfdrive.car.fingerprints import all_known_cars
from selfdrive.car.car_helpers import interfaces
from selfdrive.car.fingerprints import _FINGERPRINTS as FINGERPRINTS

class TestCarInterfaces(unittest.TestCase):

            

Reported by Pylint.

              
    car_params = CarInterface.get_params(car_name, fingerprints, car_fw)
    car_interface = CarInterface(car_params, CarController, CarState)
    assert car_params
    assert car_interface

    self.assertGreater(car_params.mass, 1)
    self.assertGreater(car_params.steerRateCost, 1e-3)


            

Reported by Bandit.

                  car_params = CarInterface.get_params(car_name, fingerprints, car_fw)
    car_interface = CarInterface(car_params, CarController, CarState)
    assert car_params
    assert car_interface

    self.assertGreater(car_params.mass, 1)
    self.assertGreater(car_params.steerRateCost, 1e-3)

    if car_params.steerControlType != car.CarParams.SteerControlType.angle:

            

Reported by Bandit.

                  # Test radar interface
    RadarInterface = importlib.import_module('selfdrive.car.%s.radar_interface' % car_params.carName).RadarInterface
    radar_interface = RadarInterface(car_params)
    assert radar_interface

    # Run radar interface once
    radar_interface.update([])
    if not car_params.radarOffCan and hasattr(radar_interface, '_update') and hasattr(radar_interface, 'trigger_msg'):
      radar_interface._update([radar_interface.trigger_msg])

            

Reported by Bandit.

              #!/usr/bin/env python3
import sys
from common.xattr import removexattr
from selfdrive.loggerd.uploader import UPLOAD_ATTR_NAME

for fn in sys.argv[1:]:
  print("unmarking %s" % fn)
  removexattr(fn, UPLOAD_ATTR_NAME)

            

Reported by Pylint.

              #!/usr/bin/env python3
import sys
from common.xattr import removexattr
from selfdrive.loggerd.uploader import UPLOAD_ATTR_NAME

for fn in sys.argv[1:]:
  print("unmarking %s" % fn)
  removexattr(fn, UPLOAD_ATTR_NAME)

            

Reported by Pylint.

              #!/usr/bin/env python3
import sys
from common.xattr import removexattr
from selfdrive.loggerd.uploader import UPLOAD_ATTR_NAME

for fn in sys.argv[1:]:
  print("unmarking %s" % fn)
  removexattr(fn, UPLOAD_ATTR_NAME)

            

Reported by Pylint.

              from selfdrive.loggerd.uploader import UPLOAD_ATTR_NAME

for fn in sys.argv[1:]:
  print("unmarking %s" % fn)
  removexattr(fn, UPLOAD_ATTR_NAME)

            

Reported by Pylint.

              
for fn in sys.argv[1:]:
  print("unmarking %s" % fn)
  removexattr(fn, UPLOAD_ATTR_NAME)

            

Reported by Pylint.

              import copy
from cereal import car
from selfdrive.car.hyundai.values import DBC, STEER_THRESHOLD, FEATURES, EV_CAR, HYBRID_CAR
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV



            

Reported by Pylint.

              from cereal import car
from selfdrive.car.hyundai.values import DBC, STEER_THRESHOLD, FEATURES, EV_CAR, HYBRID_CAR
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV


class CarState(CarStateBase):

            

Reported by Pylint.

              from cereal import car
from selfdrive.car.hyundai.values import DBC, STEER_THRESHOLD, FEATURES, EV_CAR, HYBRID_CAR
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV


class CarState(CarStateBase):

            

Reported by Pylint.

              from selfdrive.car.hyundai.values import DBC, STEER_THRESHOLD, FEATURES, EV_CAR, HYBRID_CAR
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV


class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              from selfdrive.car.hyundai.values import DBC, STEER_THRESHOLD, FEATURES, EV_CAR, HYBRID_CAR
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV


class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              import traceback
from pathlib import Path

from cereal import log
import cereal.messaging as messaging
from common.api import Api
from common.params import Params
from selfdrive.hardware import TICI
from selfdrive.loggerd.xattr_cache import getxattr, setxattr

            

Reported by Pylint.

              from pathlib import Path

from cereal import log
import cereal.messaging as messaging
from common.api import Api
from common.params import Params
from selfdrive.hardware import TICI
from selfdrive.loggerd.xattr_cache import getxattr, setxattr
from selfdrive.loggerd.config import ROOT

            

Reported by Pylint.

              from pathlib import Path

from cereal import log
import cereal.messaging as messaging
from common.api import Api
from common.params import Params
from selfdrive.hardware import TICI
from selfdrive.loggerd.xattr_cache import getxattr, setxattr
from selfdrive.loggerd.config import ROOT

            

Reported by Pylint.

              
    try:
      self.do_upload(key, fn)
    except Exception:
      pass

    return self.last_resp

  def upload(self, key, fn):

            

Reported by Bandit.

                    backoff = 0.1
    elif allow_sleep:
      cloudlog.info("upload backoff %r", backoff)
      time.sleep(backoff + random.uniform(0, backoff))
      backoff = min(backoff*2, 120)

    pm.send("uploaderState", uploader.get_msg())
    cloudlog.info("upload done, success=%r", success)


            

Reported by Bandit.

              from cereal import car
from common.numpy_fast import mean
from opendbc.can.can_define import CANDefine
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.toyota.values import CAR, DBC, STEER_THRESHOLD, NO_STOP_TIMER_CAR, TSS2_CAR



            

Reported by Pylint.

              from cereal import car
from common.numpy_fast import mean
from opendbc.can.can_define import CANDefine
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.toyota.values import CAR, DBC, STEER_THRESHOLD, NO_STOP_TIMER_CAR, TSS2_CAR



            

Reported by Pylint.

              from cereal import car
from common.numpy_fast import mean
from opendbc.can.can_define import CANDefine
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.toyota.values import CAR, DBC, STEER_THRESHOLD, NO_STOP_TIMER_CAR, TSS2_CAR



            

Reported by Pylint.

              from common.numpy_fast import mean
from opendbc.can.can_define import CANDefine
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.toyota.values import CAR, DBC, STEER_THRESHOLD, NO_STOP_TIMER_CAR, TSS2_CAR


class CarState(CarStateBase):

            

Reported by Pylint.

              from common.numpy_fast import mean
from opendbc.can.can_define import CANDefine
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.toyota.values import CAR, DBC, STEER_THRESHOLD, NO_STOP_TIMER_CAR, TSS2_CAR


class CarState(CarStateBase):

            

Reported by Pylint.

              from cereal import car
from collections import defaultdict
from common.numpy_fast import interp
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.honda.values import CAR, DBC, STEER_THRESHOLD, SPEED_FACTOR, HONDA_BOSCH, HONDA_BOSCH_ALT_BRAKE_SIGNAL


            

Reported by Pylint.

              from cereal import car
from collections import defaultdict
from common.numpy_fast import interp
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.honda.values import CAR, DBC, STEER_THRESHOLD, SPEED_FACTOR, HONDA_BOSCH, HONDA_BOSCH_ALT_BRAKE_SIGNAL


            

Reported by Pylint.

              from cereal import car
from collections import defaultdict
from common.numpy_fast import interp
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.honda.values import CAR, DBC, STEER_THRESHOLD, SPEED_FACTOR, HONDA_BOSCH, HONDA_BOSCH_ALT_BRAKE_SIGNAL


            

Reported by Pylint.

              from collections import defaultdict
from common.numpy_fast import interp
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.honda.values import CAR, DBC, STEER_THRESHOLD, SPEED_FACTOR, HONDA_BOSCH, HONDA_BOSCH_ALT_BRAKE_SIGNAL

TransmissionType = car.CarParams.TransmissionType

            

Reported by Pylint.

              from collections import defaultdict
from common.numpy_fast import interp
from opendbc.can.can_define import CANDefine
from opendbc.can.parser import CANParser
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from selfdrive.car.honda.values import CAR, DBC, STEER_THRESHOLD, SPEED_FACTOR, HONDA_BOSCH, HONDA_BOSCH_ALT_BRAKE_SIGNAL

TransmissionType = car.CarParams.TransmissionType

            

Reported by Pylint.

              
        expected_frames = fps * SEGMENT_LENGTH
        frame_tolerance = 1 if eon_dcam else 0
        probe = subprocess.check_output(cmd, shell=True, encoding='utf8')
        frame_count = int(probe.split('\n')[0].strip())
        counts.append(frame_count)

        self.assertTrue(abs(expected_frames - frame_count) <= frame_tolerance,
                        f"{camera} failed frame count check: expected {expected_frames}, got {frame_count}")

            

Reported by Bandit.

              import unittest
from pathlib import Path

from parameterized import parameterized
from tqdm import trange

from common.params import Params
from common.timeout import Timeout
from selfdrive.hardware import EON, TICI

            

Reported by Pylint.

              from pathlib import Path

from parameterized import parameterized
from tqdm import trange

from common.params import Params
from common.timeout import Timeout
from selfdrive.hardware import EON, TICI
from selfdrive.loggerd.config import ROOT

            

Reported by Pylint.

              import os
import random
import shutil
import subprocess
import time
import unittest
from pathlib import Path

from parameterized import parameterized

            

Reported by Bandit.

                def test_log_rotation(self, record_front):
    Params().put("RecordFront", str(int(record_front)))

    num_segments = int(os.getenv("SEGMENTS", random.randint(10, 15)))

    # wait for loggerd to make the dir for first segment
    route_prefix_path = None
    with Timeout(int(SEGMENT_LENGTH*3)):
      while route_prefix_path is None:

            

Reported by Bandit.

              import copy
from cereal import car
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.car.subaru.values import DBC, STEER_THRESHOLD, CAR, PREGLOBAL_CARS



            

Reported by Pylint.

              import copy
from cereal import car
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.car.subaru.values import DBC, STEER_THRESHOLD, CAR, PREGLOBAL_CARS



            

Reported by Pylint.

              import copy
from cereal import car
from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.car.subaru.values import DBC, STEER_THRESHOLD, CAR, PREGLOBAL_CARS



            

Reported by Pylint.

              from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.car.subaru.values import DBC, STEER_THRESHOLD, CAR, PREGLOBAL_CARS


class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.

              from opendbc.can.can_define import CANDefine
from selfdrive.config import Conversions as CV
from selfdrive.car.interfaces import CarStateBase
from opendbc.can.parser import CANParser
from selfdrive.car.subaru.values import DBC, STEER_THRESHOLD, CAR, PREGLOBAL_CARS


class CarState(CarStateBase):
  def __init__(self, CP):

            

Reported by Pylint.