import importlib
import unittest
from collections import defaultdict, Counter
from parameterized import parameterized_class

from cereal import log, car
from selfdrive.car.fingerprints import all_known_cars
from selfdrive.car.car_helpers import interfaces
from selfdrive.car.honda.values import HONDA_BOSCH

            

Reported by Pylint.

              from collections import defaultdict, Counter
from parameterized import parameterized_class

from cereal import log, car
from selfdrive.car.fingerprints import all_known_cars
from selfdrive.car.car_helpers import interfaces
from selfdrive.car.honda.values import HONDA_BOSCH
from selfdrive.car.honda.values import CAR as HONDA
from selfdrive.car.chrysler.values import CAR as CHRYSLER

            

Reported by Pylint.

              from collections import defaultdict, Counter
from parameterized import parameterized_class

from cereal import log, car
from selfdrive.car.fingerprints import all_known_cars
from selfdrive.car.car_helpers import interfaces
from selfdrive.car.honda.values import HONDA_BOSCH
from selfdrive.car.honda.values import CAR as HONDA
from selfdrive.car.chrysler.values import CAR as CHRYSLER

            

Reported by Pylint.

              from selfdrive.test.openpilotci import get_url
from tools.lib.logreader import LogReader

from panda.tests.safety import libpandasafety_py
from panda.tests.safety.common import package_can_msg

PandaType = log.PandaState.PandaType

ROUTES = {rt.car_fingerprint: rt.route for rt in routes}

            

Reported by Pylint.

              from selfdrive.test.openpilotci import get_url
from tools.lib.logreader import LogReader

from panda.tests.safety import libpandasafety_py
from panda.tests.safety.common import package_can_msg

PandaType = log.PandaState.PandaType

ROUTES = {rt.car_fingerprint: rt.route for rt in routes}

            

Reported by Pylint.

              from tools.lib.logreader import LogReader

from panda.tests.safety import libpandasafety_py
from panda.tests.safety.common import package_can_msg

PandaType = log.PandaState.PandaType

ROUTES = {rt.car_fingerprint: rt.route for rt in routes}


            

Reported by Pylint.

              from tools.lib.logreader import LogReader

from panda.tests.safety import libpandasafety_py
from panda.tests.safety.common import package_can_msg

PandaType = log.PandaState.PandaType

ROUTES = {rt.car_fingerprint: rt.route for rt in routes}


            

Reported by Pylint.

                  CarInterface, CarController, CarState = interfaces[cls.car_model]

    cls.CP = CarInterface.get_params(cls.car_model, fingerprint, [])
    assert cls.CP

    cls.CI = CarInterface(cls.CP, CarController, CarState)
    assert cls.CI

  def test_car_params(self):

            

Reported by Bandit.

                  assert cls.CP

    cls.CI = CarInterface(cls.CP, CarController, CarState)
    assert cls.CI

  def test_car_params(self):
    if self.CP.dashcamOnly:
      self.skipTest("no need to check carParams for dashcamOnly")


            

Reported by Bandit.

                  os.environ['NO_RADAR_SLEEP'] = "1"
    RadarInterface = importlib.import_module('selfdrive.car.%s.radar_interface' % self.CP.carName).RadarInterface
    RI = RadarInterface(self.CP)
    assert RI

    error_cnt = 0
    for msg in self.can_msgs:
      radar_data = RI.update((msg.as_builder().to_bytes(),))
      if radar_data is not None:

            

Reported by Bandit.

              from collections import Counter
from pathlib import Path

import cereal.messaging as messaging
from cereal.services import service_list
from common.basedir import BASEDIR
from common.timeout import Timeout
from common.params import Params
from selfdrive.hardware import EON, TICI

            

Reported by Pylint.

              from collections import Counter
from pathlib import Path

import cereal.messaging as messaging
from cereal.services import service_list
from common.basedir import BASEDIR
from common.timeout import Timeout
from common.params import Params
from selfdrive.hardware import EON, TICI

            

Reported by Pylint.

              from pathlib import Path

import cereal.messaging as messaging
from cereal.services import service_list
from common.basedir import BASEDIR
from common.timeout import Timeout
from common.params import Params
from selfdrive.hardware import EON, TICI
from selfdrive.loggerd.config import ROOT

            

Reported by Pylint.

              from pathlib import Path

import cereal.messaging as messaging
from cereal.services import service_list
from common.basedir import BASEDIR
from common.timeout import Timeout
from common.params import Params
from selfdrive.hardware import EON, TICI
from selfdrive.loggerd.config import ROOT

            

Reported by Pylint.

              #!/usr/bin/env python3
import json
import os
import subprocess
import time
import numpy as np
import unittest
from collections import Counter
from pathlib import Path

            

Reported by Bandit.

                  # Make sure athena isn't running
    Params().delete("DongleId")
    Params().delete("AthenadPid")
    os.system("pkill -9 -f athena")

    logger_root = Path(ROOT)
    initial_segments = set()
    if logger_root.exists():
      initial_segments = set(Path(ROOT).iterdir())

            

Reported by Bandit.

                  # Make sure athena isn't running
    Params().delete("DongleId")
    Params().delete("AthenadPid")
    os.system("pkill -9 -f athena")

    logger_root = Path(ROOT)
    initial_segments = set()
    if logger_root.exists():
      initial_segments = set(Path(ROOT).iterdir())

            

Reported by Bandit.

                  # start manager and run openpilot for a minute
    try:
      manager_path = os.path.join(BASEDIR, "selfdrive/manager/manager.py")
      proc = subprocess.Popen(["python", manager_path])

      sm = messaging.SubMaster(['carState'])
      with Timeout(150, "controls didn't start"):
        while sm.rcv_frame['carState'] < 0:
          sm.update(1000)

            

Reported by Bandit.

                  # start manager and run openpilot for a minute
    try:
      manager_path = os.path.join(BASEDIR, "selfdrive/manager/manager.py")
      proc = subprocess.Popen(["python", manager_path])

      sm = messaging.SubMaster(['carState'])
      with Timeout(150, "controls didn't start"):
        while sm.rcv_frame['carState'] < 0:
          sm.update(1000)

            

Reported by Bandit.

              import signal
from collections import namedtuple

import capnp
from tqdm import tqdm

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

            

Reported by Pylint.

              from collections import namedtuple

import capnp
from tqdm import tqdm

import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params

            

Reported by Pylint.

              import capnp
from tqdm import tqdm

import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params
from common.timeout import Timeout
from selfdrive.car.fingerprints import FW_VERSIONS

            

Reported by Pylint.

              import capnp
from tqdm import tqdm

import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params
from common.timeout import Timeout
from selfdrive.car.fingerprints import FW_VERSIONS

            

Reported by Pylint.

              from tqdm import tqdm

import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params
from common.timeout import Timeout
from selfdrive.car.fingerprints import FW_VERSIONS
from selfdrive.car.car_helpers import get_car, interfaces

            

Reported by Pylint.

              from tqdm import tqdm

import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params
from common.timeout import Timeout
from selfdrive.car.fingerprints import FW_VERSIONS
from selfdrive.car.car_helpers import get_car, interfaces

            

Reported by Pylint.

              
import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params
from common.timeout import Timeout
from selfdrive.car.fingerprints import FW_VERSIONS
from selfdrive.car.car_helpers import get_car, interfaces
from selfdrive.manager.process import PythonProcess

            

Reported by Pylint.

              
import cereal.messaging as messaging
from cereal import car, log
from cereal.services import service_list
from common.params import Params
from common.timeout import Timeout
from selfdrive.car.fingerprints import FW_VERSIONS
from selfdrive.car.car_helpers import get_car, interfaces
from selfdrive.manager.process import PythonProcess

            

Reported by Pylint.

                        os.environ['SKIP_FW_QUERY'] = "1"
          os.environ['FINGERPRINT'] = car_fingerprint

  assert(type(managed_processes[cfg.proc_name]) is PythonProcess)
  managed_processes[cfg.proc_name].prepare()
  mod = importlib.import_module(managed_processes[cfg.proc_name].module)

  thread = threading.Thread(target=mod.main, args=args)
  thread.daemon = True

            

Reported by Bandit.

                    clear_partition_hash(target_slot_number, partition)

  while True:
    out = subprocess.check_output(f"abctl --set_active {target_slot_number}", shell=True, stderr=subprocess.STDOUT, encoding='utf8')
    if ("No such file or directory" not in out) and ("lun as boot lun" in out):
      cloudlog.info(f"Swap successfull {out}")
      break
    else:
      cloudlog.error(f"Swap failed {out}")

            

Reported by Bandit.

                cloudlog.info(f"Target slot {target_slot_number}")

  # set target slot as unbootable
  os.system(f"abctl --set_unbootable {target_slot_number}")

  for partition in update:
    success = False

    for retries in range(10):

            

Reported by Bandit.

              import hashlib
import requests
import struct
import subprocess
import os
from typing import Generator

SPARSE_CHUNK_FMT = struct.Struct('H2xI4x')


            

Reported by Bandit.

              def unsparsify(f: StreamingDecompressor) -> Generator[bytes, None, None]:
  # https://source.android.com/devices/bootloader/images#sparse-format
  magic = struct.unpack("I", f.read(4))[0]
  assert(magic == 0xed26ff3a)

  # Version
  major = struct.unpack("H", f.read(2))[0]
  minor = struct.unpack("H", f.read(2))[0]
  assert(major == 1 and minor == 0)

            

Reported by Bandit.

                # Version
  major = struct.unpack("H", f.read(2))[0]
  minor = struct.unpack("H", f.read(2))[0]
  assert(major == 1 and minor == 0)

  f.read(2)  # file header size
  f.read(2)  # chunk header size

  block_sz = struct.unpack("I", f.read(4))[0]

            

Reported by Bandit.

              

def get_target_slot_number() -> int:
  current_slot = subprocess.check_output(["abctl", "--boot_slot"], encoding='utf-8').strip()
  return 1 if current_slot == "_a" else 0


def slot_number_to_suffix(slot_number: int) -> str:
  assert slot_number in (0, 1)

            

Reported by Bandit.

              

def get_target_slot_number() -> int:
  current_slot = subprocess.check_output(["abctl", "--boot_slot"], encoding='utf-8').strip()
  return 1 if current_slot == "_a" else 0


def slot_number_to_suffix(slot_number: int) -> str:
  assert slot_number in (0, 1)

            

Reported by Bandit.

              

def slot_number_to_suffix(slot_number: int) -> str:
  assert slot_number in (0, 1)
  return '_a' if slot_number == 0 else '_b'


def get_partition_path(target_slot_number: int, partition: dict) -> str:
  path = f"/dev/disk/by-partlabel/{partition['name']}"

            

Reported by Bandit.

              from typing import Any

import requests
from jsonrpc import JSONRPCResponseManager, dispatcher
from websocket import ABNF, WebSocketTimeoutException, WebSocketException, create_connection

import cereal.messaging as messaging
from cereal.services import service_list
from common.api import Api

            

Reported by Pylint.

              
import requests
from jsonrpc import JSONRPCResponseManager, dispatcher
from websocket import ABNF, WebSocketTimeoutException, WebSocketException, create_connection

import cereal.messaging as messaging
from cereal.services import service_list
from common.api import Api
from common.file_helpers import CallbackReader

            

Reported by Pylint.

              from jsonrpc import JSONRPCResponseManager, dispatcher
from websocket import ABNF, WebSocketTimeoutException, WebSocketException, create_connection

import cereal.messaging as messaging
from cereal.services import service_list
from common.api import Api
from common.file_helpers import CallbackReader
from common.basedir import PERSIST
from common.params import Params

            

Reported by Pylint.

              from jsonrpc import JSONRPCResponseManager, dispatcher
from websocket import ABNF, WebSocketTimeoutException, WebSocketException, create_connection

import cereal.messaging as messaging
from cereal.services import service_list
from common.api import Api
from common.file_helpers import CallbackReader
from common.basedir import PERSIST
from common.params import Params

            

Reported by Pylint.

              from websocket import ABNF, WebSocketTimeoutException, WebSocketException, create_connection

import cereal.messaging as messaging
from cereal.services import service_list
from common.api import Api
from common.file_helpers import CallbackReader
from common.basedir import PERSIST
from common.params import Params
from common.realtime import sec_since_boot

            

Reported by Pylint.

              from websocket import ABNF, WebSocketTimeoutException, WebSocketException, create_connection

import cereal.messaging as messaging
from cereal.services import service_list
from common.api import Api
from common.file_helpers import CallbackReader
from common.basedir import PERSIST
from common.params import Params
from common.realtime import sec_since_boot

            

Reported by Pylint.

                  return 404

  item = UploadItem(path=path, url=url, headers=headers, created_at=int(time.time() * 1000), id=None)
  upload_id = hashlib.sha1(str(item).encode()).hexdigest()
  item = item._replace(id=upload_id)

  upload_queue.put_nowait(item)

  return {"enqueued": 1, "item": item._asdict()}

            

Reported by Bandit.

              

def backoff(retries):
  return random.randrange(0, min(128, int(2 ** retries)))


def manage_tokens(api):
  if not TICI:
    return

            

Reported by Bandit.

              import time
from cffi import FFI

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

from selfdrive.manager.process_config import managed_processes


            

Reported by Pylint.

              from cffi import FFI

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

from selfdrive.manager.process_config import managed_processes

SENSOR_DECIMATION = 1

            

Reported by Pylint.

              from cffi import FFI

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

from selfdrive.manager.process_config import managed_processes

SENSOR_DECIMATION = 1

            

Reported by Pylint.

              
  def test_liblocalizer(self):
    msg = messaging.new_message('liveCalibration')
    msg.liveCalibration.validBlocks = random.randint(1, 10)
    msg.liveCalibration.rpyCalib = [random.random() / 10 for _ in range(3)]

    self.localizer_handle_msg(msg)
    liveloc = self.localizer_get_msg()
    self.assertTrue(liveloc is not None)

            

Reported by Bandit.

                def test_liblocalizer(self):
    msg = messaging.new_message('liveCalibration')
    msg.liveCalibration.validBlocks = random.randint(1, 10)
    msg.liveCalibration.rpyCalib = [random.random() / 10 for _ in range(3)]

    self.localizer_handle_msg(msg)
    liveloc = self.localizer_get_msg()
    self.assertTrue(liveloc is not None)


            

Reported by Bandit.

                  # first reset params
    Params().put('LastGPSPosition', json.dumps({"latitude": 0.0, "longitude": 0.0, "altitude": 0.0}))

    lat = 30 + (random.random() * 10.0)
    lon = -70 + (random.random() * 10.0)
    alt = 5 + (random.random() * 10.0)

    for _ in range(1000):  # because of kalman filter, send often
      msg = messaging.new_message('gpsLocationExternal')

            

Reported by Bandit.

                  Params().put('LastGPSPosition', json.dumps({"latitude": 0.0, "longitude": 0.0, "altitude": 0.0}))

    lat = 30 + (random.random() * 10.0)
    lon = -70 + (random.random() * 10.0)
    alt = 5 + (random.random() * 10.0)

    for _ in range(1000):  # because of kalman filter, send often
      msg = messaging.new_message('gpsLocationExternal')
      msg.logMonoTime = 0

            

Reported by Bandit.

              
    lat = 30 + (random.random() * 10.0)
    lon = -70 + (random.random() * 10.0)
    alt = 5 + (random.random() * 10.0)

    for _ in range(1000):  # because of kalman filter, send often
      msg = messaging.new_message('gpsLocationExternal')
      msg.logMonoTime = 0
      msg.gpsLocationExternal.flags = 1

            

Reported by Bandit.

              from selfdrive.version import dirty

MAX_CACHE_SIZE = 2e9
CACHE_DIR = Path("/data/scons_cache" if TICI else "/tmp/scons_cache")

TOTAL_SCONS_NODES = 2405
MAX_BUILD_PROGRESS = 100
PREBUILT = os.path.exists(os.path.join(BASEDIR, 'prebuilt'))


            

Reported by Bandit.

              #!/usr/bin/env python3
import os
import subprocess
import sys
import time
import textwrap
from pathlib import Path

# NOTE: Do NOT import anything here that needs be built (e.g. params)

            

Reported by Bandit.

                j_flag = "" if nproc is None else f"-j{nproc - 1}"

  for retry in [True, False]:
    scons = subprocess.Popen(["scons", j_flag, "--cache-populate"], cwd=BASEDIR, env=env, stderr=subprocess.PIPE)

    compile_output = []

    # Read progress from stderr and update spinner
    while scons.poll() is None:

            

Reported by Bandit.

                j_flag = "" if nproc is None else f"-j{nproc - 1}"

  for retry in [True, False]:
    scons = subprocess.Popen(["scons", j_flag, "--cache-populate"], cwd=BASEDIR, env=env, stderr=subprocess.PIPE)

    compile_output = []

    # Read progress from stderr and update spinner
    while scons.poll() is None:

            

Reported by Bandit.

                      elif len(line):
          compile_output.append(line)
          print(line.decode('utf8', 'replace'))
      except Exception:
        pass

    if scons.returncode != 0:
      # Read remaining output
      r = scons.stderr.read().split(b'\n')

            

Reported by Bandit.

                        for i in range(3, -1, -1):
            print("....%d" % i)
            time.sleep(1)
          subprocess.check_call(["scons", "-c"], cwd=BASEDIR, env=env)
        else:
          print("scons build failed after retry")
          sys.exit(1)
      else:
        # Build failed log errors

            

Reported by Bandit.

                        for i in range(3, -1, -1):
            print("....%d" % i)
            time.sleep(1)
          subprocess.check_call(["scons", "-c"], cwd=BASEDIR, env=env)
        else:
          print("scons build failed after retry")
          sys.exit(1)
      else:
        # Build failed log errors

            

Reported by Bandit.

              from PIL import Image
from typing import List

import cereal.messaging as messaging
from common.params import Params
from common.realtime import DT_MDL
from common.transformations.camera import eon_f_frame_size, eon_d_frame_size, tici_f_frame_size
from selfdrive.hardware import TICI
from selfdrive.controls.lib.alertmanager import set_offroad_alert

            

Reported by Pylint.

              from PIL import Image
from typing import List

import cereal.messaging as messaging
from common.params import Params
from common.realtime import DT_MDL
from common.transformations.camera import eon_f_frame_size, eon_d_frame_size, tici_f_frame_size
from selfdrive.hardware import TICI
from selfdrive.controls.lib.alertmanager import set_offroad_alert

            

Reported by Pylint.

                pic, fpic = snapshot()
  if pic is not None:
    print(pic.shape)
    jpeg_write("/tmp/back.jpg", pic)
    if fpic is not None:
      jpeg_write("/tmp/front.jpg", fpic)
  else:
    print("Error taking snapshot")

            

Reported by Bandit.

                  print(pic.shape)
    jpeg_write("/tmp/back.jpg", pic)
    if fpic is not None:
      jpeg_write("/tmp/front.jpg", fpic)
  else:
    print("Error taking snapshot")

            

Reported by Bandit.

              #!/usr/bin/env python3
import os
import subprocess
import time

import numpy as np
from PIL import Image
from typing import List


            

Reported by Bandit.

              
  # Check if camerad is already started
  try:
    subprocess.check_call(["pgrep", "camerad"])
    print("Camerad already running")
    params.put_bool("IsTakingSnapshot", False)
    params.delete("Offroad_IsTakingSnapshot")
    return None, None
  except subprocess.CalledProcessError:

            

Reported by Bandit.

              
  # Check if camerad is already started
  try:
    subprocess.check_call(["pgrep", "camerad"])
    print("Camerad already running")
    params.put_bool("IsTakingSnapshot", False)
    params.delete("Offroad_IsTakingSnapshot")
    return None, None
  except subprocess.CalledProcessError:

            

Reported by Bandit.

              import sys

import numpy as np
import sympy as sp

from selfdrive.locationd.models.constants import ObservationKind
from rednose.helpers.ekf_sym import EKF_sym, gen_code
from rednose.helpers.lst_sq_computer import LstSqComputer
from rednose.helpers.sympy_helpers import euler_rotate, quat_matrix_r, quat_rotate

            

Reported by Pylint.

              import sympy as sp

from selfdrive.locationd.models.constants import ObservationKind
from rednose.helpers.ekf_sym import EKF_sym, gen_code
from rednose.helpers.lst_sq_computer import LstSqComputer
from rednose.helpers.sympy_helpers import euler_rotate, quat_matrix_r, quat_rotate

EARTH_GM = 3.986005e14  # m^3/s^2 (gravitational constant * mass of earth)


            

Reported by Pylint.

              
from selfdrive.locationd.models.constants import ObservationKind
from rednose.helpers.ekf_sym import EKF_sym, gen_code
from rednose.helpers.lst_sq_computer import LstSqComputer
from rednose.helpers.sympy_helpers import euler_rotate, quat_matrix_r, quat_rotate

EARTH_GM = 3.986005e14  # m^3/s^2 (gravitational constant * mass of earth)



            

Reported by Pylint.

              from selfdrive.locationd.models.constants import ObservationKind
from rednose.helpers.ekf_sym import EKF_sym, gen_code
from rednose.helpers.lst_sq_computer import LstSqComputer
from rednose.helpers.sympy_helpers import euler_rotate, quat_matrix_r, quat_rotate

EARTH_GM = 3.986005e14  # m^3/s^2 (gravitational constant * mass of earth)


def parse_prr(m):

            

Reported by Pylint.

              

def parse_prr(m):
  from laika.raw_gnss import GNSSMeasurement
  sat_pos_vel_i = np.concatenate((m[GNSSMeasurement.SAT_POS],
                                  m[GNSSMeasurement.SAT_VEL]))
  R_i = np.atleast_2d(m[GNSSMeasurement.PRR_STD]**2)
  z_i = m[GNSSMeasurement.PRR]
  return z_i, R_i, sat_pos_vel_i

            

Reported by Pylint.

              

def parse_pr(m):
  from laika.raw_gnss import GNSSMeasurement
  pseudorange = m[GNSSMeasurement.PR]
  pseudorange_stdev = m[GNSSMeasurement.PR_STD]
  sat_pos_freq_i = np.concatenate((m[GNSSMeasurement.SAT_POS],
                                   np.array([m[GNSSMeasurement.GLONASS_FREQ]])))
  z_i = np.atleast_1d(pseudorange)

            

Reported by Pylint.

                  return ret

  def predict_and_update_msckf_test(self, test_data, t, kind):
    assert self.N > 0
    z = test_data
    R = np.zeros((len(test_data), len(z[0]), len(z[0])))
    ecef_pos = [self.x[:3]]
    for i, _ in enumerate(z):
      R[i, :, :] = np.diag([0.1**2] * len(z[0]))

            

Reported by Bandit.

                  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.