File size: 2,810 Bytes
547ac30 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | """Standalone prediction from PUBLIC robot+power-supply description, target program
and PREVIOUS measured power history. No target label used during prediction.
Usage: python src/predict_welding.py --input examples/rsw_v55_input.json
"""
import argparse,json,sys
from pathlib import Path
import numpy as np,torch
ROOT=Path(__file__).resolve().parents[1]
sys.path[:0]=[str(ROOT/'runtime'),str(ROOT/'runtime')]
from process_features import feature,nominal,calibrations
from robot_cell import totals_common,COMPONENTS
from welding_specialists import WeldingSet,WeldingMoE55,predict
def calculate(item,checkpoint=None):
process=item['process'];src=item['machine']['power_supply'];cfg=item['machine']['robot'];program=item['program']
m=item['machine']
hist=np.r_[m['previous_power_probes_W'],m['previous_robot_power_probes_W'],
m['previous_motions'],m['prior_external_idle_log_ratios']].astype(np.float32)
c,s=nominal(cfg,src)
prior=totals_common(c,s,program,process)
yc=np.array([prior[k] for k in ('time_s','energy_Wh','material_added_g','wire_consumed_g','gas_consumed_L','spot_count')],np.float32)
cp=np.array([prior['components'][k] for k in COMPONENTS],np.float32)
cr=float(np.exp(np.clip(np.asarray(m['previous_motions'])[::4].mean(),-.6,.6)))
sc=calibrations(src,process,np.asarray(m['previous_power_probes_W']))
cp[0]*=cr;cp[2]*=sc;yc[1]=float(cp.sum())
j,g=feature(cfg,src,program)
raw={'joint':j[None],'global_features':g[None],'history':hist[None],'mask':np.r_[np.ones(cfg['n']),np.zeros(15-cfg['n'])][None],
'y':np.zeros((1,6),np.float32),'components':np.ones((1,9),np.float32),'prior':yc[None],'component_prior':cp[None],
'robot_ids':np.array([item.get('robot_id','provided')]),'program_ids':np.array([item.get('program_id','provided')]),
'topology':np.array([item.get('topology','unknown')]),'n_axes':np.array([cfg['n']])}
checkpoint=checkpoint or ROOT/'reports'/('rsw' if process=='rsw' else 'tig')/'model.pt'
ck=torch.load(checkpoint,map_location='cpu',weights_only=False)
d=WeldingSet(raw,norm=ck['normalizer'])
net=WeldingMoE55(ck['global_dim'],ck['history_dim'],process)
net.load_state_dict(ck['model_state'])
yp,comps=predict(net,d,blend=ck['blend'])
return {'process':process,'predictions':dict(zip(('time_s','energy_Wh','material_added_g','wire_consumed_g','gas_consumed_L','spot_count'),map(float,yp[0]))),
'components_Wh':dict(zip(COMPONENTS,map(float,comps[0]))),'source':'previous public machine histories and physics-informed nominal prior'}
if __name__=='__main__':
a=argparse.ArgumentParser();a.add_argument('--input',required=True);a.add_argument('--output');aa=a.parse_args()
res=calculate(json.load(open(aa.input)))
if aa.output:Path(aa.output).write_text(json.dumps(res,indent=2))
else:print(json.dumps(res,indent=2))
|