research.py

from pathlib import Path
from fractions import Fraction as F
import json, hashlib, datetime
ROOT=Path(__file__).resolve().parent.parent
FIRST,LAST=1132,1431
E,P,J=F(25,23),F(70,69),F(300,299)
REPLAY=False
def artifact(relative_path,content):
    p=ROOT/relative_path
    assert not Path(relative_path).is_absolute() and '..' not in Path(relative_path).parts
    data=content.encode()
    if not REPLAY:
        p.parent.mkdir(parents=True,exist_ok=True)
        if p.exists():assert p.read_bytes()==data, 'Immutable artifact already exists with different bytes'
        else:p.write_bytes(data)
    return {'path':relative_path,'sha256':hashlib.sha256(data).hexdigest(),'bytes':len(data)}
def inv(n):
    if not isinstance(n,int) or n<0:raise ValueError('nonnegative original integer duration required')
    if n==0:return 0
    k=1
    while n%10==0:n//=10;k*=10
    return k*int(str(n)[::-1])
def clean(x):
    if isinstance(x,F):return x.numerator if x.denominator==1 else str(x)
    if isinstance(x,dict):return {str(k):clean(v) for k,v in x.items()}
    if isinstance(x,(list,tuple)):return [clean(v) for v in x]
    return x
def begin(n,title,question,inputs,sources):
    p=ROOT/'journal.json';rows=json.loads(p.read_text()) if p.exists() else []
    if n!=FIRST+len(rows) or n>LAST:raise ValueError('Sequential research step required')
    for checkpoint in range(1181,n,50):
        assert (ROOT/'evidence/strategy_reviews'/('review_'+str(checkpoint)+'.json')).exists(), 'Full Strategy review required before proceeding'
    pending=ROOT/'pending.json'
    if pending.exists():
        saved=json.loads(pending.read_text())
        assert (saved['step'],saved['title'],saved['question'],saved['inputs'],saved['sources'])==(n,title,question,clean(inputs),sources)
        return saved
    predecessor=json.loads((ROOT/'evidence/PREDECESSOR.json').read_text())['sha256']
    item=dict(step=n,title=title,question=question,inputs=clean(inputs),sources=sources,opened_utc=datetime.datetime.now(datetime.timezone.utc).isoformat(),predecessor_sha256=rows[-1]['sha256'] if rows else 'C1131:'+predecessor)
    pending.write_text(json.dumps(item,indent=2,ensure_ascii=False)+'\n')
    return item
def finish(item,results,finding,next_question,checks):
    assert checks and all(checks.values()),checks
    item.update(results=clean(results),finding=finding,reassessment=next_question,checks=checks,closed_utc=datetime.datetime.now(datetime.timezone.utc).isoformat())
    item['sha256']=hashlib.sha256(json.dumps(item,sort_keys=True).encode()).hexdigest()
    p=ROOT/'journal.json';rows=json.loads(p.read_text()) if p.exists() else []
    assert item['step']==FIRST+len(rows)
    rows.append(item);p.write_text(json.dumps(rows,indent=2,ensure_ascii=False)+'\n')
    (ROOT/'pending.json').unlink()
    print(json.dumps({'step':item['step'],'results':item['results'],'finding':finding,'next':next_question},ensure_ascii=False))