Collect actual passive startup flags for observer policy

This commit is contained in:
firestar5683
2026-09-19 22:05:15 -07:00
parent 0f35039b27
commit 7e2df0f81c
2 changed files with 25 additions and 2 deletions
+13 -1
View File
@@ -29,6 +29,18 @@ def local_model_placement(metadata, active, loading):
return metadata.get("uses_external_gpu") is False and (active is False or active in (b"0", "0")) and (loading is False or loading in (b"0", "0"))
def passive_startup(params):
values={}
for key,param in (('controls_ready','ControlsReady'),('firmware_query_done','FirmwareQueryDone'),('obd_multiplexing_enabled','ObdMultiplexingEnabled')):
value=params.get(param)
if type(value) is bool:values[key]=value
elif value in (b'0','0'):values[key]=False
elif value in (b'1','1'):values[key]=True
elif value is None and param=='ControlsReady':values[key]=False
else:values[key]=None
return values
def evaluate(snapshot, record, now):
ages=snapshot.get('ages',{});valid=snapshot.get('valid',{})
# These are stale-input guards, not performance/coexistence acceptance limits.
@@ -139,7 +151,7 @@ class LiveHealth:
fields=(self.proc/str(pid)/'stat').read_text().rsplit(') ',1)[1].split()
owner=owner and fields[0]!='Z' and int(fields[19])/os.sysconf('SC_CLK_TCK')<=link['monotonic']<=now
except (OSError,ValueError,KeyError,IndexError,TypeError):owner=False
return {'monotonic':now,'car_identity':car_info,'car_id':identity(car_info),'baseline_id':identity({'git':build,'model':model,'version':version}),
return {'monotonic':now,'passive_startup':passive_startup(self.params),'car_identity':car_info,'car_id':identity(car_info),'baseline_id':identity({'git':build,'model':model,'version':version}),
'git_commit':build,'driving_model':model,'driving_model_version':version,'driving_model_local':local_model_placement(load_model_artifact_metadata(model),self.params.get('UsbGpuActive'),self.params.get('UsbGpuLoading')),
'chestnut_present':chestnut_firmware_ready(),'model_geometry_valid':len(m.get('position',{}).get('t',[]))==33,'ages':ages,'valid':valid,'car':payload['carState'],'device':d,'pandas':pandas,'events':events,
'control':payload['carControl'],'selfdrive_state':payload['selfdriveState'],'calibration':payload['liveCalibration'].get('calStatus'),'processes':processes,'metrics':metrics,'link':link,'link_owner_verified':bool(owner)}
+12 -1
View File
@@ -3,7 +3,7 @@ import json
from pathlib import Path
import tempfile
import unittest
from live_health import evaluate,LiveHealth,SERVICES,METRICS,local_model_placement
from live_health import evaluate,LiveHealth,SERVICES,METRICS,local_model_placement,passive_startup
def sample(now=100.):
@@ -21,6 +21,17 @@ def authorization():
class Tests(unittest.TestCase):
def test_passive_startup_reads_actual_flags_without_writes(self):
class Params:
def __init__(self,values):self.values=values
def get(self,key):return self.values.get(key)
p=Params({'ControlsReady':False,'FirmwareQueryDone':True,'ObdMultiplexingEnabled':False})
self.assertEqual(passive_startup(p),{'controls_ready':False,'firmware_query_done':True,'obd_multiplexing_enabled':False})
self.assertEqual(passive_startup(Params({})),{'controls_ready':False,'firmware_query_done':None,'obd_multiplexing_enabled':None})
p.values={'ControlsReady':b'1','FirmwareQueryDone':b'1','ObdMultiplexingEnabled':b'0'}
self.assertEqual(passive_startup(p),{'controls_ready':True,'firmware_query_done':True,'obd_multiplexing_enabled':False})
p.values={'ControlsReady':'invalid'}
self.assertIsNone(passive_startup(p)['controls_ready'])
def test_passive_mode_requires_pinned_nonactuating_evidence(self):
d=sample();d.update(live_mode='passive-observer-v1',car_identity={'passive':True,'notCar':False,'dashcamOnly':True,'safety':[{'model':'noOutput','param':0}]},control={'enabled':False,'latActive':False,'longActive':False},selfdrive_state={'enabled':False,'active':False})
d['pandas']=[{'safetyModel':'noOutput','controlsAllowed':False,'faults':[],'safetyRxChecksInvalid':False}]