import argparse
import os
import sys
from common.basedir import BASEDIR
from tools.lib.logreader import MultiLogIterator
from tools.lib.route import Route

os.environ['BASEDIR'] = BASEDIR


            

Reported by Pylint.

              import os
import sys
from common.basedir import BASEDIR
from tools.lib.logreader import MultiLogIterator
from tools.lib.route import Route

os.environ['BASEDIR'] = BASEDIR



            

Reported by Pylint.

              import sys
from common.basedir import BASEDIR
from tools.lib.logreader import MultiLogIterator
from tools.lib.route import Route

os.environ['BASEDIR'] = BASEDIR


def get_arg_parser():

            

Reported by Pylint.

                return parser


def main(argv):
  args = get_arg_parser().parse_args(sys.argv[1:])
  if not args.data_dir:
    print('Data directory invalid.')
    return


            

Reported by Pylint.

              #!/usr/bin/env python
import argparse
import os
import sys
from common.basedir import BASEDIR
from tools.lib.logreader import MultiLogIterator
from tools.lib.route import Route

os.environ['BASEDIR'] = BASEDIR

            

Reported by Pylint.

              os.environ['BASEDIR'] = BASEDIR


def get_arg_parser():
  parser = argparse.ArgumentParser(
      description="Unlogging and save to file",
      formatter_class=argparse.ArgumentDefaultsHelpFormatter)

  parser.add_argument("data_dir", nargs='?',

            

Reported by Pylint.

              

def get_arg_parser():
  parser = argparse.ArgumentParser(
      description="Unlogging and save to file",
      formatter_class=argparse.ArgumentDefaultsHelpFormatter)

  parser.add_argument("data_dir", nargs='?',
                              help="Path to directory in which log and camera files are located.")

            

Reported by Pylint.

                    description="Unlogging and save to file",
      formatter_class=argparse.ArgumentDefaultsHelpFormatter)

  parser.add_argument("data_dir", nargs='?',
                              help="Path to directory in which log and camera files are located.")
  parser.add_argument("route_name", type=(lambda x: x.replace("#", "|")), nargs="?",
                      help="The route whose messages will be published.")
  parser.add_argument("--out_path", nargs='?', default='/data/ubloxRaw.stream',
                      help="Output pickle file path")

            

Reported by Pylint.

              
  parser.add_argument("data_dir", nargs='?',
                              help="Path to directory in which log and camera files are located.")
  parser.add_argument("route_name", type=(lambda x: x.replace("#", "|")), nargs="?",
                      help="The route whose messages will be published.")
  parser.add_argument("--out_path", nargs='?', default='/data/ubloxRaw.stream',
                      help="Output pickle file path")
  return parser


            

Reported by Pylint.

                                            help="Path to directory in which log and camera files are located.")
  parser.add_argument("route_name", type=(lambda x: x.replace("#", "|")), nargs="?",
                      help="The route whose messages will be published.")
  parser.add_argument("--out_path", nargs='?', default='/data/ubloxRaw.stream',
                      help="Output pickle file path")
  return parser


def main(argv):

            

Reported by Pylint.

              import datetime
import random

from common.basedir import BASEDIR
import cereal.messaging as messaging

if __name__ == "__main__":

  sound_dir = os.path.join(BASEDIR, "selfdrive/assets/sounds")

            

Reported by Pylint.

              import random

from common.basedir import BASEDIR
import cereal.messaging as messaging

if __name__ == "__main__":

  sound_dir = os.path.join(BASEDIR, "selfdrive/assets/sounds")
  sound_files = [f for f in os.listdir(sound_dir) if f.endswith(".wav")]

            

Reported by Pylint.

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

from common.basedir import BASEDIR
import cereal.messaging as messaging

            

Reported by Pylint.

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

from common.basedir import BASEDIR
import cereal.messaging as messaging

            

Reported by Bandit.

              
if __name__ == "__main__":

  sound_dir = os.path.join(BASEDIR, "selfdrive/assets/sounds")
  sound_files = [f for f in os.listdir(sound_dir) if f.endswith(".wav")]
  play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

  print("disabling charging")
  os.system('echo "0" > /sys/class/power_supply/battery/charging_enabled')

            

Reported by Pylint.

              if __name__ == "__main__":

  sound_dir = os.path.join(BASEDIR, "selfdrive/assets/sounds")
  sound_files = [f for f in os.listdir(sound_dir) if f.endswith(".wav")]
  play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

  print("disabling charging")
  os.system('echo "0" > /sys/class/power_supply/battery/charging_enabled')


            

Reported by Pylint.

              
  sound_dir = os.path.join(BASEDIR, "selfdrive/assets/sounds")
  sound_files = [f for f in os.listdir(sound_dir) if f.endswith(".wav")]
  play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

  print("disabling charging")
  os.system('echo "0" > /sys/class/power_supply/battery/charging_enabled')

  os.environ["LD_LIBRARY_PATH"] = ""

            

Reported by Pylint.

                sound_files = [f for f in os.listdir(sound_dir) if f.endswith(".wav")]
  play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

  print("disabling charging")
  os.system('echo "0" > /sys/class/power_supply/battery/charging_enabled')

  os.environ["LD_LIBRARY_PATH"] = ""

  sm = messaging.SubMaster(["deviceState"])

            

Reported by Pylint.

                play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

  print("disabling charging")
  os.system('echo "0" > /sys/class/power_supply/battery/charging_enabled')

  os.environ["LD_LIBRARY_PATH"] = ""

  sm = messaging.SubMaster(["deviceState"])


            

Reported by Bandit.

                play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

  print("disabling charging")
  os.system('echo "0" > /sys/class/power_supply/battery/charging_enabled')

  os.environ["LD_LIBRARY_PATH"] = ""

  sm = messaging.SubMaster(["deviceState"])


            

Reported by Bandit.

              import subprocess
import time
from pathlib import Path
from common.basedir import BASEDIR

os.environ["LD_LIBRARY_PATH"] = ""

# pull this from the provisioning tests
play_sound = os.path.join(BASEDIR, "selfdrive/ui/test/play_sound")

            

Reported by Pylint.

              
    # write stats
    stats = f"running time {time.monotonic() - start_time}s, cycle {frame}"
    with open("/tmp/sound_stats.txt", "a") as f:
      f.write(stats)
    print(stats)
    frame +=1

if __name__ == "__main__":

            

Reported by Bandit.

              #!/usr/bin/env python3
import os
import random
import subprocess
import time
from pathlib import Path
from common.basedir import BASEDIR

os.environ["LD_LIBRARY_PATH"] = ""

            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import random
import subprocess
import time
from pathlib import Path
from common.basedir import BASEDIR

os.environ["LD_LIBRARY_PATH"] = ""

            

Reported by Bandit.

              waste = os.path.join(BASEDIR, "scripts/waste")
sound_path = Path(os.path.join(BASEDIR, "selfdrive/assets/sounds"))

def sound_test():

  # max volume
  vol = 15
  sound_files = [p.absolute() for p in sound_path.iterdir() if str(p).endswith(".wav")]


            

Reported by Pylint.

              def sound_test():

  # max volume
  vol = 15
  sound_files = [p.absolute() for p in sound_path.iterdir() if str(p).endswith(".wav")]

  # start waste
  p = subprocess.Popen([waste], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


            

Reported by Pylint.

              
  # max volume
  vol = 15
  sound_files = [p.absolute() for p in sound_path.iterdir() if str(p).endswith(".wav")]

  # start waste
  p = subprocess.Popen([waste], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

  start_time = time.monotonic()

            

Reported by Pylint.

                sound_files = [p.absolute() for p in sound_path.iterdir() if str(p).endswith(".wav")]

  # start waste
  p = subprocess.Popen([waste], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

  start_time = time.monotonic()
  frame = 0
  while True:
    # start a few processes

            

Reported by Pylint.

                sound_files = [p.absolute() for p in sound_path.iterdir() if str(p).endswith(".wav")]

  # start waste
  p = subprocess.Popen([waste], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

  start_time = time.monotonic()
  frame = 0
  while True:
    # start a few processes

            

Reported by Pylint.

                sound_files = [p.absolute() for p in sound_path.iterdir() if str(p).endswith(".wav")]

  # start waste
  p = subprocess.Popen([waste], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

  start_time = time.monotonic()
  frame = 0
  while True:
    # start a few processes

            

Reported by Bandit.

                      # dynamics
        self.f_impl_expr = None
        """
        CasADi expression for the implicit dynamics :math:`f_\\text{impl}(\dot{x}, x, u, z, p) = 0`.
        Used if :py:attr:`acados_template.acados_ocp.AcadosOcpOptions.integrator_type` == 'IRK'.
        Default: :code:`None`
        """
        self.f_expl_expr = None
        """

            

Reported by Pylint.

                      """
        self.f_expl_expr = None
        """
        CasADi expression for the explicit dynamics :math:`\dot{x} = f_\\text{expl}(x, u, p)`.
        Used if :py:attr:`acados_template.acados_ocp.AcadosOcpOptions.integrator_type` == 'ERK'.
        Default: :code:`None`
        """
        self.disc_dyn_expr = None
        """

            

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.

              #


class AcadosModel():
    """
    Class containing all the information to code generate the external CasADi functions
    that are needed when creating an acados ocp solver or acados integrator.
    Thus, this class contains:
    

            

Reported by Pylint.

              #


class AcadosModel():
    """
    Class containing all the information to code generate the external CasADi functions
    that are needed when creating an acados ocp solver or acados integrator.
    Thus, this class contains:
    

            

Reported by Pylint.

                  Class containing all the information to code generate the external CasADi functions
    that are needed when creating an acados ocp solver or acados integrator.
    Thus, this class contains:
    
    a) the :py:attr:`name` of the model,
    b) all CasADi variables/expressions needed in the CasADi function generation process.
    """
    def __init__(self):
        ## common for OCP and Integrator

            

Reported by Pylint.

                      """
        The model name is used for code generation. Type: string. Default: :code:`None`
        """
        self.x = None           #: CasADi variable describing the state of the system; Default: :code:`None`
        self.xdot = None        #: CasADi variable describing the derivative of the state wrt time; Default: :code:`None`
        self.u = None           #: CasADi variable describing the input of the system; Default: :code:`None`
        self.z = []             #: CasADi variable describing the algebraic variables of the DAE; Default: :code:`empty`
        self.p = []             #: CasADi variable describing parameters of the DAE; Default: :code:`empty`
        # dynamics

            

Reported by Pylint.

                      """
        The model name is used for code generation. Type: string. Default: :code:`None`
        """
        self.x = None           #: CasADi variable describing the state of the system; Default: :code:`None`
        self.xdot = None        #: CasADi variable describing the derivative of the state wrt time; Default: :code:`None`
        self.u = None           #: CasADi variable describing the input of the system; Default: :code:`None`
        self.z = []             #: CasADi variable describing the algebraic variables of the DAE; Default: :code:`empty`
        self.p = []             #: CasADi variable describing parameters of the DAE; Default: :code:`empty`
        # dynamics

            

Reported by Pylint.

                      The model name is used for code generation. Type: string. Default: :code:`None`
        """
        self.x = None           #: CasADi variable describing the state of the system; Default: :code:`None`
        self.xdot = None        #: CasADi variable describing the derivative of the state wrt time; Default: :code:`None`
        self.u = None           #: CasADi variable describing the input of the system; Default: :code:`None`
        self.z = []             #: CasADi variable describing the algebraic variables of the DAE; Default: :code:`empty`
        self.p = []             #: CasADi variable describing parameters of the DAE; Default: :code:`empty`
        # dynamics
        self.f_impl_expr = None

            

Reported by Pylint.

                      """
        self.x = None           #: CasADi variable describing the state of the system; Default: :code:`None`
        self.xdot = None        #: CasADi variable describing the derivative of the state wrt time; Default: :code:`None`
        self.u = None           #: CasADi variable describing the input of the system; Default: :code:`None`
        self.z = []             #: CasADi variable describing the algebraic variables of the DAE; Default: :code:`empty`
        self.p = []             #: CasADi variable describing parameters of the DAE; Default: :code:`empty`
        # dynamics
        self.f_impl_expr = None
        """

            

Reported by Pylint.

              #

import os
from casadi import *
from .utils import ALLOWED_CASADI_VERSIONS, casadi_length, casadi_version_warning

def generate_c_code_nls_cost( model, cost_name, stage_type, opts ):

    casadi_version = CasadiMeta.version()

            

Reported by Pylint.

              
import os
from casadi import *
from .utils import ALLOWED_CASADI_VERSIONS, casadi_length, casadi_version_warning

def generate_c_code_nls_cost( model, cost_name, stage_type, opts ):

    casadi_version = CasadiMeta.version()
    casadi_opts = dict(mex=False, casadi_int='int', casadi_real='double')

            

Reported by Pylint.

              
def generate_c_code_nls_cost( model, cost_name, stage_type, opts ):

    casadi_version = CasadiMeta.version()
    casadi_opts = dict(mex=False, casadi_int='int', casadi_real='double')

    if casadi_version not in (ALLOWED_CASADI_VERSIONS):
        casadi_version_warning(casadi_version)


            

Reported by Pylint.

                  x = model.x
    p = model.p

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

    if stage_type == 'terminal':

            

Reported by Pylint.

                  p = model.p

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

    if stage_type == 'terminal':
        middle_name = '_cost_y_e'

            

Reported by Pylint.

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

    if stage_type == 'terminal':
        middle_name = '_cost_y_e'
        u = symbol('u', 0, 0)
        cost_expr = model.cost_y_expr_e

            

Reported by Pylint.

                  os.chdir(gen_dir_location)

    # set up expressions
    cost_jac_expr = transpose(jacobian(cost_expr, vertcat(u, x)))

    ny = casadi_length(cost_expr)

    y = symbol('y', ny, 1)


            

Reported by Pylint.

                  os.chdir(gen_dir_location)

    # set up expressions
    cost_jac_expr = transpose(jacobian(cost_expr, vertcat(u, x)))

    ny = casadi_length(cost_expr)

    y = symbol('y', ny, 1)


            

Reported by Pylint.

                  os.chdir(gen_dir_location)

    # set up expressions
    cost_jac_expr = transpose(jacobian(cost_expr, vertcat(u, x)))

    ny = casadi_length(cost_expr)

    y = symbol('y', ny, 1)


            

Reported by Pylint.

              
    y = symbol('y', ny, 1)

    y_adj = jtimes(cost_expr, vertcat(u, x), y, True)
    y_hess = jacobian(y_adj, vertcat(u, x))

    ## generate C code
    suffix_name = '_fun'
    fun_name = cost_name + middle_name + suffix_name

            

Reported by Pylint.

              #

import os
from casadi import *
from .utils import ALLOWED_CASADI_VERSIONS, casadi_length, casadi_version_warning

def generate_c_code_discrete_dynamics( model, opts ):

    casadi_version = CasadiMeta.version()

            

Reported by Pylint.

              
import os
from casadi import *
from .utils import ALLOWED_CASADI_VERSIONS, casadi_length, casadi_version_warning

def generate_c_code_discrete_dynamics( model, opts ):

    casadi_version = CasadiMeta.version()
    casadi_opts = dict(mex=False, casadi_int='int', casadi_real='double')

            

Reported by Pylint.

              
def generate_c_code_discrete_dynamics( model, opts ):

    casadi_version = CasadiMeta.version()
    casadi_opts = dict(mex=False, casadi_int='int', casadi_real='double')

    if casadi_version not in (ALLOWED_CASADI_VERSIONS):
        casadi_version_warning(casadi_version)
    

            

Reported by Pylint.

                  nx = x.size()[0]


    if isinstance(phi, casadi.MX):
        symbol = MX.sym
    elif isinstance(phi, casadi.SX):
        symbol = SX.sym
    else:
        Exception("generate_c_code_disc_dyn: disc_dyn_expr must be a CasADi expression, you have type: {}".format(type(phi)))

            

Reported by Pylint.

              

    if isinstance(phi, casadi.MX):
        symbol = MX.sym
    elif isinstance(phi, casadi.SX):
        symbol = SX.sym
    else:
        Exception("generate_c_code_disc_dyn: disc_dyn_expr must be a CasADi expression, you have type: {}".format(type(phi)))


            

Reported by Pylint.

              
    if isinstance(phi, casadi.MX):
        symbol = MX.sym
    elif isinstance(phi, casadi.SX):
        symbol = SX.sym
    else:
        Exception("generate_c_code_disc_dyn: disc_dyn_expr must be a CasADi expression, you have type: {}".format(type(phi)))

    # assume nx1 = nx !!!

            

Reported by Pylint.

                  if isinstance(phi, casadi.MX):
        symbol = MX.sym
    elif isinstance(phi, casadi.SX):
        symbol = SX.sym
    else:
        Exception("generate_c_code_disc_dyn: disc_dyn_expr must be a CasADi expression, you have type: {}".format(type(phi)))

    # assume nx1 = nx !!!
    lam = symbol('lam', nx, 1)

            

Reported by Pylint.

                  lam = symbol('lam', nx, 1)

    # generate jacobians
    ux = vertcat(u,x)
    jac_ux = jacobian(phi, ux)
    # generate adjoint
    adj_ux = jtimes(phi, ux, lam, True)
    # generate hessian
    hess_ux = jacobian(adj_ux, ux)

            

Reported by Pylint.

              
    # generate jacobians
    ux = vertcat(u,x)
    jac_ux = jacobian(phi, ux)
    # generate adjoint
    adj_ux = jtimes(phi, ux, lam, True)
    # generate hessian
    hess_ux = jacobian(adj_ux, ux)
    

            

Reported by Pylint.

                  ux = vertcat(u,x)
    jac_ux = jacobian(phi, ux)
    # generate adjoint
    adj_ux = jtimes(phi, ux, lam, True)
    # generate hessian
    hess_ux = jacobian(adj_ux, ux)
    
    ## change directory
    code_export_dir = opts["code_export_directory"]

            

Reported by Pylint.

              
    if cache_path and os.path.exists(cache_path):
      with open(cache_path, "rb") as cache_file:
        cache_value = pickle.load(cache_file)
    else:
      cache_value = func(fn, *args, **kwargs)

      if cache_path:
        with atomic_write_in_dir(cache_path, mode="wb", overwrite=True) as cache_file:

            

Reported by Bandit.

                if not os.path.exists(cache_path):
    return None
  with open(cache_path, "rb") as cache_file:
    return pickle.load(cache_file)


def read_file_check_size(f, sz, cookie):
  buff = bytearray(sz)
  bytes_read = f.readinto(buff)

            

Reported by Bandit.

              # pylint: skip-file
import json
import os
import pickle
import struct
import subprocess
import tempfile
import threading
from functools import wraps

            

Reported by Bandit.

              import os
import pickle
import struct
import subprocess
import tempfile
import threading
from functools import wraps

import numpy as np

            

Reported by Bandit.

                cmd += [fn]

  try:
    ffprobe_output = subprocess.check_output(cmd)
  except subprocess.CalledProcessError:
    raise DataUnreadableError(fn)

  return json.loads(ffprobe_output)


            

Reported by Bandit.

                vidindex_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "vidindex")
  vidindex = os.path.join(vidindex_dir, "vidindex")

  subprocess.check_call(["make"], cwd=vidindex_dir, stdout=open("/dev/null", "w"))

  with tempfile.NamedTemporaryFile() as prefix_f, \
       tempfile.NamedTemporaryFile() as index_f:
    try:
      subprocess.check_call([vidindex, typ, fn, prefix_f.name, index_f.name])

            

Reported by Bandit.

                vidindex_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "vidindex")
  vidindex = os.path.join(vidindex_dir, "vidindex")

  subprocess.check_call(["make"], cwd=vidindex_dir, stdout=open("/dev/null", "w"))

  with tempfile.NamedTemporaryFile() as prefix_f, \
       tempfile.NamedTemporaryFile() as index_f:
    try:
      subprocess.check_call([vidindex, typ, fn, prefix_f.name, index_f.name])

            

Reported by Bandit.

                with tempfile.NamedTemporaryFile() as prefix_f, \
       tempfile.NamedTemporaryFile() as index_f:
    try:
      subprocess.check_call([vidindex, typ, fn, prefix_f.name, index_f.name])
    except subprocess.CalledProcessError:
      raise DataUnreadableError("vidindex failed on file %s" % fn)
    with open(index_f.name, "rb") as f:
      index = f.read()
    with open(prefix_f.name, "rb") as f:

            

Reported by Bandit.

              
  index = np.frombuffer(index, np.uint32).reshape(-1, 2)

  assert index[-1, 0] == 0xFFFFFFFF
  assert index[-1, 1] == os.path.getsize(fn)

  return index, prefix



            

Reported by Bandit.

                index = np.frombuffer(index, np.uint32).reshape(-1, 2)

  assert index[-1, 0] == 0xFFFFFFFF
  assert index[-1, 1] == os.path.getsize(fn)

  return index, prefix


def cache_fn(func):

            

Reported by Bandit.

              class TestPlotJuggler(unittest.TestCase):

  def test_install(self):
    exit_code = os.system(os.path.join(BASEDIR, "tools/plotjuggler/install.sh"))
    self.assertEqual(exit_code, 0)

  def test_run(self):

    test_url = get_url("ffccc77938ddbc44|2021-01-04--16-55-41", 0)

            

Reported by Bandit.

                  # Launch PlotJuggler with the executable in the bin directory
    os.environ["PLOTJUGGLER_PATH"] = f'{os.path.join(BASEDIR, "tools/plotjuggler/bin/plotjuggler")}'
    p = subprocess.Popen(f'QT_QPA_PLATFORM=offscreen {os.path.join(BASEDIR, "tools/plotjuggler/juggle.py")} \
    "{test_url}"', stderr=subprocess.PIPE, shell=True,
    start_new_session=True)

    # Wait max 60 seconds for the "Done reading Rlog data" signal from the plugin
    output = "\n"
    with Timeout(120, error_msg=output):

            

Reported by Bandit.

              import time
import unittest

from common.basedir import BASEDIR
from common.timeout import Timeout
from selfdrive.test.openpilotci import get_url

class TestPlotJuggler(unittest.TestCase):


            

Reported by Pylint.

              import unittest

from common.basedir import BASEDIR
from common.timeout import Timeout
from selfdrive.test.openpilotci import get_url

class TestPlotJuggler(unittest.TestCase):

  def test_install(self):

            

Reported by Pylint.

              
from common.basedir import BASEDIR
from common.timeout import Timeout
from selfdrive.test.openpilotci import get_url

class TestPlotJuggler(unittest.TestCase):

  def test_install(self):
    exit_code = os.system(os.path.join(BASEDIR, "tools/plotjuggler/install.sh"))

            

Reported by Pylint.

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

from common.basedir import BASEDIR
from common.timeout import Timeout

            

Reported by Pylint.

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

from common.basedir import BASEDIR
from common.timeout import Timeout

            

Reported by Bandit.

              from common.timeout import Timeout
from selfdrive.test.openpilotci import get_url

class TestPlotJuggler(unittest.TestCase):

  def test_install(self):
    exit_code = os.system(os.path.join(BASEDIR, "tools/plotjuggler/install.sh"))
    self.assertEqual(exit_code, 0)


            

Reported by Pylint.

              
class TestPlotJuggler(unittest.TestCase):

  def test_install(self):
    exit_code = os.system(os.path.join(BASEDIR, "tools/plotjuggler/install.sh"))
    self.assertEqual(exit_code, 0)

  def test_run(self):


            

Reported by Pylint.

              
class TestPlotJuggler(unittest.TestCase):

  def test_install(self):
    exit_code = os.system(os.path.join(BASEDIR, "tools/plotjuggler/install.sh"))
    self.assertEqual(exit_code, 0)

  def test_run(self):


            

Reported by Pylint.

              from collections import defaultdict
from pathlib import Path

import cereal.messaging as messaging
from cereal import log
from cereal.services import service_list
from common.basedir import BASEDIR
from common.params import Params
from common.timeout import Timeout

            

Reported by Pylint.

              from collections import defaultdict
from pathlib import Path

import cereal.messaging as messaging
from cereal import log
from cereal.services import service_list
from common.basedir import BASEDIR
from common.params import Params
from common.timeout import Timeout

            

Reported by Pylint.

              from pathlib import Path

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

            

Reported by Pylint.

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

            

Reported by Pylint.

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

            

Reported by Pylint.

                def test_bootlog(self):
    # generate bootlog with fake launch log
    launch_log = ''.join([str(random.choice(string.printable)) for _ in range(100)])
    with open("/tmp/launch_log", "w") as f:
      f.write(launch_log)

    bootlog_path = self._gen_bootlog()
    lr = list(LogReader(str(bootlog_path)))


            

Reported by Bandit.

              import os
import random
import string
import subprocess
import time
import unittest
from collections import defaultdict
from pathlib import Path


            

Reported by Bandit.

              
  def _gen_bootlog(self):
    with Timeout(5):
      out = subprocess.check_output("./bootlog", cwd=os.path.join(BASEDIR, "selfdrive/loggerd"), encoding='utf-8')

    log_fn = self._get_log_fn(out)

    # check existence
    assert log_fn is not None

            

Reported by Bandit.

                  log_fn = self._get_log_fn(out)

    # check existence
    assert log_fn is not None

    return log_fn

  def _check_init_data(self, msgs):
    msg = msgs[0]

            

Reported by Bandit.

                  self.assertTrue(msgs[-1].sentinel.type == end_type)

  def test_init_data_values(self):
    os.environ["CLEAN"] = random.choice(["0", "1"])

    dongle  = ''.join(random.choice(string.printable) for n in range(random.randint(1, 100)))
    fake_params = [
      # param, initData field, value
      ("DongleId", "dongleId", dongle),

            

Reported by Bandit.

              #!/usr/bin/env python3
import numpy as np
import cereal.messaging as messaging
from PIL import ImageFont, ImageDraw, Image

font = ImageFont.truetype("arial", size=72)
def get_frame(idx):
  img = np.zeros((874, 1164, 3), np.uint8)
  img[100:400, 100:100+(idx % 10) * 100] = 255

            

Reported by Pylint.

                return img.tostring()

if __name__ == "__main__":
  from common.realtime import Ratekeeper
  rk = Ratekeeper(20)

  pm = messaging.PubMaster(['roadCameraState'])
  frm = [get_frame(x) for x in range(30)]
  idx = 0

            

Reported by Pylint.

              from PIL import ImageFont, ImageDraw, Image

font = ImageFont.truetype("arial", size=72)
def get_frame(idx):
  img = np.zeros((874, 1164, 3), np.uint8)
  img[100:400, 100:100+(idx % 10) * 100] = 255

  # big number
  im2 = Image.new("RGB", (200, 200))

            

Reported by Pylint.

              #!/usr/bin/env python3
import numpy as np
import cereal.messaging as messaging
from PIL import ImageFont, ImageDraw, Image

font = ImageFont.truetype("arial", size=72)
def get_frame(idx):
  img = np.zeros((874, 1164, 3), np.uint8)
  img[100:400, 100:100+(idx % 10) * 100] = 255

            

Reported by Pylint.

              from PIL import ImageFont, ImageDraw, Image

font = ImageFont.truetype("arial", size=72)
def get_frame(idx):
  img = np.zeros((874, 1164, 3), np.uint8)
  img[100:400, 100:100+(idx % 10) * 100] = 255

  # big number
  im2 = Image.new("RGB", (200, 200))

            

Reported by Pylint.

              
font = ImageFont.truetype("arial", size=72)
def get_frame(idx):
  img = np.zeros((874, 1164, 3), np.uint8)
  img[100:400, 100:100+(idx % 10) * 100] = 255

  # big number
  im2 = Image.new("RGB", (200, 200))
  draw = ImageDraw.Draw(im2)

            

Reported by Pylint.

              font = ImageFont.truetype("arial", size=72)
def get_frame(idx):
  img = np.zeros((874, 1164, 3), np.uint8)
  img[100:400, 100:100+(idx % 10) * 100] = 255

  # big number
  im2 = Image.new("RGB", (200, 200))
  draw = ImageDraw.Draw(im2)
  draw.text((10, 100), "%02d" % idx, font=font)

            

Reported by Pylint.

                img[100:400, 100:100+(idx % 10) * 100] = 255

  # big number
  im2 = Image.new("RGB", (200, 200))
  draw = ImageDraw.Draw(im2)
  draw.text((10, 100), "%02d" % idx, font=font)
  img[400:600, 400:600] = np.array(im2.getdata()).reshape((200, 200, 3))
  return img.tostring()


            

Reported by Pylint.

              
  # big number
  im2 = Image.new("RGB", (200, 200))
  draw = ImageDraw.Draw(im2)
  draw.text((10, 100), "%02d" % idx, font=font)
  img[400:600, 400:600] = np.array(im2.getdata()).reshape((200, 200, 3))
  return img.tostring()

if __name__ == "__main__":

            

Reported by Pylint.

                # big number
  im2 = Image.new("RGB", (200, 200))
  draw = ImageDraw.Draw(im2)
  draw.text((10, 100), "%02d" % idx, font=font)
  img[400:600, 400:600] = np.array(im2.getdata()).reshape((200, 200, 3))
  return img.tostring()

if __name__ == "__main__":
  from common.realtime import Ratekeeper

            

Reported by Pylint.