from math import atan2, sqrt

from cereal import car
from common.numpy_fast import interp
from common.realtime import DT_DMON
from selfdrive.hardware import TICI
from common.filter_simple import FirstOrderFilter
from common.stat_live import RunningStatFilter


            

Reported by Pylint.

              from math import atan2, sqrt

from cereal import car
from common.numpy_fast import interp
from common.realtime import DT_DMON
from selfdrive.hardware import TICI
from common.filter_simple import FirstOrderFilter
from common.stat_live import RunningStatFilter


            

Reported by Pylint.

              
from cereal import car
from common.numpy_fast import interp
from common.realtime import DT_DMON
from selfdrive.hardware import TICI
from common.filter_simple import FirstOrderFilter
from common.stat_live import RunningStatFilter

EventName = car.CarEvent.EventName

            

Reported by Pylint.

              from cereal import car
from common.numpy_fast import interp
from common.realtime import DT_DMON
from selfdrive.hardware import TICI
from common.filter_simple import FirstOrderFilter
from common.stat_live import RunningStatFilter

EventName = car.CarEvent.EventName


            

Reported by Pylint.

              from common.numpy_fast import interp
from common.realtime import DT_DMON
from selfdrive.hardware import TICI
from common.filter_simple import FirstOrderFilter
from common.stat_live import RunningStatFilter

EventName = car.CarEvent.EventName

# ******************************************************************************************

            

Reported by Pylint.

              from common.realtime import DT_DMON
from selfdrive.hardware import TICI
from common.filter_simple import FirstOrderFilter
from common.stat_live import RunningStatFilter

EventName = car.CarEvent.EventName

# ******************************************************************************************
#  NOTE: To fork maintainers.

            

Reported by Pylint.

              # ******************************************************************************************

class DRIVER_MONITOR_SETTINGS():
  def __init__(self, TICI=TICI, DT_DMON=DT_DMON):
    self._DT_DMON = DT_DMON
    self._AWARENESS_TIME = 35. # passive wheeltouch total timeout
    self._AWARENESS_PRE_TIME_TILL_TERMINAL = 12.
    self._AWARENESS_PROMPT_TIME_TILL_TERMINAL = 6.
    self._DISTRACTED_TIME = 11. # active monitoring total timeout

            

Reported by Pylint.

              # ******************************************************************************************

class DRIVER_MONITOR_SETTINGS():
  def __init__(self, TICI=TICI, DT_DMON=DT_DMON):
    self._DT_DMON = DT_DMON
    self._AWARENESS_TIME = 35. # passive wheeltouch total timeout
    self._AWARENESS_PRE_TIME_TILL_TERMINAL = 12.
    self._AWARENESS_PROMPT_TIME_TILL_TERMINAL = 6.
    self._DISTRACTED_TIME = 11. # active monitoring total timeout

            

Reported by Pylint.

                def _set_timers(self, active_monitoring):
    if self.active_monitoring_mode and self.awareness <= self.threshold_prompt:
      if active_monitoring:
        self.step_change = self.settings._DT_DMON / self.settings._DISTRACTED_TIME
      else:
        self.step_change = 0.
      return  # no exploit after orange alert
    elif self.awareness <= 0.:
      return

            

Reported by Pylint.

                def _set_timers(self, active_monitoring):
    if self.active_monitoring_mode and self.awareness <= self.threshold_prompt:
      if active_monitoring:
        self.step_change = self.settings._DT_DMON / self.settings._DISTRACTED_TIME
      else:
        self.step_change = 0.
      return  # no exploit after orange alert
    elif self.awareness <= 0.:
      return

            

Reported by Pylint.

              import numpy as np
import time
import threading
from cereal import log
from multiprocessing import Process, Queue
from typing import Any

import cereal.messaging as messaging
from common.params import Params

            

Reported by Pylint.

              from multiprocessing import Process, Queue
from typing import Any

import cereal.messaging as messaging
from common.params import Params
from common.realtime import Ratekeeper, DT_DMON
from lib.can import can_function
from selfdrive.car.honda.values import CruiseButtons
from selfdrive.test.helpers import set_params_enabled

            

Reported by Pylint.

              from typing import Any

import cereal.messaging as messaging
from common.params import Params
from common.realtime import Ratekeeper, DT_DMON
from lib.can import can_function
from selfdrive.car.honda.values import CruiseButtons
from selfdrive.test.helpers import set_params_enabled


            

Reported by Pylint.

              
import cereal.messaging as messaging
from common.params import Params
from common.realtime import Ratekeeper, DT_DMON
from lib.can import can_function
from selfdrive.car.honda.values import CruiseButtons
from selfdrive.test.helpers import set_params_enabled

parser = argparse.ArgumentParser(description='Bridge between CARLA and openpilot.')

            

Reported by Pylint.

              from common.params import Params
from common.realtime import Ratekeeper, DT_DMON
from lib.can import can_function
from selfdrive.car.honda.values import CruiseButtons
from selfdrive.test.helpers import set_params_enabled

parser = argparse.ArgumentParser(description='Bridge between CARLA and openpilot.')
parser.add_argument('--joystick', action='store_true')
parser.add_argument('--low_quality', action='store_true')

            

Reported by Pylint.

              from common.realtime import Ratekeeper, DT_DMON
from lib.can import can_function
from selfdrive.car.honda.values import CruiseButtons
from selfdrive.test.helpers import set_params_enabled

parser = argparse.ArgumentParser(description='Bridge between CARLA and openpilot.')
parser.add_argument('--joystick', action='store_true')
parser.add_argument('--low_quality', action='store_true')
parser.add_argument('--town', type=str, default='Town04_Opt')

            

Reported by Pylint.

              
frame_id = 0
def cam_callback(image):
  global frame_id
  img = np.frombuffer(image.raw_data, dtype=np.dtype("uint8"))
  img = np.reshape(img, (H, W, 4))
  img = img[:, :, [0, 1, 2]].copy()

  dat = messaging.new_message('roadCameraState')

            

Reported by Pylint.

                pm.send('sensorEvents', dat)

def panda_state_function(exit_event: threading.Event):
  pm = messaging.PubMaster(['pandaState'])
  while not exit_event.is_set():
    dat = messaging.new_message('pandaState')
    dat.valid = True
    dat.pandaState = {
      'ignitionLine': True,

            

Reported by Pylint.

                pm.send('gpsLocationExternal', dat)

def fake_driver_monitoring(exit_event: threading.Event):
  pm = messaging.PubMaster(['driverState','driverMonitoringState'])
  while not exit_event.is_set():
    # dmonitoringmodeld output
    dat = messaging.new_message('driverState')
    dat.driverState.faceProb = 1.0
    pm.send('driverState', dat)

            

Reported by Pylint.

                  i+=1


def bridge(q):

  # setup CARLA
  client = carla.Client("127.0.0.1", 2000)
  client.set_timeout(10.0)
  world = client.load_world(args.town)

            

Reported by Pylint.

              import unittest
import numpy as np

from cereal import car, log
from common.realtime import DT_DMON
from selfdrive.controls.lib.events import Events
from selfdrive.monitoring.driver_monitor import DriverStatus, DRIVER_MONITOR_SETTINGS

EventName = car.CarEvent.EventName

            

Reported by Pylint.

              import numpy as np

from cereal import car, log
from common.realtime import DT_DMON
from selfdrive.controls.lib.events import Events
from selfdrive.monitoring.driver_monitor import DriverStatus, DRIVER_MONITOR_SETTINGS

EventName = car.CarEvent.EventName
dm_settings = DRIVER_MONITOR_SETTINGS()

            

Reported by Pylint.

              
from cereal import car, log
from common.realtime import DT_DMON
from selfdrive.controls.lib.events import Events
from selfdrive.monitoring.driver_monitor import DriverStatus, DRIVER_MONITOR_SETTINGS

EventName = car.CarEvent.EventName
dm_settings = DRIVER_MONITOR_SETTINGS()


            

Reported by Pylint.

              from cereal import car, log
from common.realtime import DT_DMON
from selfdrive.controls.lib.events import Events
from selfdrive.monitoring.driver_monitor import DriverStatus, DRIVER_MONITOR_SETTINGS

EventName = car.CarEvent.EventName
dm_settings = DRIVER_MONITOR_SETTINGS()

TEST_TIMESPAN = 120  # seconds

            

Reported by Pylint.

              dm_settings = DRIVER_MONITOR_SETTINGS()

TEST_TIMESPAN = 120  # seconds
DISTRACTED_SECONDS_TO_ORANGE = dm_settings._DISTRACTED_TIME - dm_settings._DISTRACTED_PROMPT_TIME_TILL_TERMINAL + 1
DISTRACTED_SECONDS_TO_RED = dm_settings._DISTRACTED_TIME + 1
INVISIBLE_SECONDS_TO_ORANGE = dm_settings._AWARENESS_TIME - dm_settings._AWARENESS_PROMPT_TIME_TILL_TERMINAL + 1
INVISIBLE_SECONDS_TO_RED = dm_settings._AWARENESS_TIME + 1

def make_msg(face_detected, distracted=False, model_uncertain=False):

            

Reported by Pylint.

              dm_settings = DRIVER_MONITOR_SETTINGS()

TEST_TIMESPAN = 120  # seconds
DISTRACTED_SECONDS_TO_ORANGE = dm_settings._DISTRACTED_TIME - dm_settings._DISTRACTED_PROMPT_TIME_TILL_TERMINAL + 1
DISTRACTED_SECONDS_TO_RED = dm_settings._DISTRACTED_TIME + 1
INVISIBLE_SECONDS_TO_ORANGE = dm_settings._AWARENESS_TIME - dm_settings._AWARENESS_PROMPT_TIME_TILL_TERMINAL + 1
INVISIBLE_SECONDS_TO_RED = dm_settings._AWARENESS_TIME + 1

def make_msg(face_detected, distracted=False, model_uncertain=False):

            

Reported by Pylint.

              
TEST_TIMESPAN = 120  # seconds
DISTRACTED_SECONDS_TO_ORANGE = dm_settings._DISTRACTED_TIME - dm_settings._DISTRACTED_PROMPT_TIME_TILL_TERMINAL + 1
DISTRACTED_SECONDS_TO_RED = dm_settings._DISTRACTED_TIME + 1
INVISIBLE_SECONDS_TO_ORANGE = dm_settings._AWARENESS_TIME - dm_settings._AWARENESS_PROMPT_TIME_TILL_TERMINAL + 1
INVISIBLE_SECONDS_TO_RED = dm_settings._AWARENESS_TIME + 1

def make_msg(face_detected, distracted=False, model_uncertain=False):
  ds = log.DriverState.new_message()

            

Reported by Pylint.

              TEST_TIMESPAN = 120  # seconds
DISTRACTED_SECONDS_TO_ORANGE = dm_settings._DISTRACTED_TIME - dm_settings._DISTRACTED_PROMPT_TIME_TILL_TERMINAL + 1
DISTRACTED_SECONDS_TO_RED = dm_settings._DISTRACTED_TIME + 1
INVISIBLE_SECONDS_TO_ORANGE = dm_settings._AWARENESS_TIME - dm_settings._AWARENESS_PROMPT_TIME_TILL_TERMINAL + 1
INVISIBLE_SECONDS_TO_RED = dm_settings._AWARENESS_TIME + 1

def make_msg(face_detected, distracted=False, model_uncertain=False):
  ds = log.DriverState.new_message()
  ds.faceOrientation = [0., 0., 0.]

            

Reported by Pylint.

              TEST_TIMESPAN = 120  # seconds
DISTRACTED_SECONDS_TO_ORANGE = dm_settings._DISTRACTED_TIME - dm_settings._DISTRACTED_PROMPT_TIME_TILL_TERMINAL + 1
DISTRACTED_SECONDS_TO_RED = dm_settings._DISTRACTED_TIME + 1
INVISIBLE_SECONDS_TO_ORANGE = dm_settings._AWARENESS_TIME - dm_settings._AWARENESS_PROMPT_TIME_TILL_TERMINAL + 1
INVISIBLE_SECONDS_TO_RED = dm_settings._AWARENESS_TIME + 1

def make_msg(face_detected, distracted=False, model_uncertain=False):
  ds = log.DriverState.new_message()
  ds.faceOrientation = [0., 0., 0.]

            

Reported by Pylint.

              DISTRACTED_SECONDS_TO_ORANGE = dm_settings._DISTRACTED_TIME - dm_settings._DISTRACTED_PROMPT_TIME_TILL_TERMINAL + 1
DISTRACTED_SECONDS_TO_RED = dm_settings._DISTRACTED_TIME + 1
INVISIBLE_SECONDS_TO_ORANGE = dm_settings._AWARENESS_TIME - dm_settings._AWARENESS_PROMPT_TIME_TILL_TERMINAL + 1
INVISIBLE_SECONDS_TO_RED = dm_settings._AWARENESS_TIME + 1

def make_msg(face_detected, distracted=False, model_uncertain=False):
  ds = log.DriverState.new_message()
  ds.faceOrientation = [0., 0., 0.]
  ds.facePosition = [0., 0.]

            

Reported by Pylint.

                  extra_args.append(f'-l {layout}')

  extra_args = " ".join(extra_args)
  subprocess.call(f'{pj} --plugin_folders {os.path.join(juggle_dir, "bin")} {extra_args}', shell=True, env=env, cwd=juggle_dir)

def juggle_route(route_name, segment_number, segment_count, qlog, can, layout):
  if 'cabana' in route_name:
    query = parse_qs(urlparse(route_name).query)
    api = CommaApi(get_token())

            

Reported by Bandit.

              import argparse
from tempfile import NamedTemporaryFile

from common.basedir import BASEDIR
from selfdrive.test.process_replay.compare_logs import save_log
from tools.lib.api import CommaApi
from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route

            

Reported by Pylint.

              from tempfile import NamedTemporaryFile

from common.basedir import BASEDIR
from selfdrive.test.process_replay.compare_logs import save_log
from tools.lib.api import CommaApi
from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route
from urllib.parse import urlparse, parse_qs

            

Reported by Pylint.

              
from common.basedir import BASEDIR
from selfdrive.test.process_replay.compare_logs import save_log
from tools.lib.api import CommaApi
from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route
from urllib.parse import urlparse, parse_qs


            

Reported by Pylint.

              from common.basedir import BASEDIR
from selfdrive.test.process_replay.compare_logs import save_log
from tools.lib.api import CommaApi
from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route
from urllib.parse import urlparse, parse_qs

juggle_dir = os.path.dirname(os.path.realpath(__file__))

            

Reported by Pylint.

              from selfdrive.test.process_replay.compare_logs import save_log
from tools.lib.api import CommaApi
from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route
from urllib.parse import urlparse, parse_qs

juggle_dir = os.path.dirname(os.path.realpath(__file__))


            

Reported by Pylint.

              from tools.lib.api import CommaApi
from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route
from urllib.parse import urlparse, parse_qs

juggle_dir = os.path.dirname(os.path.realpath(__file__))

def load_segment(segment_name):

            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import sys
import multiprocessing
import subprocess
import argparse
from tempfile import NamedTemporaryFile

from common.basedir import BASEDIR

            

Reported by Pylint.

              import os
import sys
import multiprocessing
import subprocess
import argparse
from tempfile import NamedTemporaryFile

from common.basedir import BASEDIR
from selfdrive.test.process_replay.compare_logs import save_log

            

Reported by Bandit.

              from tools.lib.auth_config import get_token
from tools.lib.logreader import LogReader
from tools.lib.route import Route
from urllib.parse import urlparse, parse_qs

juggle_dir = os.path.dirname(os.path.realpath(__file__))

def load_segment(segment_name):
  print(f"Loading {segment_name}")

            

Reported by Pylint.

              
if __name__ == '__main__':
  from multiprocessing import Process, Queue
  q: Queue[str] = Queue()
  p = Process(target=wheel_poll_thread, args=(q,))
  p.start()

            

Reported by Pylint.

              axis_map = []
button_map = []

def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')


            

Reported by Pylint.

              
def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')

  # Get the device name.
  #buf = bytearray(63)

            

Reported by Pylint.

              #!/usr/bin/env python3
# set up wheel
import array
import os
import struct
from fcntl import ioctl
from typing import NoReturn

# Iterate over the joystick devices.

            

Reported by Pylint.

              axis_map = []
button_map = []

def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')


            

Reported by Pylint.

              axis_map = []
button_map = []

def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')


            

Reported by Pylint.

              axis_map = []
button_map = []

def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')


            

Reported by Pylint.

              axis_map = []
button_map = []

def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')


            

Reported by Pylint.

              axis_map = []
button_map = []

def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')


            

Reported by Pylint.

              
def wheel_poll_thread(q: 'Queue[str]') -> NoReturn:
  # Open the joystick device.
  fn = '/dev/input/js0'
  print('Opening %s...' % fn)
  jsdev = open(fn, 'rb')

  # Get the device name.
  #buf = bytearray(63)

            

Reported by Pylint.

              #!/usr/bin/env python
import argparse

import cereal.messaging as messaging
from common.numpy_fast import interp, clip
from common.params import Params
from inputs import get_gamepad
from tools.lib.kbhit import KBHit


            

Reported by Pylint.

              import argparse

import cereal.messaging as messaging
from common.numpy_fast import interp, clip
from common.params import Params
from inputs import get_gamepad
from tools.lib.kbhit import KBHit



            

Reported by Pylint.

              
import cereal.messaging as messaging
from common.numpy_fast import interp, clip
from common.params import Params
from inputs import get_gamepad
from tools.lib.kbhit import KBHit


class Keyboard:

            

Reported by Pylint.

              import cereal.messaging as messaging
from common.numpy_fast import interp, clip
from common.params import Params
from inputs import get_gamepad
from tools.lib.kbhit import KBHit


class Keyboard:
  def __init__(self):

            

Reported by Pylint.

              from common.numpy_fast import interp, clip
from common.params import Params
from inputs import get_gamepad
from tools.lib.kbhit import KBHit


class Keyboard:
  def __init__(self):
    self.kb = KBHit()

            

Reported by Pylint.

              
  def update(self):
    key = self.kb.getch().lower()
    self.cancel = False
    if key == 'r':
      self.axes_values = {ax: 0. for ax in self.axes_values}
    elif key == 'c':
      self.cancel = True
    elif key in self.axes_map:

            

Reported by Pylint.

                  if key == 'r':
      self.axes_values = {ax: 0. for ax in self.axes_values}
    elif key == 'c':
      self.cancel = True
    elif key in self.axes_map:
      axis = self.axes_map[key]
      incr = self.axis_increment if key in ['w', 'a'] else -self.axis_increment
      self.axes_values[axis] = clip(self.axes_values[axis] + incr, -1, 1)
    else:

            

Reported by Pylint.

                def update(self):
    joystick_event = get_gamepad()[0]
    event = (joystick_event.code, joystick_event.state)
    self.cancel = False
    if event[0] == self.cancel_button and event[1] == 0:  # state 0 is falling edge
      self.cancel = True
    elif event[0] in self.axes_values:
      norm = -interp(event[1], [0, self.max_axis_value], [-1., 1.])
      self.axes_values[event[0]] = norm if abs(norm) > 0.05 else 0.  # center can be noisy, deadzone of 5%

            

Reported by Pylint.

                  event = (joystick_event.code, joystick_event.state)
    self.cancel = False
    if event[0] == self.cancel_button and event[1] == 0:  # state 0 is falling edge
      self.cancel = True
    elif event[0] in self.axes_values:
      norm = -interp(event[1], [0, self.max_axis_value], [-1., 1.])
      self.axes_values[event[0]] = norm if abs(norm) > 0.05 else 0.  # center can be noisy, deadzone of 5%
    else:
      return False

            

Reported by Pylint.

              #!/usr/bin/env python
import argparse

import cereal.messaging as messaging
from common.numpy_fast import interp, clip
from common.params import Params
from inputs import get_gamepad
from tools.lib.kbhit import KBHit


            

Reported by Pylint.

                      in_file=in_file,
        out_file=out_file
    )
    status = os.system(os_cmd)
    if (status != 0):
        raise Exception('Rendering of {} failed! Exiting.\n'.format(in_file))

    os.chdir(cwd)


            

Reported by Bandit.

              import urllib.request
import shutil
import numpy as np
from casadi import SX, MX, DM, Function, CasadiMeta

ALLOWED_CASADI_VERSIONS = ('3.5.5', '3.5.4', '3.5.3', '3.5.2', '3.5.1', '3.4.5', '3.4.0')

TERA_VERSION = "0.0.34"


            

Reported by Pylint.

              
def make_model_consistent(model):
    x = model.x
    xdot = model.xdot
    u = model.u
    z = model.z
    p = model.p

    if isinstance(x, MX):

            

Reported by Pylint.

              def make_model_consistent(model):
    x = model.x
    xdot = model.xdot
    u = model.u
    z = model.z
    p = model.p

    if isinstance(x, MX):
        symbol = MX.sym

            

Reported by Pylint.

                      return tera_path

    repo_url = "https://github.com/acados/tera_renderer/releases"
    url = "{}/download/v{}/t_renderer-v{}-{}".format(
        repo_url, TERA_VERSION, TERA_VERSION, platform2tera[sys.platform])

    manual_install = 'For manual installation follow these instructions:\n'
    manual_install += '1 Download binaries from {}\n'.format(url)
    manual_install += '2 Copy them in {}/bin\n'.format(acados_path)

            

Reported by Pylint.

              
    if input(msg) == 'y':
        print("Dowloading {}".format(url))
        with urllib.request.urlopen(url) as response, open(tera_path, 'wb') as out_file:
            shutil.copyfileobj(response, out_file)
        print("Successfully downloaded t_renderer.")
        os.chmod(tera_path, 0o755)
        return tera_path


            

Reported by Bandit.

                      with urllib.request.urlopen(url) as response, open(tera_path, 'wb') as out_file:
            shutil.copyfileobj(response, out_file)
        print("Successfully downloaded t_renderer.")
        os.chmod(tera_path, 0o755)
        return tera_path

    msg_cancel = "\nYou cancelled automatic download.\n\n"
    msg_cancel += manual_install
    msg_cancel += "Once installed re-run your script.\n\n"

            

Reported by Bandit.

                  acados_path = os.path.dirname(os.path.abspath(__file__))

    template_glob = acados_path + '/c_templates_tera/*'
    acados_template_path = acados_path + '/c_templates_tera'

    # call tera as system cmd
    os_cmd = "{tera_path} '{template_glob}' '{in_file}' '{json_path}' '{out_file}'".format(
        tera_path=tera_path,
        template_glob=template_glob,

            

Reported by Pylint.

                      try:
            layout_of_key = layout[key]
        except KeyError:
            raise Exception("ocp_check_against_layout_recursion: field" \
                            " '{0}' is not in layout but in OCP description.".format(key))

        if isinstance(item, dict):
            ocp_check_against_layout_recursion(item, ocp_dims, layout_of_key)


            

Reported by Pylint.

              #
# Copyright 2019 Gianluca Frison, Dimitris Kouzoupis, Robin Verschueren,
# Andrea Zanelli, Niels van Duijkeren, Jonathan Frey, Tommaso Sartor,
# Branimir Novoselnik, Rien Quirynen, Rezart Qelibari, Dang Doan,
# Jonas Koenemann, Yutao Chen, Tobias Schöls, Jonas Schlagenhauf, Moritz Diehl
#
# This file is part of acados.
#
# The 2-Clause BSD License

            

Reported by Pylint.

              #!/usr/bin/env python3
import cereal.messaging as messaging
from opendbc.can.packer import CANPacker
from opendbc.can.parser import CANParser
from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp  # pylint: disable=no-name-in-module,import-error
from selfdrive.car.honda.values import FINGERPRINTS, CAR
from selfdrive.car import crc8_pedal

packer = CANPacker("honda_civic_touring_2016_can_generated")

            

Reported by Pylint.

              #!/usr/bin/env python3
import cereal.messaging as messaging
from opendbc.can.packer import CANPacker
from opendbc.can.parser import CANParser
from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp  # pylint: disable=no-name-in-module,import-error
from selfdrive.car.honda.values import FINGERPRINTS, CAR
from selfdrive.car import crc8_pedal

packer = CANPacker("honda_civic_touring_2016_can_generated")

            

Reported by Pylint.

              #!/usr/bin/env python3
import cereal.messaging as messaging
from opendbc.can.packer import CANPacker
from opendbc.can.parser import CANParser
from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp  # pylint: disable=no-name-in-module,import-error
from selfdrive.car.honda.values import FINGERPRINTS, CAR
from selfdrive.car import crc8_pedal

packer = CANPacker("honda_civic_touring_2016_can_generated")

            

Reported by Pylint.

              from opendbc.can.packer import CANPacker
from opendbc.can.parser import CANParser
from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp  # pylint: disable=no-name-in-module,import-error
from selfdrive.car.honda.values import FINGERPRINTS, CAR
from selfdrive.car import crc8_pedal

packer = CANPacker("honda_civic_touring_2016_can_generated")
rpacker = CANPacker("acura_ilx_2016_nidec")


            

Reported by Pylint.

              from opendbc.can.parser import CANParser
from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp  # pylint: disable=no-name-in-module,import-error
from selfdrive.car.honda.values import FINGERPRINTS, CAR
from selfdrive.car import crc8_pedal

packer = CANPacker("honda_civic_touring_2016_can_generated")
rpacker = CANPacker("acura_ilx_2016_nidec")



            

Reported by Pylint.

              #!/usr/bin/env python3
import cereal.messaging as messaging
from opendbc.can.packer import CANPacker
from opendbc.can.parser import CANParser
from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp  # pylint: disable=no-name-in-module,import-error
from selfdrive.car.honda.values import FINGERPRINTS, CAR
from selfdrive.car import crc8_pedal

packer = CANPacker("honda_civic_touring_2016_can_generated")

            

Reported by Pylint.

              rpacker = CANPacker("acura_ilx_2016_nidec")


def get_car_can_parser():
  dbc_f = 'honda_civic_touring_2016_can_generated'
  signals = [
    ("STEER_TORQUE", 0xe4, 0),
    ("STEER_TORQUE_REQUEST", 0xe4, 0),
    ("COMPUTER_BRAKE", 0x1fa, 0),

            

Reported by Pylint.

              

def get_car_can_parser():
  dbc_f = 'honda_civic_touring_2016_can_generated'
  signals = [
    ("STEER_TORQUE", 0xe4, 0),
    ("STEER_TORQUE_REQUEST", 0xe4, 0),
    ("COMPUTER_BRAKE", 0x1fa, 0),
    ("COMPUTER_BRAKE_REQUEST", 0x1fa, 0),

            

Reported by Pylint.

              
def get_car_can_parser():
  dbc_f = 'honda_civic_touring_2016_can_generated'
  signals = [
    ("STEER_TORQUE", 0xe4, 0),
    ("STEER_TORQUE_REQUEST", 0xe4, 0),
    ("COMPUTER_BRAKE", 0x1fa, 0),
    ("COMPUTER_BRAKE_REQUEST", 0x1fa, 0),
    ("GAS_COMMAND", 0x200, 0),

            

Reported by Pylint.

                  ("COMPUTER_BRAKE_REQUEST", 0x1fa, 0),
    ("GAS_COMMAND", 0x200, 0),
  ]
  checks = [
    (0xe4, 100),
    (0x1fa, 50),
    (0x200, 50),
  ]
  return CANParser(dbc_f, signals, checks, 0)

            

Reported by Pylint.

                  cls.assertTrue(f"cd {UPDATER_PATH} && make clean && make", "updater failed to build")

    # restore the checked-in version, since that's what actually runs on devices
    os.system(f"git reset --hard {UPDATER_PATH}")

  def setUp(self):
    self._clear_dir()

  def tearDown(self):

            

Reported by Bandit.

                    shutil.rmtree("/data/neoupdate")

  def _assert_ok(self, cmd, msg=None):
    self.assertTrue(os.system(cmd) == 0, msg)

  def _assert_fails(self, cmd):
    self.assertFalse(os.system(cmd) == 0)

  def test_background_download(self):

            

Reported by Bandit.

                  self.assertTrue(os.system(cmd) == 0, msg)

  def _assert_fails(self, cmd):
    self.assertFalse(os.system(cmd) == 0)

  def test_background_download(self):
    self._assert_ok(f"{UPDATER} bgcache 'file://{UPDATE_MANIFEST}'")

  def test_background_download_bad_manifest(self):

            

Reported by Bandit.

              import time
import unittest

from common.basedir import BASEDIR

UPDATER_PATH = os.path.join(BASEDIR, "installer/updater")
UPDATER = os.path.join(UPDATER_PATH, "updater")
UPDATE_MANIFEST = os.path.join(UPDATER_PATH, "update.json")


            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import shutil
import subprocess
import tempfile
import time
import unittest

from common.basedir import BASEDIR

            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import shutil
import subprocess
import tempfile
import time
import unittest

from common.basedir import BASEDIR

            

Reported by Bandit.

              UPDATE_MANIFEST = os.path.join(UPDATER_PATH, "update.json")


class TestUpdater(unittest.TestCase):

  @classmethod
  def setUpClass(cls):
    # test that the updater builds
    cls.assertTrue(f"cd {UPDATER_PATH} && make clean && make", "updater failed to build")

            

Reported by Pylint.

              
class TestUpdater(unittest.TestCase):

  @classmethod
  def setUpClass(cls):
    # test that the updater builds
    cls.assertTrue(f"cd {UPDATER_PATH} && make clean && make", "updater failed to build")

    # restore the checked-in version, since that's what actually runs on devices

            

Reported by Pylint.

              class TestUpdater(unittest.TestCase):

  @classmethod
  def setUpClass(cls):
    # test that the updater builds
    cls.assertTrue(f"cd {UPDATER_PATH} && make clean && make", "updater failed to build")

    # restore the checked-in version, since that's what actually runs on devices
    os.system(f"git reset --hard {UPDATER_PATH}")

            

Reported by Pylint.

                @classmethod
  def setUpClass(cls):
    # test that the updater builds
    cls.assertTrue(f"cd {UPDATER_PATH} && make clean && make", "updater failed to build")

    # restore the checked-in version, since that's what actually runs on devices
    os.system(f"git reset --hard {UPDATER_PATH}")

  def setUp(self):

            

Reported by Pylint.

              
os.environ["OMP_NUM_THREADS"] = "4"

import onnxruntime as ort

def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:

            

Reported by Pylint.

              #!/usr/bin/env python3

import os
import sys
import numpy as np

os.environ["OMP_NUM_THREADS"] = "4"

import onnxruntime as ort

            

Reported by Pylint.

              
os.environ["OMP_NUM_THREADS"] = "4"

import onnxruntime as ort

def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:

            

Reported by Pylint.

              
import onnxruntime as ort

def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)

            

Reported by Pylint.

              
import onnxruntime as ort

def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)

            

Reported by Pylint.

              import onnxruntime as ort

def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)
    dd.append(st)

            

Reported by Pylint.

              import onnxruntime as ort

def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)
    dd.append(st)

            

Reported by Pylint.

              
def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)
    dd.append(st)
    gt += len(st)

            

Reported by Pylint.

              
def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)
    dd.append(st)
    gt += len(st)

            

Reported by Pylint.

              def read(sz):
  dd = []
  gt = 0
  while gt < sz * 4:
    st = os.read(0, sz * 4 - gt)
    assert(len(st) > 0)
    dd.append(st)
    gt += len(st)
  return np.frombuffer(b''.join(dd), dtype=np.float32)

            

Reported by Pylint.