nanobrush / scripts /map_references.py
nmarshall's picture
research: map reference sources
9c9a58f
Raw History Blame Contribute Delete
23.5 kB
#!/usr/bin/env python3
"""Deterministic static IXPL/resource index; does not compile or execute references."""
import argparse
import ast
from collections import Counter, defaultdict
import hashlib
import json
from pathlib import Path, PurePosixPath
import re
import sys
import references
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'assets/analysis/source-map.json'
TEXT_SUFFIXES = {'.ixpl', '.txt', '.cfg', '.shader', '.bat'}
def lex(source):
"""Return offset/line-preserving tokens; comments disappear, strings stay opaque.
IXPL delimiter strings (backslash quote delimiter ... delimiter quote) may
contain arbitrary multiline text. This is a source index, not an IXPL parser.
"""
tokens = []
i, line = 0, 1
while i < len(source):
start, begin = i, line
c = source[i]
if c.isspace():
line += c == '\n'
i += 1
continue
if source.startswith('//', i):
end = source.find('\n', i)
i = len(source) if end < 0 else end
continue
if source.startswith('/*', i):
end = source.find('*/', i + 2)
if end < 0:
raise ValueError(f'Unterminated comment at line {line}')
i = end + 2
line += source[start:i].count('\n')
continue
if source.startswith('\\"', i):
if i + 2 >= len(source):
raise ValueError(f'Incomplete delimiter string at line {line}')
delimiter = source[i + 2]
closing = {'[': ']', '{': '}', '(': ')', '<': '>'}.get(delimiter, delimiter) + '"'
end = source.find(closing, i + 3)
if end < 0:
raise ValueError(f'Unterminated delimiter string at line {line}')
i = end + 2
kind = 'string'
elif c in '\"\'':
i += 1
while i < len(source):
if source[i] == '\\':
i += 2
elif source[i] == c:
i += 1
break
else:
i += 1
else:
raise ValueError(f'Unterminated quoted literal at line {line}')
kind = 'string' if c == '"' else 'char'
elif c.isalpha() or c in '_$@':
i += 1
while i < len(source) and (source[i].isalnum() or source[i] in '_$'):
i += 1
kind = 'identifier'
elif c.isdigit():
i += 1
while i < len(source) and (source[i].isalnum() or source[i] == '.'):
i += 1
kind = 'number'
else:
i += 2 if source[i:i+2] in ('::', '+=', '==', '!=', '@=', '<=', '>=') else 1
kind = 'symbol'
tokens.append({'kind': kind, 'text': source[start:i], 'offset': start, 'end': i, 'line': begin})
line += source[start:i].count('\n')
return tokens
def masked(source, tokens):
result = list(source)
for i, c in enumerate(source):
if c not in '\r\n':
result[i] = ' '
for token in tokens:
if token['kind'] not in ('string', 'char'):
result[token['offset']:token['end']] = source[token['offset']:token['end']]
return ''.join(result)
def arguments(tokens, opening):
"""Split a call's balanced parentheses/arrays/braces, without parsing types."""
stack, values, start = [], [], opening + 1
pairs = {')': '(', ']': '[', '}': '{'}
for i in range(opening, len(tokens)):
t = tokens[i]['text'] if tokens[i]['kind'] == 'symbol' else ''
if t in ('(', '[', '{'):
stack.append(t)
elif t in pairs:
if not stack or stack[-1] != pairs[t]:
raise ValueError(f'Unbalanced call at line {tokens[opening]["line"]}')
stack.pop()
if not stack:
if i > start:
values.append(tokens[start:i])
return values, i
elif t == ',' and len(stack) == 1:
values.append(tokens[start:i]); start = i + 1
raise ValueError(f'Unterminated call at line {tokens[opening]["line"]}')
def text_of(source, tokens):
return source[tokens[0]['offset']:tokens[-1]['end']] if tokens else ''
def constant_value(expression, values):
node = ast.parse(expression, mode='eval').body
def visit(n):
if isinstance(n, ast.Constant) and type(n.value) is int:
return n.value
if isinstance(n, ast.Name) and n.id in values:
return values[n.id]
if isinstance(n, ast.UnaryOp) and isinstance(n.op, (ast.UAdd, ast.USub)):
return visit(n.operand) * (-1 if isinstance(n.op, ast.USub) else 1)
if isinstance(n, ast.BinOp) and isinstance(n.op, (ast.Add, ast.Sub)):
return visit(n.left) + (visit(n.right) if isinstance(n.op, ast.Add) else -visit(n.right))
raise ValueError('Unsupported or unresolved constant expression')
return visit(node)
def scan_ixpl(source):
tokens = lex(source)
code = masked(source, tokens)
aliases = dict(re.findall(r'\bnamespace\s+(\w+)\s*=\s*(\w+)\s*;', code))
constants = []
for m in re.finditer(r'\b((?:_?INT_\w+|CMD_\w+|coreVersion|akulaCoreVersion))\s*=\s*([^,;\r\n]+)', code):
constants.append({'name': m[1], 'expression': m[2].strip(), 'line': source.count('\n', 0, m.start(1)) + 1})
calls = []
for i, token in enumerate(tokens[:-1]):
if token['kind'] != 'identifier' or tokens[i+1]['text'] != '(':
continue
qualified = i >= 2 and tokens[i-1]['text'] == '::'
if not qualified and token['text'] != 'interrupt':
continue
args, end = arguments(tokens, i+1)
qualifier = tokens[i-2]['text'] if qualified else None
calls.append({'qualifier': aliases.get(qualifier, qualifier), 'method': token['text'],
'line': token['line'], 'offset': token['offset'],
'arguments': [text_of(source, a) for a in args]})
includes = [{'include': m[1], 'line': source.count('\n', 0, m.start(1)) + 1}
for m in re.finditer(r'^[ \t]*#[ \t]*include[ \t]+"([^"\r\n]+)"', source, re.M)
if any(t['text'] == '#' and t['offset'] <= m.start(1) and source.count('\n',0,t['offset']) + 1 == source.count('\n',0,m.start(1)) + 1 for t in tokens)]
# Source-level wrappers, with complete balanced signatures (including tuple args).
headers = []
token_offsets = {t['offset']:i for i,t in enumerate(tokens)}
known_types = {"void", "int", "long", "float", "double", "vec3", "string", "Tool", "Plane", "UV", "AABB", "Config"} | set(re.findall(r"\btype\s+(\w+)\s*=",code))
type_pattern = "|".join(re.escape(t) for t in sorted(known_types))
for match in re.finditer(r'^[ \t]*((?:(?:' + type_pattern + r')\b[^\n\r;={}]*?|\([^\n\r;={}]*\)[\[\]& \t]*))[ \t&]+([A-Za-z_]\w*)[ \t]*\(', code, re.M):
prefix = match[1].strip()
if not prefix or prefix.split()[0] in ('if', 'while', 'for', 'return', 'switch', 'else'):
continue
opening = token_offsets.get(match.end()-1)
if opening is None:
continue
args, end = arguments(tokens, opening)
headers.append({'name': match[2], 'line': source.count('\n',0,match.start(2))+1,
'offset': match.start(2), 'signature': re.sub(r'\s+', ' ', source[match.start(1):tokens[end]['end']]).strip(),
'declaration': tokens[end+1]['text'] == ';' if end+1 < len(tokens) else False})
wrappers = []
for call in calls:
if call['method'] == 'interrupt':
preceding = [h for h in headers if h['offset'] < call['offset'] and not h['declaration']]
wrapper = preceding[-1] if preceding else None
wrappers.append({**call, 'wrapper': wrapper})
bindings = {}
for i,t in enumerate(tokens[:-2]):
if t['kind'] == 'identifier' and tokens[i+1]['text'] == '=':
j, depth = i+2, 0
while j < len(tokens):
value = tokens[j]['text'] if tokens[j]['kind'] == 'symbol' else ''
if value in ('(', '[', '{'): depth += 1
elif value in (')', ']', '}'): depth -= 1
if depth < 0 or (depth == 0 and value in (',', ';')): break
j += 1
expr = tokens[i+2:j]
if expr and all(x['kind'] == 'string' or x['text'] == '+' or x['kind'] == 'identifier' for x in expr):
bindings.setdefault(t['text'], []).append({'expression': text_of(source,expr), 'line':t['line']})
return {'includes': includes, 'aliases': aliases, 'constants': constants, 'qualified_calls': calls,
'interrupt_wrappers': wrappers, 'string_bindings': bindings}
def literal_string(expression, bindings, trail=()):
"""Resolve only an unambiguous concatenation of literals and local bindings.
Scope/flow is not reconstructed: conflicting same-named declarations remain
unresolved, even when a human could infer the active one.
"""
parts = lex(expression)
if not parts or len(parts) % 2 == 0 or any(t['text'] != '+' for t in parts[1::2]):
return None
result = ''
for token in parts[::2]:
value = token['text']
if token['kind'] == 'string':
if value.startswith('\\"'):
result += value[3:-2]
else:
result += re.sub(r"\\([rnt\\\"'])", lambda m: {'r':'\r','n':'\n','t':'\t'}.get(m[1],m[1]), value[1:-1])
elif token['kind'] == 'identifier' and value in bindings and value not in trail:
candidates = [literal_string(x['expression'],bindings,(*trail,value)) for x in bindings[value]]
if not candidates or any(x is None for x in candidates) or len(set(candidates)) != 1:
return None
result += candidates[0]
else:
return None
return result
def category(path):
p = PurePosixPath(path)
if p.suffix == '.ixpl':
if '/plugins/extra stuff/' in path: return 'experimental-plugin'
if '/plugins/' in path: return 'plugin'
if '/models/' in path: return 'model-generator'
if '/include/core/' in path: return 'host-api'
if '/include/tasks/' in path: return 'task-client'
return 'ixpl-library'
if p.suffix == '.exe': return 'executable-unexamined'
if 'grammar' in p.name: return 'grammar'
return {'.bat':'batch-script','.cfg':'configuration','.shader':'shader','.png':'image'}.get(p.suffix,'reference-text')
def index_build(root, inventory):
files = {x['path']:x for x in inventory['files']}
paths = set(files)
include_roots = [PurePosixPath('tools') / p.replace('\\','/') for p in (root/'tools/includePaths.txt').read_text().splitlines() if p.strip()]
# Normalize relative traversal in trusted include search directories only.
include_roots = [Path(root/p).resolve().relative_to(root.resolve()).as_posix() for p in include_roots]
records = []
for path, item in sorted(files.items()):
raw = (root/path).read_bytes()
if hashlib.sha256(raw).hexdigest() != item['sha256']:
raise ValueError(f'Modified/unhydrated source: {root/path}')
record = {'path':path,'sha256':item['sha256'],'bytes':len(raw),'category':category(path),
'coverage':'lexically-indexed' if PurePosixPath(path).suffix.lower() in {'.ixpl','.shader','.cfg'} or category(path)=='grammar' else 'text-inventoried' if PurePosixPath(path).suffix.lower() in TEXT_SUFFIXES else 'inventory-only',
'owner':'Nanobrush reference research',
'followup':{'host-api':'P4.03','plugin':'P4.08','experimental-plugin':'P4.09','task-client':'P4.05','ixpl-library':'P4.07','model-generator':'P9.06','grammar':'P6.01','executable-unexamined':'P3.06','shader':'P9.01','configuration':'P8.09','image':'P9.05','batch-script':'P4.12'}.get(category(path),'P3.02')}
if PurePosixPath(path).suffix.lower() in TEXT_SUFFIXES:
try: text,encoding = raw.decode('utf-8-sig'),'utf-8-sig'
except UnicodeDecodeError: text,encoding = raw.decode('cp1252'),'cp1252-fallback'
record.update(encoding=encoding,lines=len(text.splitlines()))
if path.endswith('.ixpl'):
record.update(scan_ixpl(text))
for edge in record['includes']:
candidates = [PurePosixPath(path).parent/edge['include'],*(PurePosixPath(d)/edge['include'] for d in include_roots)]
edge['candidates'] = sorted({c.as_posix() for c in candidates if c.as_posix() in paths})
edge['status'] = 'resolved-candidate' if len(edge['candidates'])==1 else 'ambiguous' if edge['candidates'] else 'missing'
elif path.endswith('.shader'):
record['sections'] = [{'name':m[1],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'\[@([^\]]+)\]',text)]
record['uniforms'] = [{'type':m[1],'name':m[2],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'\buniform\s+(\w+)\s+(\w+)\s*;',text)]
elif path.endswith('.cfg'):
record['keys'] = [{'type':m[1],'key':m[2],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'^\s*(bool|int|long|float|double|vec3|string)\s+([\w.]+)\s*=',text,re.M)]
elif record['category']=='grammar':
record['rules'] = [{'name':m[1],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'^\s*(#?\w+)\s*=\s*\{',text,re.M)]
elif path.endswith('.bat'):
record['lines_text'] = [{'line':i,'text':line} for i,line in enumerate(text.splitlines(),1) if line.strip()]
records.append(record)
core = [r for r in records if r['category']=='host-api']
values, pending = {},[(r,c) for r in core for c in r.get('constants',[])]
while pending:
progress = False
for r,c in pending[:]:
try: value = constant_value(c['expression'],values)
except (ValueError,SyntaxError): continue
values[c['name']] = value;c['value']=value;pending.remove((r,c));progress=True
if not progress:break
for r in records:
if r['category']!='host-api':continue
for wrapper in r.get('interrupt_wrappers',[]):
argument = wrapper['arguments'][0]
try: wrapper['interrupt_id'] = constant_value(argument,values)
except (ValueError,SyntaxError): wrapper['interrupt_id']=None
graph = {r['path']:sorted({p for e in r.get('includes',[]) for p in e['candidates']}) for r in records if r['path'].endswith('.ixpl')}
for r in records:
if r['category'] not in ('plugin','experimental-plugin','model-generator'):continue
seen,stack = set(),list(graph[r['path']])
while stack:
path=stack.pop()
if path in seen:continue
seen.add(path);stack.extend(graph.get(path,[]))
r['transitive_includes']=sorted(seen)
r['core_calls']=sorted({c['method'] for c in r['qualified_calls'] if c['qualifier']=='Core'})
r['interface_lookups']=[c for c in r['qualified_calls'] if c['method']=='getToolID']
r['registrations']=[c for c in r['qualified_calls'] if c['method'] in ('initTool','registerPackage','setRuntimeOnAllLoaded','setRuntimeOnFinalization')]
r['registered_names'] = []
for call in r['registrations']:
if call['method'] != 'initTool': continue
args = call['arguments']
name = literal_string(args[0],r['string_bindings']) if args else None
if name is None and len(args)>1: name = literal_string(args[1],r['string_bindings'])
r['registered_names'].append({'line':call['line'],'name':name,'arguments':args})
r['resource_calls']=[c for c in r['qualified_calls'] if c['method'] in ('loadConfigFile','saveConfigFile','loadGUITexture','loadTextFile','saveTextFile','callProgram','saveWorldToFile','loadWorldFromFile')]
for call in r['resource_calls']:
literal = literal_string(call['arguments'][0],r['string_bindings']) if call['arguments'] else None
call['literal_path'] = literal
call['exact_name_candidates'] = sorted(p for p in paths if literal is not None and (p == literal.replace('\\','/') or PurePosixPath(p).name == PurePosixPath(literal.replace('\\','/')).name))
call['status'] = 'dynamic' if literal is None else 'exact-name-candidate' if call['exact_name_candidates'] else 'not-present-by-exact-name'
return {'build':inventory['build'],'archive_sha256':inventory['archive_sha256'],'include_search_roots':include_roots,
'categories':dict(sorted(Counter(r['category'] for r in records).items())),
'unresolved_core_constants':[{'path':r['path'],**c} for r,c in pending], 'files':records}
def differences(builds):
result=[]
for old,new in zip(builds,builds[1:]):
a={f['path']:f['sha256'] for f in old['files']};b={f['path']:f['sha256'] for f in new['files']}
added,removed=sorted(b.keys()-a.keys()),sorted(a.keys()-b.keys())
result.append({'from':old['build'],'to':new['build'],'added':added,'removed':removed,
'changed':[p for p in sorted(a.keys()&b.keys()) if a[p]!=b[p]],
'identical_rename_candidates':[{'from':p,'to':q} for p in removed for q in added if a[p]==b[q]],
'identical_copy_candidates':[{'from':p,'to':q} for p in sorted(a) for q in added if p not in removed and a[p]==b[q]],
'modified_same_basename_candidates':[{'from':p,'to':q} for p in removed for q in added if PurePosixPath(p).name==PurePosixPath(q).name and a[p]!=b[q]]})
return result
def validate_report(builds):
"""Corpus-level closure checks, separate from the permissive lexical scanner."""
for build in builds:
if build['unresolved_core_constants']:
raise ValueError('Unresolved Core constants: ' + build['build'])
core = [f for f in build['files'] if f['category'] == 'host-api']
declared = {c['value'] for f in core for c in f['constants'] if c['name'].startswith('INT_')}
wrappers = [w for f in core for w in f.get('interrupt_wrappers', [])]
if {w.get('interrupt_id') for w in wrappers} != declared or declared != set(range(max(declared)+1)):
raise ValueError('Core endpoint coverage/range differs: ' + build['build'])
if any(w['wrapper'] is None or w['wrapper']['line'] != w['line'] for w in wrappers):
raise ValueError('A Core interrupt has no same-line signature: ' + build['build'])
if any(len(e['candidates']) != 1 for f in build['files'] for e in f.get('includes', [])):
raise ValueError('Include candidate closure differs: ' + build['build'])
def render_api_index(build):
lines = ['# Core interrupt index', '',
'Source-only endpoint enumeration for build `2016-03-10-c`. IDs are evaluated source constants, not a binary dispatch trace. Full signatures, arguments and older builds are in [source-map.json](../assets/analysis/source-map.json). See [HOST_APIS.md](HOST_APIS.md).', '']
for item in build['files']:
if item['category'] != 'host-api' or not item.get('interrupt_wrappers'):
continue
lines += ['## ' + PurePosixPath(item['path']).name, '',
'Source: `' + item['path'] + '`; SHA-256 `' + item['sha256'] + '`.', '',
'| ID | Constant(s) | Wrapper(s), source lines |', '| ---: | --- | --- |']
groups = defaultdict(list)
for wrapper in item['interrupt_wrappers']:
groups[wrapper['interrupt_id']].append(wrapper)
for value, wrappers in sorted(groups.items()):
constants = sorted({w['arguments'][0].lstrip('+').strip() for w in wrappers})
names = ', '.join('`' + w['wrapper']['name'] + '`:' + str(w['line']) for w in wrappers)
lines.append('| ' + str(value) + ' | ' + ', '.join('`' + c + '`' for c in constants) + ' | ' + names + ' |')
lines.append('')
return '\n'.join(lines)
def render_plugin_index(build):
lines = ['# Plugin tool registration index', '',
'Baseline `2016-03-10-c`. These are lexical `initTool` call sites, including internal/service tools and experiments. Conservative name resolution does not prove runtime reachability, load order or simultaneous registration. See [PLUGINS.md](PLUGINS.md).', '']
for item in build['files']:
if not item.get('registered_names'):
continue
lines += ['## ' + PurePosixPath(item['path']).stem, '',
'Source: `' + item['path'] + '`; SHA-256 `' + item['sha256'] + '`.', '',
'| Tool name expression resolved to | Line | Press callback expression |', '| --- | ---: | --- |']
for entry in item['registered_names']:
args = entry['arguments']
direct = literal_string(args[0], item['string_bindings']) is not None
index = 1 if direct else 2
callback = args[index] if len(args) > index else 'unknown'
escape = lambda value: value.replace('|', '\\|').replace('\n', ' ')
lines.append('| `' + escape(entry['name'] or '(unresolved)') + '` | ' + str(entry['line']) + ' | `' + escape(callback) + '` |')
lines.append('')
return '\n'.join(lines)
def main():
parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--check',action='store_true');args=parser.parse_args()
builds=[]
for p in json.loads(references.MANIFEST.read_text())['packages']:
inventory=references.inspect_archive(references.REFERENCES/'archives'/p['filename'],p)
root=references.REFERENCES/'extracted'/p['build'];references.check_extracted(root,inventory)
builds.append(index_build(root,inventory))
validate_report(builds)
report={'schema':1,'method':'static lexical source index; no reference execution','generator_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'limitations':['Include resolution records candidates, not measured compiler search order.','Qualified calls are lexical references, not proven execution or a typed call graph.','CP1252 is an explicit fallback, not an established encoding contract.','No host implementation, binary serializer, or bytecode layout is inferred from an API declaration.'],
'builds':builds,'version_differences':differences(builds)}
documents = {ROOT/'docs/API_INDEX.md':render_api_index(builds[-1]), ROOT/'docs/PLUGIN_INDEX.md':render_plugin_index(builds[-1])}
if args.check:
if not OUT.exists() or json.loads(OUT.read_text())!=report:raise ValueError('Static map is absent/stale; regenerate it')
for path,content in documents.items():
if not path.exists() or path.read_text()!=content:raise ValueError(f'Stale generated index: {path}')
else:
references.write_json(OUT,report)
for path,content in documents.items():path.write_text(content)
print(json.dumps({'builds':len(builds),'files':sum(len(b['files']) for b in builds),'latest_categories':builds[-1]['categories'],'report':str(OUT.relative_to(ROOT))},indent=2))
if __name__=='__main__':
try:main()
except (OSError,ValueError) as error:print(f'Source map: {error}',file=sys.stderr);sys.exit(1)