Master mpc rebased

common/colors.py

@@ -0,0 +1,36 @@
+class COLORS:
+  def __init__(self):
+    self.HEADER = '\033[95m'
+    self.OKBLUE = '\033[94m'
+    self.CBLUE = '\33[44m'
+    self.BOLD = '\033[1m'
+    self.CITALIC = '\33[3m'
+    self.OKGREEN = '\033[92m'
+    self.CWHITE = '\33[37m'
+    self.ENDC = '\033[0m' + self.CWHITE
+    self.UNDERLINE = '\033[4m'
+    self.PINK = '\33[38;5;207m'
+    self.PRETTY_YELLOW = self.BASE(220)
+
+    self.RED = '\033[91m'
+    self.PURPLE_BG = '\33[45m'
+    self.YELLOW = '\033[93m'
+    self.BLUE_GREEN = self.BASE(85)
+
+    self.FAIL = self.RED
+    # self.INFO = self.PURPLE_BG
+    self.INFO = self.BASE(207)
+    self.SUCCESS = self.OKGREEN
+    self.PROMPT = self.YELLOW
+    self.DBLUE = '\033[36m'
+    self.CYAN = self.BASE(39)
+    self.WARNING = '\033[33m'
+
+  def BASE(self, col):  # seems to support more colors
+    return '\33[38;5;{}m'.format(col)
+
+  def BASEBG(self, col):  # seems to support more colors
+    return '\33[48;5;{}m'.format(col)
+
+
+COLORS = COLORS()

common/op_params.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+import os
+import json
+from atomicwrites import atomic_write
+from common.colors import COLORS
+from common.basedir import BASEDIR
+from selfdrive.hardware import TICI
+try:
+  from common.realtime import sec_since_boot
+except ImportError:
+  import time
+  sec_since_boot = time.time
+
+warning = lambda msg: print('{}opParams WARNING: {}{}'.format(COLORS.WARNING, msg, COLORS.ENDC))
+error = lambda msg: print('{}opParams ERROR: {}{}'.format(COLORS.FAIL, msg, COLORS.ENDC))
+
+NUMBER = [float, int]  # value types
+NONE_OR_NUMBER = [type(None), float, int]
+
+BASEDIR = os.path.dirname(BASEDIR)
+PARAMS_DIR = os.path.join(BASEDIR, 'community', 'params')
+IMPORTED_PATH = os.path.join(PARAMS_DIR, '.imported')
+OLD_PARAMS_FILE = os.path.join(BASEDIR, 'op_params.json')
+
+
+class Param:
+  def __init__(self, default, allowed_types=[], description=None, *, static=False, live=False, hidden=False):  # pylint: disable=dangerous-default-value
+    self.default_value = default  # value first saved and returned if actual value isn't a valid type
+    if not isinstance(allowed_types, list):
+      allowed_types = [allowed_types]
+    self.allowed_types = allowed_types  # allowed python value types for opEdit
+    self.description = description  # description to be shown in opEdit
+    self.hidden = hidden  # hide this param to user in opEdit
+    self.live = live  # show under the live menu in opEdit
+    self.static = static  # use cached value, never reads to update
+    self._create_attrs()
+
+  def is_valid(self, value):
+    if not self.has_allowed_types:  # always valid if no allowed types, otherwise checks to make sure
+      return True
+    return type(value) in self.allowed_types
+
+  def _create_attrs(self):  # Create attributes and check Param is valid
+    self.has_allowed_types = isinstance(self.allowed_types, list) and len(self.allowed_types) > 0
+    self.has_description = self.description is not None
+    self.is_list = list in self.allowed_types
+    self.read_frequency = None if self.static else (1 if self.live else 10)  # how often to read param file (sec)
+    self.last_read = -1
+    if self.has_allowed_types:
+      assert type(self.default_value) in self.allowed_types, 'Default value type must be in specified allowed_types!'
+    if self.is_list:
+      self.allowed_types.remove(list)
+
+
+def _read_param(key):  # Returns None, False if a json error occurs
+  try:
+    with open(os.path.join(PARAMS_DIR, key), 'r') as f:
+      value = json.loads(f.read())
+    return value, True
+  except json.decoder.JSONDecodeError:
+    return None, False
+
+
+def _write_param(key, value):
+  param_path = os.path.join(PARAMS_DIR, key)
+  with atomic_write(param_path, overwrite=True) as f:
+    f.write(json.dumps(value))
+  os.chmod(param_path, 0o666)
+
+
+def _import_params():
+  if os.path.exists(OLD_PARAMS_FILE) and not os.path.exists(IMPORTED_PATH):  # if opParams needs to import from old params file
+    try:
+      with open(OLD_PARAMS_FILE, 'r') as f:
+        old_params = json.loads(f.read())
+      for key in old_params:
+        _write_param(key, old_params[key])
+      open(IMPORTED_PATH, 'w').close()
+    except:  # pylint: disable=bare-except
+      pass
+
+
+class opParams:
+  def __init__(self):
+    """
+      To add your own parameter to opParams in your fork, simply add a new entry in self.fork_params, instancing a new Param class with at minimum a default value.
+      The allowed_types and description args are not required but highly recommended to help users edit their parameters with opEdit safely.
+        - The description value will be shown to users when they use opEdit to change the value of the parameter.
+        - The allowed_types arg is used to restrict what kinds of values can be entered with opEdit so that users can't crash openpilot with unintended behavior.
+          (setting a param intended to be a number with a boolean, or viceversa for example)
+          Limiting the range of floats or integers is still recommended when `.get`ting the parameter.
+          When a None value is allowed, use `type(None)` instead of None, as opEdit checks the type against the values in the arg with `isinstance()`.
+        - If you want your param to update within a second, specify live=True. If your param is designed to be read once, specify static=True.
+          Specifying neither will have the param update every 10 seconds if constantly .get()
+          If the param is not static, call the .get() function on it in the update function of the file you're reading from to use live updating
+      Here's an example of a good fork_param entry:
+      self.fork_params = {'camera_offset': Param(0.06, allowed_types=NUMBER), live=True}  # NUMBER allows both floats and ints
+    """
+
+    self.fork_params = {'TR': Param(1.8, NUMBER, live=True)}
+
+    self._to_delete = []  # a list of unused params you want to delete from users' params file
+    self._to_reset = []  # a list of params you want reset to their default values
+    self._run_init()  # restores, reads, and updates params
+
+  def _run_init(self):  # does first time initializing of default params
+    # Two required parameters for opEdit
+    self.fork_params['username'] = Param(None, [type(None), str, bool], 'Your identifier provided with any crash logs sent to Sentry.\nHelps the developer reach out to you if anything goes wrong')
+    self.fork_params['op_edit_live_mode'] = Param(False, bool, 'This parameter controls which mode opEdit starts in', hidden=True)
+
+    self.params = self._load_params(can_import=True)
+    self._add_default_params()  # adds missing params and resets values with invalid types to self.params
+    self._delete_and_reset()  # removes old params
+
+  def get(self, key=None, *, force_update=False):  # key=None returns dict of all params
+    if key is None:
+      return self._get_all_params(to_update=force_update)
+    self._check_key_exists(key, 'get')
+    param_info = self.fork_params[key]
+    rate = param_info.read_frequency  # will be None if param is static, so check below
+
+    if (not param_info.static and sec_since_boot() - self.fork_params[key].last_read >= rate) or force_update:
+      value, success = _read_param(key)
+      self.fork_params[key].last_read = sec_since_boot()
+      if not success:  # in case of read error, use default and overwrite param
+        value = param_info.default_value
+        _write_param(key, value)
+      self.params[key] = value
+
+    if param_info.is_valid(value := self.params[key]):
+      return value  # all good, returning user's value
+    print(warning('User\'s value type is not valid! Returning default'))  # somehow... it should always be valid
+    return param_info.default_value  # return default value because user's value of key is not in allowed_types to avoid crashing openpilot
+
+  def put(self, key, value):
+    self._check_key_exists(key, 'put')
+    if not self.fork_params[key].is_valid(value):
+      raise Exception('opParams: Tried to put a value of invalid type!')
+    self.params.update({key: value})
+    _write_param(key, value)
+
+  def _load_params(self, can_import=False):
+    if not os.path.exists(PARAMS_DIR):
+      os.makedirs(PARAMS_DIR)
+      if can_import:
+        _import_params()  # just imports old params. below we read them in
+
+    params = {}
+    for key in os.listdir(PARAMS_DIR):  # PARAMS_DIR is guaranteed to exist
+      if key.startswith('.') or key not in self.fork_params:
+        continue
+      value, success = _read_param(key)
+      if not success:
+        value = self.fork_params[key].default_value
+        _write_param(key, value)
+      params[key] = value
+    return params
+
+  def _get_all_params(self, to_update=False):
+    if to_update:
+      self.params = self._load_params()
+    return {k: self.params[k] for k, p in self.fork_params.items() if k in self.params and not p.hidden}
+
+  def _check_key_exists(self, key, met):
+    if key not in self.fork_params:
+      raise Exception('opParams: Tried to {} an unknown parameter! Key not in fork_params: {}'.format(met, key))
+
+  def _add_default_params(self):
+    for key, param in self.fork_params.items():
+      if key not in self.params:
+        self.params[key] = param.default_value
+        _write_param(key, self.params[key])
+      elif not param.is_valid(self.params[key]):
+        print(warning('Value type of user\'s {} param not in allowed types, replacing with default!'.format(key)))
+        self.params[key] = param.default_value
+        _write_param(key, self.params[key])
+
+  def _delete_and_reset(self):
+    for key in list(self.params):
+      if key in self._to_delete:
+        del self.params[key]
+        os.remove(os.path.join(PARAMS_DIR, key))
+      elif key in self._to_reset and key in self.fork_params:
+        self.params[key] = self.fork_params[key].default_value
+        _write_param(key, self.params[key])