| """Renderer profile plus live pure-function checks of native controller limits.""" |
| import ast,hashlib,importlib.util,json,sys,types |
| from pathlib import Path |
| HERE=Path(__file__).resolve().parent |
| PROJECT=HERE.parents[1] |
| profile=PROJECT/'runtime/tgcc_revision_20261027/renderer_profile_20261028.py' |
| spec=importlib.util.spec_from_file_location('renderer_profile',profile) |
| renderer=importlib.util.module_from_spec(spec);spec.loader.exec_module(renderer) |
|
|
|
|
| def attest(robot,output): |
| import inspect,numpy as np |
| from omnigibson.controllers.controller_view import ControllerView |
| from omnigibson.utils.backend_utils import _compute_backend as cb |
| def array(value):return cb.to_torch(value).detach().cpu().numpy() |
| records={} |
| for name in ('base','trunk','arm_left','arm_right','gripper_left','gripper_right'): |
| c=robot.controllers[name];member=0 |
| if isinstance(c,tuple):member=c[1];c=ControllerView._controller_groups[c[0]] |
| path=Path(inspect.getfile(type(c)));record=dict(controller=type(c).__name__,source_file=str(path),source_sha256=hashlib.sha256(path.read_bytes()).hexdigest()) |
| if name in ('trunk','arm_left','arm_right'): |
| assert c._motor_type=='position' and not c.use_delta_commands and not getattr(c,'_use_impedances',False) |
| lo,hi=[v[c.dof_idx] for v in c._control_limits[c.control_type]] |
| assert np.all(np.isfinite(array(lo))) and np.all(array(lo)<array(hi)) |
| for command in (lo-1.,hi+1.): |
| |
| |
| target=c._update_goal(member,command)['target'] |
| assert np.all(array(target)>=array(lo)-1e-6) and np.all(array(target)<=array(hi)+1e-6) |
| record.update(joint_target_clipping_verified=True,lower=array(lo).tolist(),upper=array(hi).tolist()) |
| else: |
| assert c._command_input_limits is not None and c._command_output_limits is not None |
| lo,hi=c._command_output_limits |
| for sign in (-1.,1.): |
| value=c._preprocess_command(cb.ones(c.command_dim)*sign*1000.) |
| assert np.all(array(value)>=array(lo)-1e-6) and np.all(array(value)<=array(hi)+1e-6) |
| record.update(native_input_clipping_verified=True,output_lower=array(lo).tolist(),output_upper=array(hi).tolist()) |
| records[name]=record |
| receipt=dict(schema='tgcc.native_controller_limits.v1',status='passed',controllers=records, |
| physics_steps_used=0,commands_dispatched_by_check=0,scope='live controller input and absolute joint target clipping; not collision or task success verification', |
| client_extension_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest()) |
| (output/'native_limit_attestation.json').write_text(json.dumps(receipt,indent=2)+'\n') |
| return receipt |
|
|
|
|
| def main(): |
| client=Path(sys.argv[1]).resolve();args=sys.argv[2:];output=Path(args[args.index('--output')+1]);output.mkdir(parents=True,exist_ok=True) |
| raw=client.read_bytes();tree=renderer.renderer_tree(raw) |
| |
| |
| budget=next(n for n in ast.walk(tree) if isinstance(n,ast.Assign) and any( |
| isinstance(t,ast.Attribute) and t.attr=='texture_memory_budget' for t in n.targets)) |
| if not isinstance(budget.value,ast.Constant) or budget.value.value!=.10:raise ValueError('Unexpected renderer profile') |
| budget.value=ast.copy_location(ast.Constant(.05),budget.value) |
| fn=next(n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=='audit_controller_contract');fn.name='_original_audit_controller_contract' |
| index=tree.body.index(fn) |
| tree.body[index+1:index+1]=ast.parse('audit_controller_contract = audited_contract(_original_audit_controller_contract)').body |
| capture=next(n for n in ast.walk(tree) if isinstance(n,ast.FunctionDef) and n.name=='capture_payload') |
| ret=next(i for i,n in enumerate(capture.body) if isinstance(n,ast.Return)) |
| capture.body[ret:ret]=ast.parse("payload['native_limit_attestation']=native_attestation[0]").body |
| receipt=dict(schema='tgcc.explicit_native_limits_client.v1',original_client_sha256=hashlib.sha256(raw).hexdigest(), |
| launcher_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),renderer_extension_sha256=hashlib.sha256(profile.read_bytes()).hexdigest(), |
| effective_ast_sha256=hashlib.sha256(ast.dump(tree,include_attributes=False).encode()).hexdigest(), |
| texture_budget=.05,native_controllers_unchanged=True,record_video=False) |
| (output/'client_extension.json').write_text(json.dumps(receipt,indent=2)+'\n') |
| attestation=[] |
| def audited_contract(original): |
| def wrapped(robot,*a,**kw): |
| value=original(robot,*a,**kw);attestation[:]=[attest(robot,output)];return value |
| return wrapped |
| sys.argv=[str(client)]+args;sys.path.insert(0,str(client.parent));module=types.ModuleType('__main__');module.__file__=str(client) |
| module.audited_contract=audited_contract;module.native_attestation=attestation;sys.modules['__main__']=module |
| exec(compile(ast.fix_missing_locations(tree),str(client),'exec'),module.__dict__) |
|
|
|
|
| if __name__=='__main__':main() |
|
|
