45 lines
3.2 KiB
Python
45 lines
3.2 KiB
Python
"""Short-window RoboDopamine inference and advisory state conversion."""
|
|
import math,time
|
|
from .progress import ProgressMonitor
|
|
from .backends import InferenceError
|
|
from .service import BrainService
|
|
from robot_bt_coordinator.plan import strict_json,canonical
|
|
|
|
class DenseFeedbackService(BrainService):
|
|
def __init__(self,backend,record_dir):super().__init__(backend,record_dir);self.monitors={}
|
|
def evaluate(self,goal,observations,now,cancel=None):
|
|
raw='';started=time.monotonic()
|
|
try:
|
|
key=(goal['run_id'],goal['subtask_id'])
|
|
if not all(isinstance(x,str) and x for x in key) or not isinstance(goal['task_description'],str) or not goal['task_description'].strip():raise InferenceError('INVALID_INPUT')
|
|
if not isinstance(observations,list) or not 1<=len(observations)<=32:raise InferenceError('INVALID_WINDOW')
|
|
previous=None
|
|
for frame in observations:
|
|
at=frame['stamp']
|
|
if type(at) not in (int,float) or not math.isfinite(at) or not goal['capture_after']<=at<=now or now-at>10 or (previous is not None and at<=previous):raise InferenceError('INVALID_WINDOW')
|
|
if not isinstance(frame['views'],dict) or not 1<=len(frame['views'])<=3 or any(not isinstance(k,str) or not k or not isinstance(v,str) or not v for k,v in frame['views'].items()):raise InferenceError('INVALID_WINDOW')
|
|
previous=at
|
|
if now-previous>2:raise InferenceError('STALE_WINDOW')
|
|
for old in list(self.monitors):
|
|
last=self.monitors[old].last_stamp
|
|
if old!=key and (last is None or now-last>300 or now<last):del self.monitors[old]
|
|
if key not in self.monitors:
|
|
if len(self.monitors)>=128:raise InferenceError('MONITOR_CAPACITY')
|
|
self.monitors[key]=ProgressMonitor(*key,goal['capture_after'])
|
|
elif self.monitors[key].capture_after!=goal['capture_after']:raise InferenceError('CONTEXT_MISMATCH')
|
|
snapshots=[]
|
|
for frame in observations:
|
|
views={camera:self._snapshot({'image_path':path})['image_path'] for camera,path in frame['views'].items()}
|
|
snapshots.append(dict(stamp=frame['stamp'],views=views))
|
|
observations=snapshots
|
|
request=dict(goal,frames=observations)
|
|
raw=self._call('dense_feedback',request,'Output JSON {progress: number in [0,1],hop: raw model value}; evaluate only the given subtask, no success verdict. '+canonical(goal),cancel)
|
|
data=strict_json(raw)
|
|
if not isinstance(data,dict) or set(data)!={'progress','hop'}:raise InferenceError('PARSE_ERROR')
|
|
result=self.monitors[key].update(dict(data,run_id=key[0],subtask_id=key[1],sequence=goal['sequence'],stamp=previous),now+time.monotonic()-started)
|
|
except (InferenceError,ValueError,TypeError,KeyError) as ex:result=dict(state='UNKNOWN',completion_authority=False,error_code=getattr(ex,'code','INVALID_INPUT'))
|
|
result['record_ref']=self._record('dense_feedback',goal,raw,result,observations);return result
|
|
def release(self,run_id):
|
|
for key in list(self.monitors):
|
|
if key[0]==run_id:del self.monitors[key]
|