Spaces:
Running on Zero
Running on Zero
File size: 11,748 Bytes
033aa66 71e8313 033aa66 71e8313 033aa66 71e8313 033aa66 71e8313 033aa66 71e8313 033aa66 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | """Translate SJS with pinned GradResearch code, then lower its custom profile.
The untouched @sjs profile stays in the research archive. The portable download
uses actual components, typed ports and flows; descriptive behavior remains docs.
"""
import json
from sjs import translator
FIELDS = {
'root': {'$schema', 'model_meta', 'subsystems', 'item_flows', 'requirements', 'behaviour', 'allocations'},
'model_meta': {'system_name', 'system_id', 'description', 'version', 'lifecycle_stage', 'domain', 'maturity', 'authors', 'date_modified', 'context'},
'subsystems': {'subsystem_id', 'subsystem_name', 'sysml_equivalent', 'functional_basis', 'description', 'domain', 'maturity', 'parts', 'ports', 'interfaces', 'notes'},
'ports': {'port_id', 'name', 'direction', 'flow_type', 'connector_type', 'notes'},
'parts': {'part_id', 'description', 'quantity', 'notes'},
'interfaces': {'interface_id', 'mating_subsystem', 'port_this', 'port_mate', 'flow_ref', 'interface_type', 'notes'},
'item_flows': {'flow_id', 'name', 'flow_type', 'notes'},
'requirements': {'req_id', 'category', 'priority', 'statement', 'rationale', 'satisfied_by', 'notes'},
'behaviour': {'actions', 'state_machines'},
'actions': {'action_id', 'name', 'owner', 'steps', 'notes'},
'steps': {'step_no', 'description'},
'state_machines': {'sm_id', 'name', 'owner', 'entry_state', 'states', 'notes'},
'states': {'name', 'description', 'entry_action', 'exit_action', 'transitions', 'notes'},
'transitions': {'name', 'trigger', 'guard', 'action', 'target'},
'allocations': {'allocation_id', 'type', 'from', 'to', 'rationale', 'notes'},
}
def validate_model(model):
errors = []
if not isinstance(model, dict):
return ['SJS must be an object.']
def fields(value, kind):
if not isinstance(value, dict):
raise ValueError(f'{kind} must contain objects.')
unknown = set(value) - FIELDS[kind]
if unknown:
errors.append(f'Unsupported {kind} fields: {sorted(unknown)}; omit rather than silently lose them.')
for key, child in value.items():
if key in FIELDS:
if key in {'model_meta', 'behaviour'}:
fields(child, key)
else:
if not isinstance(child, list):
raise ValueError(f'{key} must be an array.')
for entry in child:
fields(entry, key)
try:
fields(model, 'root')
if model.get('$schema') != 'sjs/1.2':
errors.append('Use $schema sjs/1.2.')
errors.extend(translator().validate(model))
if not model.get('model_meta', {}).get('date_modified'):
errors.append('model_meta.date_modified is required for a stable GradResearch round-trip.')
if not model.get('subsystems'):
errors.append('At least one supported subsystem is required.')
from quality import model_items
ids = set()
for path, item in model_items(model):
for key, value in item.items():
if key.endswith('_id'):
if not isinstance(value, str) or not value or value in ids:
errors.append(f'Unique nonempty ID required at {path}/{key}.')
else:
ids.add(value)
systems = {s['subsystem_id']: s for s in model['subsystems']}
flows = {f['flow_id']: f for f in model.get('item_flows', [])}
actions = {a['action_id']: a for a in model.get('behaviour', {}).get('actions', [])}
for subsystem in systems.values():
for key in ['subsystem_name', 'description', 'domain']:
if not isinstance(subsystem.get(key), str) or not subsystem[key].strip():
errors.append(f'{subsystem["subsystem_id"]} requires {key}.')
if subsystem.get('sysml_equivalent', 'part def') != 'part def':
errors.append('Use part def subsystems; the host creates their assembly usages.')
ports = {p['port_id']: p for p in subsystem.get('ports', [])}
for p in ports.values():
if p.get('direction') not in {'in', 'out', 'inout'} or not p.get('name') or not isinstance(p.get('flow_type'), str) or not p['flow_type']:
errors.append(f'Port {p["port_id"]} requires name, direction and flow_type.')
for part in subsystem.get('parts', []):
if type(part.get('quantity')) is not int or part['quantity'] <= 0:
errors.append('Part quantity must be a positive integer.')
for interface in subsystem.get('interfaces', []):
mate = systems.get(interface.get('mating_subsystem'), {})
remote = {p['port_id']: p for p in mate.get('ports', [])}
a, b = ports.get(interface.get('port_this')), remote.get(interface.get('port_mate'))
flow = flows.get(interface.get('flow_ref'))
if not a or not b or not flow:
errors.append(f'Interface {interface.get("interface_id")} has a missing subsystem, port or flow reference.')
elif a['flow_type'] != b['flow_type'] or a['flow_type'] != flow.get('flow_type'):
errors.append(f'Interface {interface["interface_id"]} connects incompatible flow types.')
elif (a['direction'], b['direction']) in {('in','in'), ('out','out')}:
errors.append(f'Interface {interface["interface_id"]} has incompatible port directions.')
for flow in flows.values():
if not flow.get('name') or not isinstance(flow.get('flow_type'), str) or not flow['flow_type']:
errors.append('Item flows require name and flow_type.')
for kind in ['actions', 'state_machines']:
for item in model.get('behaviour', {}).get(kind, []):
if item.get('owner') not in systems or not item.get('name'):
errors.append(f'{kind} requires a name and a valid subsystem owner.')
if kind == 'state_machines':
states = item.get('states', [])
names = [s.get('name') for s in states]
if not names or any(not isinstance(n, str) or not n for n in names) or len(set(names)) != len(names) or item.get('entry_state') not in names:
errors.append('State machine requires unique named states and an existing entry_state.')
for state in states:
if any(t.get('target') not in names for t in state.get('transitions', [])):
errors.append('State transition target is missing.')
for allocation in model.get('allocations', []):
if allocation.get('from') not in actions or allocation.get('to') not in systems:
errors.append('Allocation must reference an existing action and subsystem.')
for requirement in model.get('requirements', []):
if not requirement.get('statement') or not set(requirement.get('satisfied_by', [])) <= systems.keys():
errors.append('Requirement needs a statement and existing satisfied_by subsystem IDs.')
except (KeyError, TypeError, ValueError) as error:
errors.append(str(error))
return errors
def note(value):
return '/* ' + json.dumps(value, ensure_ascii=False).replace('*/', '* /') + ' */'
def translate(model):
errors = validate_model(model)
if errors:
raise ValueError('; '.join(errors))
native = translator()
profile = native.sysml_from_sjs(model)
restored = native.sjs_from_sysml(profile)
if restored != native.canonicalize(model):
raise ValueError('GradResearch SJS round-trip changed the model.')
return profile, portable_sysml(restored)
def to_sysml(model):
return translate(model)[1]
def portable_sysml(model):
systems = model['subsystems']
indices = {s['subsystem_id']: i for i, s in enumerate(systems, 1)}
types = sorted({p['flow_type'] for s in systems for p in s.get('ports', [])} |
{f['flow_type'] for f in model.get('item_flows', [])})
type_names = {t: f'FlowType{i}' for i, t in enumerate(types, 1)}
ports = {(s['subsystem_id'], p['port_id']): (j, p) for s in systems for j, p in enumerate(s.get('ports', []), 1)}
lines = ['package PatentModel {', ' doc ' + note(model['model_meta'])]
for t, name in type_names.items():
lines.extend([f' item def {name} {{', ' doc ' + note({'flow_type': t}), ' }'])
for i, s in enumerate(systems, 1):
lines.extend([f' part def Component{i} {{', ' doc ' + note(s)])
for j, p in enumerate(s.get('ports', []), 1):
lines.extend([f' port port{j} {{', ' doc ' + note(p),
f' {p["direction"]} item payload: {type_names[p["flow_type"]]};', ' }'])
for j, part in enumerate(s.get('parts', []), 1):
lines.extend([f' part member{j}[{part["quantity"]}] {{', ' doc ' + note(part), ' }'])
lines.append(' }')
actions = model.get('behaviour', {}).get('actions', [])
for i, a in enumerate(actions, 1):
lines.extend([f' action def Function{i} {{', ' doc ' + note(a), ' }'])
for i, sm in enumerate(model.get('behaviour', {}).get('state_machines', []), 1):
lines.extend([f' state def Behavior{i} {{', ' doc ' + note(sm)])
for j, state in enumerate(sm['states'], 1):
lines.extend([f' state state{j} {{', ' doc ' + note(state), ' }'])
lines.append(' }')
for i, r in enumerate(model.get('requirements', []), 1):
lines.extend([f' requirement def Requirement{i} {{', ' doc ' + note(r), ' }'])
lines.append(' part systemModel {')
for i, s in enumerate(systems, 1):
lines.append(f' part component{i}: Component{i};')
for i, a in enumerate(actions, 1):
lines.append(f' action function{i}: Function{i};')
lines.append(f' allocate function{i} to component{indices[a["owner"]]};')
for i, sm in enumerate(model.get('behaviour', {}).get('state_machines', []), 1):
lines.extend([f' state behavior{i}: Behavior{i};', ' // ' + note({'owner': sm['owner']})])
flow_lookup = {f['flow_id']: f for f in model.get('item_flows', [])}
count = 0
for s in systems:
for connection in s.get('interfaces', []):
count += 1
a, b = (s['subsystem_id'], connection['port_this']), (connection['mating_subsystem'], connection['port_mate'])
aj, ap = ports[a]; bj, bp = ports[b]
x, y = f'component{indices[a[0]]}.port{aj}', f'component{indices[b[0]]}.port{bj}'
lines.append(' // ' + note({'interface': connection, 'flow': flow_lookup[connection['flow_ref']]}))
if ap['direction'] == bp['direction'] == 'inout':
lines.append(f' connect {x} to {y};')
else:
if ap['direction'] == 'in' or bp['direction'] == 'out':
x, y = y, x
lines.append(f' flow transfer{count} from {x}.payload to {y}.payload;')
action_indices = {a['action_id']: i for i, a in enumerate(actions, 1)}
for allocation in model.get('allocations', []):
lines.extend([' // ' + note(allocation),
f' allocate function{action_indices[allocation["from"]]} to component{indices[allocation["to"]]};'])
lines.extend([' }', '}'])
return '\n'.join(lines) + '\n'
|