#

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

def generate_c_code_gnsf( model, opts ):

    casadi_version = CasadiMeta.version()

            

Reported by Pylint.

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

def generate_c_code_gnsf( model, opts ):

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

            

Reported by Pylint.

              
def generate_c_code_gnsf( 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)

    model_name = model.name

            

Reported by Pylint.

                  # -> evaluated SX GNSF functions with MX.
    u = model.u

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

    y = symbol("y", gnsf_ny, 1)

            

Reported by Pylint.

                  u = model.u

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

    y = symbol("y", gnsf_ny, 1)
    uhat = symbol("uhat", gnsf_nuhat, 1)

            

Reported by Pylint.

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

    y = symbol("y", gnsf_ny, 1)
    uhat = symbol("uhat", gnsf_nuhat, 1)
    p = model.p
    x1 = symbol("gnsf_x1", gnsf_nx1, 1)

            

Reported by Pylint.

              
    ## generate C code
    fun_name = model_name + '_gnsf_phi_fun'
    phi_fun_ = Function(fun_name, [y, uhat, p], [phi_fun(y, uhat, p)])
    phi_fun_.generate(fun_name, casadi_opts)

    fun_name = model_name + '_gnsf_phi_fun_jac_y'
    phi_fun_jac_y = model.phi_fun_jac_y
    phi_fun_jac_y_ = Function(fun_name, [y, uhat, p], phi_fun_jac_y(y, uhat, p))

            

Reported by Pylint.

              
    fun_name = model_name + '_gnsf_phi_fun_jac_y'
    phi_fun_jac_y = model.phi_fun_jac_y
    phi_fun_jac_y_ = Function(fun_name, [y, uhat, p], phi_fun_jac_y(y, uhat, p))
    phi_fun_jac_y_.generate(fun_name, casadi_opts)

    fun_name = model_name + '_gnsf_phi_jac_y_uhat'
    phi_jac_y_uhat = model.phi_jac_y_uhat
    phi_jac_y_uhat_ = Function(fun_name, [y, uhat, p], phi_jac_y_uhat(y, uhat, p))

            

Reported by Pylint.

              
    fun_name = model_name + '_gnsf_phi_jac_y_uhat'
    phi_jac_y_uhat = model.phi_jac_y_uhat
    phi_jac_y_uhat_ = Function(fun_name, [y, uhat, p], phi_jac_y_uhat(y, uhat, p))
    phi_jac_y_uhat_.generate(fun_name, casadi_opts)

    fun_name = model_name + '_gnsf_f_lo_fun_jac_x1k1uz'
    f_lo_fun_jac_x1k1uz = model.f_lo_fun_jac_x1k1uz
    f_lo_fun_jac_x1k1uz_eval = f_lo_fun_jac_x1k1uz(x1, x1dot, z1, u, p)

            

Reported by Pylint.

                  if not isinstance(f_lo_fun_jac_x1k1uz_eval, tuple) and is_empty(f_lo_fun_jac_x1k1uz_eval):
        f_lo_fun_jac_x1k1uz_eval = [empty_var]

    f_lo_fun_jac_x1k1uz_ = Function(fun_name, [x1, x1dot, z1, u, p],
                 f_lo_fun_jac_x1k1uz_eval)
    f_lo_fun_jac_x1k1uz_.generate(fun_name, casadi_opts)

    fun_name = model_name + '_gnsf_get_matrices_fun'
    get_matrices_fun_ = Function(fun_name, [dummy], get_matrices_fun(1))

            

Reported by Pylint.

              import time
import json

from common.basedir import BASEDIR
from common.params import Params
from selfdrive.controls.lib.alertmanager import set_offroad_alert

if __name__ == "__main__":
  params = Params()

            

Reported by Pylint.

              import json

from common.basedir import BASEDIR
from common.params import Params
from selfdrive.controls.lib.alertmanager import set_offroad_alert

if __name__ == "__main__":
  params = Params()


            

Reported by Pylint.

              
from common.basedir import BASEDIR
from common.params import Params
from selfdrive.controls.lib.alertmanager import set_offroad_alert

if __name__ == "__main__":
  params = Params()

  with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) as f:

            

Reported by Pylint.

              #!/usr/bin/env python3
import os
import sys
import time
import json

from common.basedir import BASEDIR
from common.params import Params
from selfdrive.controls.lib.alertmanager import set_offroad_alert

            

Reported by Pylint.

              from selfdrive.controls.lib.alertmanager import set_offroad_alert

if __name__ == "__main__":
  params = Params()

  with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) as f:
    offroad_alerts = json.load(f)

  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])

            

Reported by Pylint.

              if __name__ == "__main__":
  params = Params()

  with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) as f:
    offroad_alerts = json.load(f)

  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])
  while True:
    print("setting alert update")

            

Reported by Pylint.

                params = Params()

  with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) as f:
    offroad_alerts = json.load(f)

  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])
  while True:
    print("setting alert update")
    params.put_bool("UpdateAvailable", True)

            

Reported by Pylint.

                with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) as f:
    offroad_alerts = json.load(f)

  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])
  while True:
    print("setting alert update")
    params.put_bool("UpdateAvailable", True)
    r = open(os.path.join(BASEDIR, "RELEASES.md"), "r").read()
    r = r[:r.find('\n\n')]  # Slice latest release notes

            

Reported by Pylint.

                with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) as f:
    offroad_alerts = json.load(f)

  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])
  while True:
    print("setting alert update")
    params.put_bool("UpdateAvailable", True)
    r = open(os.path.join(BASEDIR, "RELEASES.md"), "r").read()
    r = r[:r.find('\n\n')]  # Slice latest release notes

            

Reported by Pylint.

                  offroad_alerts = json.load(f)

  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])
  while True:
    print("setting alert update")
    params.put_bool("UpdateAvailable", True)
    r = open(os.path.join(BASEDIR, "RELEASES.md"), "r").read()
    r = r[:r.find('\n\n')]  # Slice latest release notes
    params.put("ReleaseNotes", r + "\n")

            

Reported by Pylint.

              #!/usr/bin/env python
import numpy as np

# copied from common.transformations/camera.py
eon_focal_length = 910.0  # pixels
eon_dcam_focal_length = 860.0  # pixels

webcam_focal_length = -908.0/1.5  # pixels


            

Reported by Pylint.

              import numpy as np

# copied from common.transformations/camera.py
eon_focal_length = 910.0  # pixels
eon_dcam_focal_length = 860.0  # pixels

webcam_focal_length = -908.0/1.5  # pixels

eon_intrinsics = np.array([

            

Reported by Pylint.

              
# copied from common.transformations/camera.py
eon_focal_length = 910.0  # pixels
eon_dcam_focal_length = 860.0  # pixels

webcam_focal_length = -908.0/1.5  # pixels

eon_intrinsics = np.array([
  [eon_focal_length,   0.,   1164/2.],

            

Reported by Pylint.

              eon_focal_length = 910.0  # pixels
eon_dcam_focal_length = 860.0  # pixels

webcam_focal_length = -908.0/1.5  # pixels

eon_intrinsics = np.array([
  [eon_focal_length,   0.,   1164/2.],
  [  0.,  eon_focal_length,  874/2.],
  [  0.,    0.,     1.]])

            

Reported by Pylint.

                [  0.,    0.,     1.]])

if __name__ == "__main__":
  import cv2  # pylint: disable=import-error
  trans_webcam_to_eon_rear = np.dot(eon_intrinsics, np.linalg.inv(webcam_intrinsics))
  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))
  print("trans_webcam_to_eon_rear:\n", trans_webcam_to_eon_rear)
  print("trans_webcam_to_eon_front:\n", trans_webcam_to_eon_front)


            

Reported by Pylint.

              
if __name__ == "__main__":
  import cv2  # pylint: disable=import-error
  trans_webcam_to_eon_rear = np.dot(eon_intrinsics, np.linalg.inv(webcam_intrinsics))
  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))
  print("trans_webcam_to_eon_rear:\n", trans_webcam_to_eon_rear)
  print("trans_webcam_to_eon_front:\n", trans_webcam_to_eon_front)

  cap = cv2.VideoCapture(1)

            

Reported by Pylint.

              if __name__ == "__main__":
  import cv2  # pylint: disable=import-error
  trans_webcam_to_eon_rear = np.dot(eon_intrinsics, np.linalg.inv(webcam_intrinsics))
  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))
  print("trans_webcam_to_eon_rear:\n", trans_webcam_to_eon_rear)
  print("trans_webcam_to_eon_front:\n", trans_webcam_to_eon_front)

  cap = cv2.VideoCapture(1)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 853)

            

Reported by Pylint.

                import cv2  # pylint: disable=import-error
  trans_webcam_to_eon_rear = np.dot(eon_intrinsics, np.linalg.inv(webcam_intrinsics))
  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))
  print("trans_webcam_to_eon_rear:\n", trans_webcam_to_eon_rear)
  print("trans_webcam_to_eon_front:\n", trans_webcam_to_eon_front)

  cap = cv2.VideoCapture(1)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 853)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)

            

Reported by Pylint.

                trans_webcam_to_eon_rear = np.dot(eon_intrinsics, np.linalg.inv(webcam_intrinsics))
  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))
  print("trans_webcam_to_eon_rear:\n", trans_webcam_to_eon_rear)
  print("trans_webcam_to_eon_front:\n", trans_webcam_to_eon_front)

  cap = cv2.VideoCapture(1)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 853)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)


            

Reported by Pylint.

                print("trans_webcam_to_eon_rear:\n", trans_webcam_to_eon_rear)
  print("trans_webcam_to_eon_front:\n", trans_webcam_to_eon_front)

  cap = cv2.VideoCapture(1)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 853)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)

  while (True):
    ret, img = cap.read()

            

Reported by Pylint.

              #!/usr/bin/env python
import numpy as np

# copied from common.transformations/camera.py
eon_dcam_focal_length = 860.0  # pixels
webcam_focal_length = 908.0  # pixels

eon_dcam_intrinsics = np.array([
  [eon_dcam_focal_length,   0,   1152/2.],

            

Reported by Pylint.

              import numpy as np

# copied from common.transformations/camera.py
eon_dcam_focal_length = 860.0  # pixels
webcam_focal_length = 908.0  # pixels

eon_dcam_intrinsics = np.array([
  [eon_dcam_focal_length,   0,   1152/2.],
  [  0,  eon_dcam_focal_length,  864/2.],

            

Reported by Pylint.

              
# copied from common.transformations/camera.py
eon_dcam_focal_length = 860.0  # pixels
webcam_focal_length = 908.0  # pixels

eon_dcam_intrinsics = np.array([
  [eon_dcam_focal_length,   0,   1152/2.],
  [  0,  eon_dcam_focal_length,  864/2.],
  [  0,    0,     1]])

            

Reported by Pylint.

                [  0.,  webcam_focal_length,  720/2.],
  [  0.,    0.,     1.]])

cam_id = 2

if __name__ == "__main__":
  import cv2  # pylint: disable=import-error

  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))

            

Reported by Pylint.

              cam_id = 2

if __name__ == "__main__":
  import cv2  # pylint: disable=import-error

  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))

  cap = cv2.VideoCapture(cam_id)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)

            

Reported by Pylint.

              if __name__ == "__main__":
  import cv2  # pylint: disable=import-error

  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))

  cap = cv2.VideoCapture(cam_id)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)


            

Reported by Pylint.

              
  trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))

  cap = cv2.VideoCapture(cam_id)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

  while (True):
    ret, img = cap.read()

            

Reported by Pylint.

                trans_webcam_to_eon_front = np.dot(eon_dcam_intrinsics, np.linalg.inv(webcam_intrinsics))

  cap = cv2.VideoCapture(cam_id)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

  while (True):
    ret, img = cap.read()
    if ret:

            

Reported by Pylint.

              
  cap = cv2.VideoCapture(cam_id)
  cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

  while (True):
    ret, img = cap.read()
    if ret:
      img = cv2.warpPerspective(img, trans_webcam_to_eon_front, (1152, 864), borderMode=cv2.BORDER_CONSTANT, borderValue=0)

            

Reported by Pylint.

                cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
  cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

  while (True):
    ret, img = cap.read()
    if ret:
      img = cv2.warpPerspective(img, trans_webcam_to_eon_front, (1152, 864), borderMode=cv2.BORDER_CONSTANT, borderValue=0)
      img = img[:, -864//2:, :]
      cv2.imshow('preview', img)

            

Reported by Pylint.

              #!/usr/bin/env python3
# type: ignore
import cereal.messaging as messaging

all_sockets = ['roadCameraState', 'driverCameraState', 'wideRoadCameraState']
prev_id = [None,None,None]
this_id = [None,None,None]
dt = [None,None,None]
num_skipped = [0,0,0]

            

Reported by Pylint.

              #!/usr/bin/env python3
# type: ignore
import cereal.messaging as messaging

all_sockets = ['roadCameraState', 'driverCameraState', 'wideRoadCameraState']
prev_id = [None,None,None]
this_id = [None,None,None]
dt = [None,None,None]
num_skipped = [0,0,0]

            

Reported by Pylint.

              num_skipped = [0,0,0]

if __name__ == "__main__":
  sm = messaging.SubMaster(all_sockets)
  while True:
    sm.update()

    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:

            

Reported by Pylint.

              
if __name__ == "__main__":
  sm = messaging.SubMaster(all_sockets)
  while True:
    sm.update()

    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue

            

Reported by Pylint.

              if __name__ == "__main__":
  sm = messaging.SubMaster(all_sockets)
  while True:
    sm.update()

    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue
      this_id[i] = sm[all_sockets[i]].frameId

            

Reported by Pylint.

                while True:
    sm.update()

    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue
      this_id[i] = sm[all_sockets[i]].frameId
      if prev_id[i] is None:
        prev_id[i] = this_id[i]

            

Reported by Pylint.

                while True:
    sm.update()

    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue
      this_id[i] = sm[all_sockets[i]].frameId
      if prev_id[i] is None:
        prev_id[i] = this_id[i]

            

Reported by Pylint.

                  sm.update()

    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue
      this_id[i] = sm[all_sockets[i]].frameId
      if prev_id[i] is None:
        prev_id[i] = this_id[i]
        continue

            

Reported by Pylint.

              
    for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue
      this_id[i] = sm[all_sockets[i]].frameId
      if prev_id[i] is None:
        prev_id[i] = this_id[i]
        continue
      dt[i] = this_id[i] - prev_id[i]

            

Reported by Pylint.

                  for i in range(len(all_sockets)):
      if not sm.updated[all_sockets[i]]:
        continue
      this_id[i] = sm[all_sockets[i]].frameId
      if prev_id[i] is None:
        prev_id[i] = this_id[i]
        continue
      dt[i] = this_id[i] - prev_id[i]
      if dt[i] != 1:

            

Reported by Pylint.

              
  def test_params_put_and_get(self):
    self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("DongleId") == b"cb38263377b873ee"

  def test_persist_params_put_and_get(self):
    p = Params(persistent_params=True)
    p.put("DongleId", "cb38263377b873ee")
    assert p.get("DongleId") == b"cb38263377b873ee"

            

Reported by Bandit.

                def test_persist_params_put_and_get(self):
    p = Params(persistent_params=True)
    p.put("DongleId", "cb38263377b873ee")
    assert p.get("DongleId") == b"cb38263377b873ee"

  def test_params_non_ascii(self):
    st = b"\xe1\x90\xff"
    self.params.put("CarParams", st)
    assert self.params.get("CarParams") == st

            

Reported by Bandit.

                def test_params_non_ascii(self):
    st = b"\xe1\x90\xff"
    self.params.put("CarParams", st)
    assert self.params.get("CarParams") == st

  def test_params_get_cleared_panda_disconnect(self):
    self.params.put("CarParams", "test")
    self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("CarParams") == b"test"

            

Reported by Bandit.

                def test_params_get_cleared_panda_disconnect(self):
    self.params.put("CarParams", "test")
    self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("CarParams") == b"test"
    self.params.clear_all(ParamKeyType.CLEAR_ON_PANDA_DISCONNECT)
    assert self.params.get("CarParams") is None
    assert self.params.get("DongleId") is not None

  def test_params_get_cleared_manager_start(self):

            

Reported by Bandit.

                  self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("CarParams") == b"test"
    self.params.clear_all(ParamKeyType.CLEAR_ON_PANDA_DISCONNECT)
    assert self.params.get("CarParams") is None
    assert self.params.get("DongleId") is not None

  def test_params_get_cleared_manager_start(self):
    self.params.put("CarParams", "test")
    self.params.put("DongleId", "cb38263377b873ee")

            

Reported by Bandit.

                  assert self.params.get("CarParams") == b"test"
    self.params.clear_all(ParamKeyType.CLEAR_ON_PANDA_DISCONNECT)
    assert self.params.get("CarParams") is None
    assert self.params.get("DongleId") is not None

  def test_params_get_cleared_manager_start(self):
    self.params.put("CarParams", "test")
    self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("CarParams") == b"test"

            

Reported by Bandit.

                def test_params_get_cleared_manager_start(self):
    self.params.put("CarParams", "test")
    self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("CarParams") == b"test"
    self.params.clear_all(ParamKeyType.CLEAR_ON_MANAGER_START)
    assert self.params.get("CarParams") is None
    assert self.params.get("DongleId") is not None

  def test_params_two_things(self):

            

Reported by Bandit.

                  self.params.put("DongleId", "cb38263377b873ee")
    assert self.params.get("CarParams") == b"test"
    self.params.clear_all(ParamKeyType.CLEAR_ON_MANAGER_START)
    assert self.params.get("CarParams") is None
    assert self.params.get("DongleId") is not None

  def test_params_two_things(self):
    self.params.put("DongleId", "bob")
    self.params.put("AthenadPid", "123")

            

Reported by Bandit.

                  assert self.params.get("CarParams") == b"test"
    self.params.clear_all(ParamKeyType.CLEAR_ON_MANAGER_START)
    assert self.params.get("CarParams") is None
    assert self.params.get("DongleId") is not None

  def test_params_two_things(self):
    self.params.put("DongleId", "bob")
    self.params.put("AthenadPid", "123")
    assert self.params.get("DongleId") == b"bob"

            

Reported by Bandit.

                def test_params_two_things(self):
    self.params.put("DongleId", "bob")
    self.params.put("AthenadPid", "123")
    assert self.params.get("DongleId") == b"bob"
    assert self.params.get("AthenadPid") == b"123"

  def test_params_get_block(self):
    def _delayed_writer():
      time.sleep(0.1)

            

Reported by Bandit.

              import json
import os
import signal
import subprocess
import time

import requests

from selfdrive.manager.process_config import managed_processes

            

Reported by Bandit.

              
  params = Params()
  manage_athenad_pid = params.get("AthenadPid")
  assert manage_athenad_pid is not None
  try:
    os.kill(int(manage_athenad_pid), 0)
    # process is running
  except OSError:
    assert False, "manage_athenad is dead"

            

Reported by Bandit.

                  os.kill(int(manage_athenad_pid), 0)
    # process is running
  except OSError:
    assert False, "manage_athenad is dead"

  def expect_athena_starts(timeout=30):
    now = time.time()
    athenad_pid = None
    while athenad_pid is None:

            

Reported by Bandit.

                  athenad_pid = None
    while athenad_pid is None:
      try:
        athenad_pid = subprocess.check_output(["pgrep", "-P", manage_athenad_pid], encoding="utf-8").strip()
        return athenad_pid
      except subprocess.CalledProcessError:
        if time.time() - now > timeout:
          assert False, f"Athena did not start within {timeout} seconds"
        time.sleep(0.5)

            

Reported by Bandit.

                  athenad_pid = None
    while athenad_pid is None:
      try:
        athenad_pid = subprocess.check_output(["pgrep", "-P", manage_athenad_pid], encoding="utf-8").strip()
        return athenad_pid
      except subprocess.CalledProcessError:
        if time.time() - now > timeout:
          assert False, f"Athena did not start within {timeout} seconds"
        time.sleep(0.5)

            

Reported by Bandit.

                      return athenad_pid
      except subprocess.CalledProcessError:
        if time.time() - now > timeout:
          assert False, f"Athena did not start within {timeout} seconds"
        time.sleep(0.5)

  def athena_post(payload, max_retries=5, wait=5):
    tries = 0
    while 1:

            

Reported by Bandit.

                    "id": 0,
      "jsonrpc": "2.0"
    }, max_retries=12, wait=5)
    assert resp.get('result') == "hello", f'Athena failed to register ({resp})'

    last_pingtime = params.get("LastAthenaPingTime", encoding='utf8')
    assert last_pingtime, last_pingtime
    assert ((int(last_pingtime)/1e9) - test_t0) < (sec_since_boot() - test_t0)


            

Reported by Bandit.

                  assert resp.get('result') == "hello", f'Athena failed to register ({resp})'

    last_pingtime = params.get("LastAthenaPingTime", encoding='utf8')
    assert last_pingtime, last_pingtime
    assert ((int(last_pingtime)/1e9) - test_t0) < (sec_since_boot() - test_t0)

  try:
    athenad_pid = expect_athena_starts()
    # kill athenad and ensure it is restarted (check_output will throw if it is not)

            

Reported by Bandit.

              
    last_pingtime = params.get("LastAthenaPingTime", encoding='utf8')
    assert last_pingtime, last_pingtime
    assert ((int(last_pingtime)/1e9) - test_t0) < (sec_since_boot() - test_t0)

  try:
    athenad_pid = expect_athena_starts()
    # kill athenad and ensure it is restarted (check_output will throw if it is not)
    os.kill(int(athenad_pid), signal.SIGINT)

            

Reported by Bandit.

                    "id": 0,
      "jsonrpc": "2.0"
    })
    assert resp.get('result'), resp
    assert 'sim_id' in resp['result'], resp['result']

    print("ATHENA: takeSnapshot")
    resp = athena_post({
      "method": "takeSnapshot",

            

Reported by Bandit.

              import subprocess
from typing import List, Union

from cereal import log
from selfdrive.hardware.base import HardwareBase, ThermalConfig

NetworkType = log.DeviceState.NetworkType
NetworkStrength = log.DeviceState.NetworkStrength


            

Reported by Pylint.

              import re
import serial
import struct
import subprocess
from typing import List, Union

from cereal import log
from selfdrive.hardware.base import HardwareBase, ThermalConfig


            

Reported by Bandit.

              
def service_call(call: List[str]) -> Union[bytes, None]:
  try:
    ret = subprocess.check_output(["service", "call", *call], encoding='utf8').strip()
    if 'Parcel' not in ret:
      return None
    return parse_service_call_bytes(ret)
  except subprocess.CalledProcessError:
    return None

            

Reported by Bandit.

              
def service_call(call: List[str]) -> Union[bytes, None]:
  try:
    ret = subprocess.check_output(["service", "call", *call], encoding='utf8').strip()
    if 'Parcel' not in ret:
      return None
    return parse_service_call_bytes(ret)
  except subprocess.CalledProcessError:
    return None

            

Reported by Bandit.

              
def getprop(key: str) -> Union[str, None]:
  try:
    return subprocess.check_output(["getprop", key], encoding='utf8').strip()
  except subprocess.CalledProcessError:
    return None


class Android(HardwareBase):

            

Reported by Bandit.

              
def getprop(key: str) -> Union[str, None]:
  try:
    return subprocess.check_output(["getprop", key], encoding='utf8').strip()
  except subprocess.CalledProcessError:
    return None


class Android(HardwareBase):

            

Reported by Bandit.

                  else:
      reason_args = ["s16", reason]

    subprocess.check_output([
      "service", "call", "power", "16",  # IPowerManager.reboot
      "i32", "0",  # no confirmation,
      *reason_args,
      "i32", "1"  # wait
    ])

            

Reported by Bandit.

                  else:
      reason_args = ["s16", reason]

    subprocess.check_output([
      "service", "call", "power", "16",  # IPowerManager.reboot
      "i32", "0",  # no confirmation,
      *reason_args,
      "i32", "1"  # wait
    ])

            

Reported by Bandit.

              
      rsrp = msg.extra.split("$QCRSRP: ")[1].split("\r")[0].split(",")
      msg.channel = int(rsrp[1])
    except Exception:
      pass

    return msg

  def get_network_type(self):

            

Reported by Bandit.

                  if network_type == NetworkType.none:
      return network_strength
    if network_type == NetworkType.wifi:
      out = subprocess.check_output('dumpsys connectivity', shell=True).decode('utf-8')
      network_strength = NetworkStrength.unknown
      for line in out.split('\n'):
        signal_str = "SignalStrength: "
        if signal_str in line:
          lvl_idx_start = line.find(signal_str) + len(signal_str)

            

Reported by Bandit.

              
import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

jwt = get_token()

route = sys.argv[1]

            

Reported by Pylint.

              import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

jwt = get_token()

route = sys.argv[1]
segment = int(sys.argv[2])

            

Reported by Pylint.

              #!/usr/bin/env python3
import sys

if len(sys.argv) < 4:
  print("%s <route> <segment> <frame number>" % sys.argv[0])
  print('example: ./fetch_image_from_route.py "02c45f73a2e5c6e9|2020-06-01--18-03-08" 3 500')
  exit(0)

import requests

            

Reported by Pylint.

              import sys

if len(sys.argv) < 4:
  print("%s <route> <segment> <frame number>" % sys.argv[0])
  print('example: ./fetch_image_from_route.py "02c45f73a2e5c6e9|2020-06-01--18-03-08" 3 500')
  exit(0)

import requests
from PIL import Image

            

Reported by Pylint.

              
if len(sys.argv) < 4:
  print("%s <route> <segment> <frame number>" % sys.argv[0])
  print('example: ./fetch_image_from_route.py "02c45f73a2e5c6e9|2020-06-01--18-03-08" 3 500')
  exit(0)

import requests
from PIL import Image
from tools.lib.auth_config import get_token

            

Reported by Pylint.

              if len(sys.argv) < 4:
  print("%s <route> <segment> <frame number>" % sys.argv[0])
  print('example: ./fetch_image_from_route.py "02c45f73a2e5c6e9|2020-06-01--18-03-08" 3 500')
  exit(0)

import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

            

Reported by Pylint.

              if len(sys.argv) < 4:
  print("%s <route> <segment> <frame number>" % sys.argv[0])
  print('example: ./fetch_image_from_route.py "02c45f73a2e5c6e9|2020-06-01--18-03-08" 3 500')
  exit(0)

import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

            

Reported by Pylint.

                print('example: ./fetch_image_from_route.py "02c45f73a2e5c6e9|2020-06-01--18-03-08" 3 500')
  exit(0)

import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

jwt = get_token()

            

Reported by Pylint.

                exit(0)

import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

jwt = get_token()


            

Reported by Pylint.

              
import requests
from PIL import Image
from tools.lib.auth_config import get_token
from tools.lib.framereader import FrameReader

jwt = get_token()

route = sys.argv[1]

            

Reported by Pylint.

              #!/usr/bin/env python3

import requests
from common.params import Params
import sys


if __name__ == "__main__":
  if len(sys.argv) < 2:

            

Reported by Pylint.

              #!/usr/bin/env python3

import requests
from common.params import Params
import sys


if __name__ == "__main__":
  if len(sys.argv) < 2:

            

Reported by Pylint.

              
import requests
from common.params import Params
import sys


if __name__ == "__main__":
  if len(sys.argv) < 2:
    print("%s <github username>" % sys.argv[0])

            

Reported by Pylint.

              

if __name__ == "__main__":
  if len(sys.argv) < 2:
    print("%s <github username>" % sys.argv[0])
    exit(1)

  username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")

            

Reported by Pylint.

              
if __name__ == "__main__":
  if len(sys.argv) < 2:
    print("%s <github username>" % sys.argv[0])
    exit(1)

  username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")


            

Reported by Pylint.

              if __name__ == "__main__":
  if len(sys.argv) < 2:
    print("%s <github username>" % sys.argv[0])
    exit(1)

  username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")

  if keys.status_code == 200:

            

Reported by Pylint.

              if __name__ == "__main__":
  if len(sys.argv) < 2:
    print("%s <github username>" % sys.argv[0])
    exit(1)

  username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")

  if keys.status_code == 200:

            

Reported by Pylint.

                  print("%s <github username>" % sys.argv[0])
    exit(1)

  username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")

  if keys.status_code == 200:
    Params().put("GithubSshKeys", keys.text)
    Params().put("GithubUsername", username)

            

Reported by Pylint.

                  exit(1)

  username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")

  if keys.status_code == 200:
    Params().put("GithubSshKeys", keys.text)
    Params().put("GithubUsername", username)
    print("Setup ssh keys successfully")

            

Reported by Pylint.

                username = sys.argv[1]
  keys = requests.get(f"https://github.com/{username}.keys")

  if keys.status_code == 200:
    Params().put("GithubSshKeys", keys.text)
    Params().put("GithubUsername", username)
    print("Setup ssh keys successfully")
  else:
    print("Error getting public keys from github")

            

Reported by Pylint.