File size: 6,523 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 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """Learn a machine state from prior SENSOR-LIKE power observations only.
Synthetic probe is created outside the predictive path. It emulates measurements
from a separate, previously completed, no-cut spindle/axis diagnostic; current
job energy and hidden machine_gain are NEVER passed to the expert at inference.
This is an ANALYTIC TEST, not a real sensor dataset or a validated power meter.
"""
from __future__ import annotations
import json, math, hashlib, random
from pathlib import Path
import numpy as np
import torch
from torch import nn
MACHINE_GROUPS=['drilling_horizontal','boring_horizontal','boring_vertical','reaming_horizontal','reaming_vertical','tapping_horizontal','tapping_vertical','grinding_surface','sawing_band','laser_sheet','plasma_sheet']
SENSOR_NAMES=['idle_total_W','spindle_1000_W','spindle_3000_W','axis_x_20mms_W','axis_z_20mms_W','coolant_increment_W']
HISTORY_DIM=len(SENSOR_NAMES)
def synthesize_history(gain:float, orientation:str, seed_string:str):
"""Synthetic prior measurement, independent of target job, never used by predictor.
This function must ONLY be invoked by generate_machine_histories.py.
A separate no-cut standard probe is performed prior to all job predictions.
"""
rng=random.Random(int(hashlib.sha256(seed_string.encode()).hexdigest()[:15],16))
g=float(gain);z=orientation=='vertical'
off=(62+17*g)+(15+(27 if z else 13)*g)+(17+(6 if z else 10))
s=lambda rpm: (46+.021*rpm+1.1e-6*rpm*rpm)*g*(.91+.13*g)
motor_factor=.96+.07*g
axis_x=(6+(.78*20+.018*20**2))*motor_factor
axis_z=(6+(.78*20+.018*20**2)*1.32)*motor_factor
val=[off,off+s(1000),off+s(3000),off+axis_x,off+axis_z,120.]
# Probe power sensor read-out error: reproducible, 0.3% Gaussian per channel.
measured=[max(0.,v*(1.+rng.gauss(0.,.003))) for v in val]
return measured
def save_histories(root:Path):
"""Generate *independent historical* power-probe values for all 7 NEW hole groups.
The private source's gain parameter is consulted ONLY here to synthesize
earlier instrument observations. Neither it nor current-job energy labels
are written into active machine history, checkpoint inputs or training data.
"""
import csv
root=Path(root)
(root/'data/machine_history').mkdir(parents=True,exist_ok=True)
coverage={}
for split in ('train','val','test'):
original=root/'data/new_before_energy_revision'/f'{split}.csv'
rows=list(csv.DictReader(original.open(encoding='utf-8')))
mapping={};groups={}
for r in rows:
key=r['operation']+'_'+r['orientation']
if key not in MACHINE_GROUPS:continue
pid=r['program_id'];meta=json.loads(r['initial_profile'])
grp=int(pid.split('_')[-1])//6
machine=f'MACHINE_{key}_{grp:02d}'
if machine in groups and abs(groups[machine][0]-float(meta['hidden_machine_gain']))>1e-10:
raise ValueError('Inconsistent hidden machine metadata for '+machine)
groups[machine]=(float(meta['hidden_machine_gain']),meta['orientation'])
mapping[pid]=machine
records={key:synthesize_history(g,o,key) for key,(g,o) in groups.items()}
obj={'sensor_names':SENSOR_NAMES,'machine_measurements':records,'program_to_machine':mapping,
'provenance':'Synthetic power measurements collected before target job; 0.3% sensor error; zero target-label access. No hidden gain in serialized data.'}
(root/'data/machine_history'/f'{split}.json').write_text(json.dumps(obj,indent=2),encoding='utf-8')
coverage[split]={'machines':len(groups),'programs':len(mapping)}
(root/'data/machine_history/README.md').write_text("""# SYNTHETIC independent prior power tests — NOT CNC sensor readings
Applies to drilling-horizontal and both orientations of boring, reaming and tapping.
Each synthetic six-program machine group is assigned a reproducible, independently
noised prior no-cut power probe (6 watt readings). This probe is calculated by the
**private teaching simulator** from its latent efficiency coefficient, but the
coefficient NEVER appears in active inputs or targets. A program's future total
energy and component targets are not used to synthesize its prior probe.
Machine groups do not cross train/val/test; the gain VALUE may repeat across splits,
so these experiments do not establish extrapolation to entirely unseen efficiency
ranges. Real machine validation is still necessary. If no telemetry exists,
model uses cold-start fallback without any surrogate hidden gain.
""",encoding='utf-8')
return coverage
def history_for_program(root:Path,split:str,pid:str):
f=Path(root)/'data/machine_history'/f'{split}.json'
js=json.loads(f.read_text())
m=js['program_to_machine'].get(pid)
return js['machine_measurements'].get(m) if m else None
class MachineCharacterizationExpert(nn.Module):
"""Only previous independent power probes; no private machine gain feature."""
def __init__(self,components=60,latent_dim=None):
super().__init__()
from config import choice
latent_dim=choice(6,10,16) if latent_dim is None else latent_dim
a,b=choice((20,16),(32,24),(48,32))
c,d=choice((24,12),(40,20),(64,32))
self.encoder=nn.Sequential(nn.Linear(HISTORY_DIM,a),nn.GELU(),nn.Linear(a,b),nn.GELU(),nn.Linear(b,latent_dim))
self.component=nn.Embedding(components,6)
self.operation=nn.Embedding(len(MACHINE_GROUPS),5)
self.gate=nn.Sequential(nn.Linear(latent_dim+6+5,c),nn.GELU(),nn.Linear(c,d),nn.GELU(),nn.Linear(d,1))
nn.init.zeros_(self.gate[-1].weight);nn.init.zeros_(self.gate[-1].bias)
self.register_buffer('input_mean',torch.zeros(len(MACHINE_GROUPS),HISTORY_DIM))
self.register_buffer('input_std',torch.ones(len(MACHINE_GROUPS),HISTORY_DIM))
def set_scaler(self,mean,std):
self.input_mean.copy_(torch.as_tensor(mean,dtype=torch.float32))
self.input_std.copy_(torch.as_tensor(std,dtype=torch.float32))
def forward(self,powers,component_index,available,group_index=None):
if group_index is None:group_index=torch.zeros_like(component_index)
h=self.encoder(torch.clamp((powers-self.input_mean[group_index])/self.input_std[group_index],-8,8))
delta=self.gate(torch.cat([h,self.component(component_index),self.operation(group_index)],dim=-1)).squeeze(-1)
return torch.where(available,torch.exp(.55*torch.tanh(delta)),torch.ones_like(delta))
|