if TICI:
      tzpath = os.path.join("/usr/share/zoneinfo/", timezone)
      subprocess.check_call(f'sudo su -c "ln -snf {tzpath} /data/etc/tmptime && \
                              mv /data/etc/tmptime /data/etc/localtime"', shell=True)
      subprocess.check_call(f'sudo su -c "echo \"{timezone}\" > /data/etc/timezone"', shell=True)
    else:
      subprocess.check_call(f'sudo timedatectl set-timezone {timezone}', shell=True)
  except subprocess.CalledProcessError:
    cloudlog.exception(f"Error setting timezone to {timezone}")

            

Reported by Bandit.

                    tzpath = os.path.join("/usr/share/zoneinfo/", timezone)
      subprocess.check_call(f'sudo su -c "ln -snf {tzpath} /data/etc/tmptime && \
                              mv /data/etc/tmptime /data/etc/localtime"', shell=True)
      subprocess.check_call(f'sudo su -c "echo \"{timezone}\" > /data/etc/timezone"', shell=True)
    else:
      subprocess.check_call(f'sudo timedatectl set-timezone {timezone}', shell=True)
  except subprocess.CalledProcessError:
    cloudlog.exception(f"Error setting timezone to {timezone}")


            

Reported by Bandit.

                                            mv /data/etc/tmptime /data/etc/localtime"', shell=True)
      subprocess.check_call(f'sudo su -c "echo \"{timezone}\" > /data/etc/timezone"', shell=True)
    else:
      subprocess.check_call(f'sudo timedatectl set-timezone {timezone}', shell=True)
  except subprocess.CalledProcessError:
    cloudlog.exception(f"Error setting timezone to {timezone}")


def main():

            

Reported by Bandit.

              import subprocess

import requests
from timezonefinder import TimezoneFinder

from common.params import Params
from selfdrive.hardware import TICI
from selfdrive.swaglog import cloudlog


            

Reported by Pylint.

              import json
import os
import time
import subprocess

import requests
from timezonefinder import TimezoneFinder

from common.params import Params

            

Reported by Bandit.

                tf = TimezoneFinder()

  # Get allowed timezones
  valid_timezones = subprocess.check_output('timedatectl list-timezones', shell=True, encoding='utf8').strip().split('\n')

  while True:
    time.sleep(60)

    is_onroad = not params.get_bool("IsOffroad")

            

Reported by Bandit.

                tf = TimezoneFinder()

  # Get allowed timezones
  valid_timezones = subprocess.check_output('timedatectl list-timezones', shell=True, encoding='utf8').strip().split('\n')

  while True:
    time.sleep(60)

    is_onroad = not params.get_bool("IsOffroad")

            

Reported by Bandit.

              from selfdrive.controls.lib.drive_helpers import CONTROL_N, MPC_COST_LAT, LAT_MPC_N, CAR_ROTATION_RADIUS
from selfdrive.controls.lib.lane_planner import LanePlanner, TRAJECTORY_SIZE
from selfdrive.config import Conversions as CV
import cereal.messaging as messaging
from cereal import log

LaneChangeState = log.LateralPlan.LaneChangeState
LaneChangeDirection = log.LateralPlan.LaneChangeDirection


            

Reported by Pylint.

              from selfdrive.controls.lib.drive_helpers import CONTROL_N, MPC_COST_LAT, LAT_MPC_N, CAR_ROTATION_RADIUS
from selfdrive.controls.lib.lane_planner import LanePlanner, TRAJECTORY_SIZE
from selfdrive.config import Conversions as CV
import cereal.messaging as messaging
from cereal import log

LaneChangeState = log.LateralPlan.LaneChangeState
LaneChangeDirection = log.LateralPlan.LaneChangeDirection


            

Reported by Pylint.

              from selfdrive.controls.lib.lane_planner import LanePlanner, TRAJECTORY_SIZE
from selfdrive.config import Conversions as CV
import cereal.messaging as messaging
from cereal import log

LaneChangeState = log.LateralPlan.LaneChangeState
LaneChangeDirection = log.LateralPlan.LaneChangeDirection

LOG_MPC = os.environ.get('LOG_MPC', False)

            

Reported by Pylint.

                  heading_pts = np.interp(v_ego * self.t_idxs[:LAT_MPC_N + 1], np.linalg.norm(self.path_xyz, axis=1), self.plan_yaw)
    self.y_pts = y_pts

    assert len(y_pts) == LAT_MPC_N + 1
    assert len(heading_pts) == LAT_MPC_N + 1
    # for now CAR_ROTATION_RADIUS is disabled
    # to use it, enable it in the MPC
    assert abs(CAR_ROTATION_RADIUS) < 1e-3
    self.libmpc.run_mpc(self.cur_state, self.mpc_solution,

            

Reported by Bandit.

                  self.y_pts = y_pts

    assert len(y_pts) == LAT_MPC_N + 1
    assert len(heading_pts) == LAT_MPC_N + 1
    # for now CAR_ROTATION_RADIUS is disabled
    # to use it, enable it in the MPC
    assert abs(CAR_ROTATION_RADIUS) < 1e-3
    self.libmpc.run_mpc(self.cur_state, self.mpc_solution,
                        float(v_ego),

            

Reported by Bandit.

                  assert len(heading_pts) == LAT_MPC_N + 1
    # for now CAR_ROTATION_RADIUS is disabled
    # to use it, enable it in the MPC
    assert abs(CAR_ROTATION_RADIUS) < 1e-3
    self.libmpc.run_mpc(self.cur_state, self.mpc_solution,
                        float(v_ego),
                        CAR_ROTATION_RADIUS,
                        list(y_pts),
                        list(heading_pts))

            

Reported by Bandit.

              from common.params_pyx import Params, ParamKeyType, UnknownKeyName, put_nonblocking # pylint: disable=no-name-in-module, import-error
assert Params
assert ParamKeyType
assert UnknownKeyName
assert put_nonblocking

if __name__ == "__main__":
  import sys
  from common.params_pyx import keys # pylint: disable=no-name-in-module, import-error

            

Reported by Bandit.

              from common.params_pyx import Params, ParamKeyType, UnknownKeyName, put_nonblocking # pylint: disable=no-name-in-module, import-error
assert Params
assert ParamKeyType
assert UnknownKeyName
assert put_nonblocking

if __name__ == "__main__":
  import sys
  from common.params_pyx import keys # pylint: disable=no-name-in-module, import-error

            

Reported by Bandit.

              from common.params_pyx import Params, ParamKeyType, UnknownKeyName, put_nonblocking # pylint: disable=no-name-in-module, import-error
assert Params
assert ParamKeyType
assert UnknownKeyName
assert put_nonblocking

if __name__ == "__main__":
  import sys
  from common.params_pyx import keys # pylint: disable=no-name-in-module, import-error

            

Reported by Bandit.

              assert Params
assert ParamKeyType
assert UnknownKeyName
assert put_nonblocking

if __name__ == "__main__":
  import sys
  from common.params_pyx import keys # pylint: disable=no-name-in-module, import-error


            

Reported by Bandit.

                if len(sys.argv) == 3:
    name = sys.argv[1]
    val = sys.argv[2]
    assert name.encode("utf-8") in keys.keys(), f"unknown param: {name}"
    print(f"SET: {name} = {val}")
    params.put(name, val)
  elif len(sys.argv) == 2:
    name = sys.argv[1]
    assert name.encode("utf-8") in keys.keys(), f"unknown param: {name}"

            

Reported by Bandit.

                  params.put(name, val)
  elif len(sys.argv) == 2:
    name = sys.argv[1]
    assert name.encode("utf-8") in keys.keys(), f"unknown param: {name}"
    print(f"GET: {name} = {params.get(name)}")
  else:
    for k in keys.keys():
      print(f"GET: {k} = {params.get(k)}")

            

Reported by Bandit.

              import argparse
import os
import sys
import zmq
import time
import signal
import multiprocessing
from uuid import uuid4
from collections import namedtuple

            

Reported by Pylint.

              from collections import deque
from datetime import datetime

from cereal import log as capnp_log
from cereal.services import service_list
from cereal.messaging import pub_sock, MultiplePublishersError
from cereal.visionipc.visionipc_pyx import VisionIpcServer, VisionStreamType  # pylint: disable=no-name-in-module, import-error
from common import realtime
from common.transformations.camera import eon_f_frame_size, tici_f_frame_size

            

Reported by Pylint.

              from datetime import datetime

from cereal import log as capnp_log
from cereal.services import service_list
from cereal.messaging import pub_sock, MultiplePublishersError
from cereal.visionipc.visionipc_pyx import VisionIpcServer, VisionStreamType  # pylint: disable=no-name-in-module, import-error
from common import realtime
from common.transformations.camera import eon_f_frame_size, tici_f_frame_size


            

Reported by Pylint.

              from datetime import datetime

from cereal import log as capnp_log
from cereal.services import service_list
from cereal.messaging import pub_sock, MultiplePublishersError
from cereal.visionipc.visionipc_pyx import VisionIpcServer, VisionStreamType  # pylint: disable=no-name-in-module, import-error
from common import realtime
from common.transformations.camera import eon_f_frame_size, tici_f_frame_size


            

Reported by Pylint.

              
from cereal import log as capnp_log
from cereal.services import service_list
from cereal.messaging import pub_sock, MultiplePublishersError
from cereal.visionipc.visionipc_pyx import VisionIpcServer, VisionStreamType  # pylint: disable=no-name-in-module, import-error
from common import realtime
from common.transformations.camera import eon_f_frame_size, tici_f_frame_size

from tools.lib.kbhit import KBHit

            

Reported by Pylint.

              
from cereal import log as capnp_log
from cereal.services import service_list
from cereal.messaging import pub_sock, MultiplePublishersError
from cereal.visionipc.visionipc_pyx import VisionIpcServer, VisionStreamType  # pylint: disable=no-name-in-module, import-error
from common import realtime
from common.transformations.camera import eon_f_frame_size, tici_f_frame_size

from tools.lib.kbhit import KBHit

            

Reported by Pylint.

              import selfdrive.boardd.boardd as boardd

from common.realtime import sec_since_boot
from cereal import log
import unittest


def generate_random_can_data_list():
  can_list = []

            

Reported by Pylint.

              
def generate_random_can_data_list():
  can_list = []
  cnt = random.randint(1, 64)
  for _ in range(cnt):
    can_data = np.random.bytes(random.randint(1, 8))
    can_list.append([random.randint(0, 128), random.randint(0, 128), can_data, random.randint(0, 128)])
  return can_list, cnt


            

Reported by Bandit.

                can_list = []
  cnt = random.randint(1, 64)
  for _ in range(cnt):
    can_data = np.random.bytes(random.randint(1, 8))
    can_list.append([random.randint(0, 128), random.randint(0, 128), can_data, random.randint(0, 128)])
  return can_list, cnt


class TestBoarddApiMethods(unittest.TestCase):

            

Reported by Bandit.

                cnt = random.randint(1, 64)
  for _ in range(cnt):
    can_data = np.random.bytes(random.randint(1, 8))
    can_list.append([random.randint(0, 128), random.randint(0, 128), can_data, random.randint(0, 128)])
  return can_list, cnt


class TestBoarddApiMethods(unittest.TestCase):
  def test_correctness(self):

            

Reported by Bandit.

                cnt = random.randint(1, 64)
  for _ in range(cnt):
    can_data = np.random.bytes(random.randint(1, 8))
    can_list.append([random.randint(0, 128), random.randint(0, 128), can_data, random.randint(0, 128)])
  return can_list, cnt


class TestBoarddApiMethods(unittest.TestCase):
  def test_correctness(self):

            

Reported by Bandit.

                cnt = random.randint(1, 64)
  for _ in range(cnt):
    can_data = np.random.bytes(random.randint(1, 8))
    can_list.append([random.randint(0, 128), random.randint(0, 128), can_data, random.randint(0, 128)])
  return can_list, cnt


class TestBoarddApiMethods(unittest.TestCase):
  def test_correctness(self):

            

Reported by Bandit.

              #!/usr/bin/env python3
import os
import subprocess
from typing import List, Optional

from common.basedir import BASEDIR
from selfdrive.swaglog import cloudlog



            

Reported by Bandit.

              

def run_cmd(cmd: List[str]) -> str:
    return subprocess.check_output(cmd, encoding='utf8').strip()


def run_cmd_default(cmd: List[str], default: Optional[str] = None) -> Optional[str]:
  try:
    return run_cmd(cmd)

            

Reported by Bandit.

                  if not prebuilt:
      # This is needed otherwise touched files might show up as modified
      try:
        subprocess.check_call(["git", "update-index", "--refresh"])
      except subprocess.CalledProcessError:
        pass
      dirty = (subprocess.call(["git", "diff-index", "--quiet", branch, "--"]) != 0)

      # Log dirty files

            

Reported by Bandit.

                  if not prebuilt:
      # This is needed otherwise touched files might show up as modified
      try:
        subprocess.check_call(["git", "update-index", "--refresh"])
      except subprocess.CalledProcessError:
        pass
      dirty = (subprocess.call(["git", "diff-index", "--quiet", branch, "--"]) != 0)

      # Log dirty files

            

Reported by Bandit.

                      subprocess.check_call(["git", "update-index", "--refresh"])
      except subprocess.CalledProcessError:
        pass
      dirty = (subprocess.call(["git", "diff-index", "--quiet", branch, "--"]) != 0)

      # Log dirty files
      if dirty and comma_remote:
        try:
          dirty_files = run_cmd(["git", "diff-index", branch, "--"])

            

Reported by Bandit.

                      subprocess.check_call(["git", "update-index", "--refresh"])
      except subprocess.CalledProcessError:
        pass
      dirty = (subprocess.call(["git", "diff-index", "--quiet", branch, "--"]) != 0)

      # Log dirty files
      if dirty and comma_remote:
        try:
          dirty_files = run_cmd(["git", "diff-index", branch, "--"])

            

Reported by Bandit.

              
  def test_getxattr_none(self):
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a is None

  def test_listxattr_none(self):
    l = listxattr(self.tmpfn)
    assert l == []


            

Reported by Bandit.

              
  def test_listxattr_none(self):
    l = listxattr(self.tmpfn)
    assert l == []

  def test_setxattr(self):
    setxattr(self.tmpfn, TestParams.USER_TEST, b'123')
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a == b'123'

            

Reported by Bandit.

                def test_setxattr(self):
    setxattr(self.tmpfn, TestParams.USER_TEST, b'123')
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a == b'123'

  def test_listxattr(self):
    setxattr(self.tmpfn, 'user.test1', b'123')
    setxattr(self.tmpfn, 'user.test2', b'123')
    l = listxattr(self.tmpfn)

            

Reported by Bandit.

                  setxattr(self.tmpfn, 'user.test1', b'123')
    setxattr(self.tmpfn, 'user.test2', b'123')
    l = listxattr(self.tmpfn)
    assert l == ['user.test1', 'user.test2']

  def test_removexattr(self):
    setxattr(self.tmpfn, TestParams.USER_TEST, b'123')
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a == b'123'

            

Reported by Bandit.

                def test_removexattr(self):
    setxattr(self.tmpfn, TestParams.USER_TEST, b'123')
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a == b'123'
    removexattr(self.tmpfn, TestParams.USER_TEST)
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a is None

if __name__ == "__main__":

            

Reported by Bandit.

                  assert a == b'123'
    removexattr(self.tmpfn, TestParams.USER_TEST)
    a = getxattr(self.tmpfn, TestParams.USER_TEST)
    assert a is None

if __name__ == "__main__":
  unittest.main()

            

Reported by Bandit.

              from multiprocessing import Process
from pathlib import Path
from unittest import mock
from websocket import ABNF
from websocket._exceptions import WebSocketConnectionClosedException

from selfdrive import swaglog
from selfdrive.athena import athenad
from selfdrive.athena.athenad import MAX_RETRY_COUNT, dispatcher

            

Reported by Pylint.

              from pathlib import Path
from unittest import mock
from websocket import ABNF
from websocket._exceptions import WebSocketConnectionClosedException

from selfdrive import swaglog
from selfdrive.athena import athenad
from selfdrive.athena.athenad import MAX_RETRY_COUNT, dispatcher
from selfdrive.athena.tests.helpers import MockWebsocket, MockParams, MockApi, EchoSocket, with_http_server

            

Reported by Pylint.

              from selfdrive.athena import athenad
from selfdrive.athena.athenad import MAX_RETRY_COUNT, dispatcher
from selfdrive.athena.tests.helpers import MockWebsocket, MockParams, MockApi, EchoSocket, with_http_server
from cereal import messaging

class TestAthenadMethods(unittest.TestCase):
  @classmethod
  def setUpClass(cls):
    cls.SOCKET_PORT = 45454

            

Reported by Pylint.

                      break

  def test_echo(self):
    assert dispatcher["echo"]("bob") == "bob"

  def test_getMessage(self):
    with self.assertRaises(TimeoutError) as _:
      dispatcher["getMessage"]("controlsState")


            

Reported by Bandit.

                  time.sleep(0.1)
    try:
      deviceState = dispatcher["getMessage"]("deviceState")
      assert deviceState['deviceState']
    finally:
      p.terminate()

  def test_listDataDirectory(self):
    route = '2021-03-29--13-32-47'

            

Reported by Bandit.

                  ws_recv.put_nowait(b'ping')
    try:
      recv = ws_send.get(timeout=5)
      assert recv == (b'ping', ABNF.OPCODE_BINARY), recv
    finally:
      # signal websocket close to athenad.ws_proxy_recv
      ws_recv.put_nowait(WebSocketConnectionClosedException())
      socket_thread.join()


            

Reported by Bandit.

              import sys
import argparse
import json
from hexdump import hexdump
import codecs
codecs.register_error("strict", codecs.backslashreplace_errors)

from cereal import log
import cereal.messaging as messaging

            

Reported by Pylint.

              import codecs
codecs.register_error("strict", codecs.backslashreplace_errors)

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

if __name__ == "__main__":


            

Reported by Pylint.

              codecs.register_error("strict", codecs.backslashreplace_errors)

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

if __name__ == "__main__":

  parser = argparse.ArgumentParser(description='Sniff a communcation socket')

            

Reported by Pylint.

              codecs.register_error("strict", codecs.backslashreplace_errors)

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

if __name__ == "__main__":

  parser = argparse.ArgumentParser(description='Sniff a communcation socket')

            

Reported by Pylint.

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

if __name__ == "__main__":

  parser = argparse.ArgumentParser(description='Sniff a communcation socket')
  parser.add_argument('--pipe', action='store_true')

            

Reported by Pylint.

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

if __name__ == "__main__":

  parser = argparse.ArgumentParser(description='Sniff a communcation socket')
  parser.add_argument('--pipe', action='store_true')

            

Reported by Pylint.

              from typing import Any
from collections import defaultdict

from tqdm import tqdm

import panda.python.uds as uds
from cereal import car
from selfdrive.car.fingerprints import FW_VERSIONS, get_attr_from_cars
from selfdrive.car.isotp_parallel_query import IsoTpParallelQuery

            

Reported by Pylint.

              
from tqdm import tqdm

import panda.python.uds as uds
from cereal import car
from selfdrive.car.fingerprints import FW_VERSIONS, get_attr_from_cars
from selfdrive.car.isotp_parallel_query import IsoTpParallelQuery
from selfdrive.car.toyota.values import CAR as TOYOTA
from selfdrive.swaglog import cloudlog

            

Reported by Pylint.

              
from tqdm import tqdm

import panda.python.uds as uds
from cereal import car
from selfdrive.car.fingerprints import FW_VERSIONS, get_attr_from_cars
from selfdrive.car.isotp_parallel_query import IsoTpParallelQuery
from selfdrive.car.toyota.values import CAR as TOYOTA
from selfdrive.swaglog import cloudlog

            

Reported by Pylint.

              from tqdm import tqdm

import panda.python.uds as uds
from cereal import car
from selfdrive.car.fingerprints import FW_VERSIONS, get_attr_from_cars
from selfdrive.car.isotp_parallel_query import IsoTpParallelQuery
from selfdrive.car.toyota.values import CAR as TOYOTA
from selfdrive.swaglog import cloudlog


            

Reported by Pylint.

              if __name__ == "__main__":
  import time
  import argparse
  import cereal.messaging as messaging
  from selfdrive.car.vin import get_vin

  parser = argparse.ArgumentParser(description='Get firmware version of ECUs')
  parser.add_argument('--scan', action='store_true')
  parser.add_argument('--debug', action='store_true')

            

Reported by Pylint.

              if __name__ == "__main__":
  import time
  import argparse
  import cereal.messaging as messaging
  from selfdrive.car.vin import get_vin

  parser = argparse.ArgumentParser(description='Get firmware version of ECUs')
  parser.add_argument('--scan', action='store_true')
  parser.add_argument('--debug', action='store_true')

            

Reported by Pylint.