research.py

"""Append one completed, reassessed research action; never edit predecessors."""
from pathlib import Path
from datetime import datetime, timezone
from fractions import Fraction
import hashlib, json

ROOT = Path(__file__).resolve().parent
JOURNAL = ROOT / 'journal.json'
PARENT = ROOT / 'evidence/C1584_parent_record.json'

def encode(v):
    if isinstance(v, Fraction):
        return v.numerator if v.denominator == 1 else str(v)
    if isinstance(v, Path): return str(v)
    if isinstance(v, dict): return {str(k):encode(x) for k,x in v.items()}
    if isinstance(v, (list,tuple)): return [encode(x) for x in v]
    return v

def digest(v):
    return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(',',':'),ensure_ascii=False).encode()).hexdigest()

def begin(title, question, sources=()):
    journal=json.loads(JOURNAL.read_text()) if JOURNAL.exists() else []
    assert not (ROOT/'pending.json').exists(), 'Complete pending action first'
    record={'step':1585+len(journal),'title':title,'question':question,
            'sources':list(sources),'opened_utc':datetime.now(timezone.utc).isoformat(),
            'predecessor_sha256':journal[-1]['sha256'] if journal else json.loads(PARENT.read_text())['sha256']}
    assert record['step']<=1634
    (ROOT/'pending.json').write_text(json.dumps(record,indent=2)+'\n')
    return record['step']

def finish(results, finding, reassessment, checks=None):
    record=json.loads((ROOT/'pending.json').read_text())
    record.update(results=encode(results),finding=finding,reassessment=reassessment,
                  checks=encode(checks or {}),closed_utc=datetime.now(timezone.utc).isoformat())
    assert all(record['checks'].values()), record['checks']
    record['sha256']=digest(record)
    journal=json.loads(JOURNAL.read_text()) if JOURNAL.exists() else []
    journal.append(record)
    JOURNAL.write_text(json.dumps(journal,indent=2,ensure_ascii=False)+'\n')
    (ROOT/'pending.json').unlink()
    print(json.dumps({'step':record['step'],'finding':finding,'next':reassessment},ensure_ascii=False))

def inverse(n):
    assert isinstance(n,int) and n>0
    s=str(n); core=s.rstrip('0')
    return int(core[::-1])*10**(len(s)-len(core))

def path(legs, anchor=1406):
    p=[anchor]
    for n in legs: p.append(p[-1]+inverse(n))
    return p

def key(x,a,k=Fraction(25,23)):
    return a+k*(x-a)

def model(name,value=None):
    p=ROOT/'model'/f'{name}.json'
    if value is not None: p.write_text(json.dumps(encode(value),indent=2,ensure_ascii=False)+'\n')
    return json.loads(p.read_text())