RFTSystems commited on
Commit
21686b4
·
verified ·
1 Parent(s): 6fa522a

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +30 -38
app.py CHANGED
@@ -1,6 +1,6 @@
1
  # ============================================================
2
- # RFT-Ω FRAMEWORK — TOTAL-PROOF API (Sprint 1, stable UI)
3
- # Author: Liam Grinstead (RFT Systems) | All Rights Reserved
4
  # ============================================================
5
  """
6
  Deterministic, signed validation harness for the
@@ -10,7 +10,6 @@ Rendered Frame Theory (RFT-Ω) harmonic stability kernel.
10
  import os, io, json, time, math, hashlib, zipfile, random
11
  from datetime import datetime
12
  from typing import Dict, Any, List, Tuple
13
-
14
  import numpy as np
15
  import gradio as gr
16
 
@@ -42,7 +41,7 @@ ABOUT_BLOCK = {
42
  "legal": LEGAL_NOTICE,
43
  }
44
 
45
- # ------------------ Rate limit ------------------------------
46
  RUN_HISTORY_TS: List[float] = []
47
  MAX_RUNS_PER_MINUTE = 60
48
 
@@ -63,49 +62,41 @@ PROFILES = {
63
  "Extreme Perturbation":{"base": (0.82, 0.77), "w": (0.50, 0.50)},
64
  }
65
 
66
- def _rng(seed: int) -> np.random.RandomState:
67
- return np.random.RandomState(seed)
68
 
69
- def simulate_step(rng: np.random.RandomState, profile: str, sigma: float, noise_dist: str) -> Dict[str, Any]:
70
  base_q, base_z = PROFILES[profile]["base"]
71
  wq, wz = PROFILES[profile]["w"]
72
  if noise_dist == "uniform":
73
  q_noise = rng.uniform(-sigma, sigma)
74
- z_noise = rng.uniform(-sigma * 0.8, sigma * 0.8)
75
  else:
76
  q_noise = rng.normal(0, sigma)
77
- z_noise = rng.normal(0, sigma * 0.8)
78
  q = float(np.clip(base_q + wq*q_noise, 0.0, 0.99))
79
  z = float(np.clip(base_z + wz*z_noise, 0.0, 0.99))
80
- variance = abs(q_noise) + abs(z_noise)
81
- if variance > 0.15:
82
- status = "critical"
83
- elif variance > 0.07:
84
- status = "perturbed"
85
- else:
86
- status = "nominal"
87
- return {"sigma": float(round(sigma,6)), "QΩ": q, "ζ_sync": z, "status": status}
88
 
89
  # ------------------ Schedules -------------------------------
90
- def build_sigma_series(schedule_type: str, params: Dict[str, Any]) -> List[float]:
91
- if schedule_type == "single":
92
- return [float(params.get("sigma", 0.05))]
93
- elif schedule_type == "ramp":
94
- return list(np.linspace(float(params.get("start",0.0)),
95
- float(params.get("stop",0.3)),
96
- int(params.get("steps",10))))
97
- elif schedule_type == "random":
98
  r = random.Random(int(params.get("seed",0)))
99
- return [r.uniform(float(params.get("min",0.0)),
100
- float(params.get("max",0.3)))
101
- for _ in range(int(params.get("steps",10)))]
102
- elif schedule_type == "impulse":
103
- base, spike, at, steps = float(params.get("base",0.05)), float(params.get("spike",0.25)), int(params.get("at",5)), int(params.get("steps",10))
104
  s=[base]*steps
105
  if 0<=at<len(s): s[at]=spike
106
  return s
107
- elif schedule_type == "step":
108
- before, after, at, steps = float(params.get("before",0.05)), float(params.get("after",0.2)), int(params.get("at",5)), int(params.get("steps",10))
109
  s=[before]*steps
110
  for i in range(steps):
111
  if i>=at: s[i]=after
@@ -119,8 +110,7 @@ def make_run_id(seed:int,profile:str)->str:
119
 
120
  def write_bundle(run_dir:str,config:Dict[str,Any],results:Dict[str,Any])->Tuple[str,str]:
121
  os.makedirs(run_dir,exist_ok=True)
122
- run_json=os.path.join(run_dir,"run.json")
123
- with open(run_json,"w") as f: json.dump({"config":config,"results":results},f,indent=2)
124
  canonical=json.dumps({"config":config,"results":results},sort_keys=True)
125
  digest=sha512_hex(canonical)
126
  with open(os.path.join(run_dir,"run.sha512"),"w") as f: f.write(digest+"\n")
@@ -225,7 +215,9 @@ else:
225
 
226
  # ------------------ Launch (stable queue) -------------------
227
  if __name__=="__main__":
228
- demo.queue(concurrency_count=1,max_size=5).launch(server_name="0.0.0.0",
229
- server_port=7860,
230
- show_error=True,
231
- debug=False)
 
 
 
1
  # ============================================================
2
+ # RENDERED FRAME THEORY-Ω FRAMEWORK — API (Sprint 1, Stable Build)
3
+ # Author: Liam Grinstead (RFT Systems) | All Rights Reserved
4
  # ============================================================
5
  """
6
  Deterministic, signed validation harness for the
 
10
  import os, io, json, time, math, hashlib, zipfile, random
11
  from datetime import datetime
12
  from typing import Dict, Any, List, Tuple
 
13
  import numpy as np
14
  import gradio as gr
15
 
 
41
  "legal": LEGAL_NOTICE,
42
  }
43
 
44
+ # ------------------ Rate-limit -------------------------------
45
  RUN_HISTORY_TS: List[float] = []
46
  MAX_RUNS_PER_MINUTE = 60
47
 
 
62
  "Extreme Perturbation":{"base": (0.82, 0.77), "w": (0.50, 0.50)},
63
  }
64
 
65
+ def _rng(seed:int)->np.random.RandomState: return np.random.RandomState(seed)
 
66
 
67
+ def simulate_step(rng:np.random.RandomState, profile:str, sigma:float, noise_dist:str)->Dict[str,Any]:
68
  base_q, base_z = PROFILES[profile]["base"]
69
  wq, wz = PROFILES[profile]["w"]
70
  if noise_dist == "uniform":
71
  q_noise = rng.uniform(-sigma, sigma)
72
+ z_noise = rng.uniform(-sigma*0.8, sigma*0.8)
73
  else:
74
  q_noise = rng.normal(0, sigma)
75
+ z_noise = rng.normal(0, sigma*0.8)
76
  q = float(np.clip(base_q + wq*q_noise, 0.0, 0.99))
77
  z = float(np.clip(base_z + wz*z_noise, 0.0, 0.99))
78
+ variance = abs(q_noise)+abs(z_noise)
79
+ if variance > 0.15: status="critical"
80
+ elif variance > 0.07: status="perturbed"
81
+ else: status="nominal"
82
+ return {"sigma": round(sigma,6),"":q,"ζ_sync":z,"status":status}
 
 
 
83
 
84
  # ------------------ Schedules -------------------------------
85
+ def build_sigma_series(schedule_type:str, params:Dict[str,Any])->List[float]:
86
+ if schedule_type=="single":
87
+ return [float(params.get("sigma",0.05))]
88
+ elif schedule_type=="ramp":
89
+ return list(np.linspace(float(params.get("start",0.0)), float(params.get("stop",0.3)), int(params.get("steps",10))))
90
+ elif schedule_type=="random":
 
 
91
  r = random.Random(int(params.get("seed",0)))
92
+ return [r.uniform(float(params.get("min",0.0)), float(params.get("max",0.3))) for _ in range(int(params.get("steps",10)))]
93
+ elif schedule_type=="impulse":
94
+ base,spike,at,steps = float(params.get("base",0.05)), float(params.get("spike",0.25)), int(params.get("at",5)), int(params.get("steps",10))
 
 
95
  s=[base]*steps
96
  if 0<=at<len(s): s[at]=spike
97
  return s
98
+ elif schedule_type=="step":
99
+ before,after,at,steps = float(params.get("before",0.05)), float(params.get("after",0.2)), int(params.get("at",5)), int(params.get("steps",10))
100
  s=[before]*steps
101
  for i in range(steps):
102
  if i>=at: s[i]=after
 
110
 
111
  def write_bundle(run_dir:str,config:Dict[str,Any],results:Dict[str,Any])->Tuple[str,str]:
112
  os.makedirs(run_dir,exist_ok=True)
113
+ with open(os.path.join(run_dir,"run.json"),"w") as f: json.dump({"config":config,"results":results},f,indent=2)
 
114
  canonical=json.dumps({"config":config,"results":results},sort_keys=True)
115
  digest=sha512_hex(canonical)
116
  with open(os.path.join(run_dir,"run.sha512"),"w") as f: f.write(digest+"\n")
 
215
 
216
  # ------------------ Launch (stable queue) -------------------
217
  if __name__=="__main__":
218
+ demo.queue(concurrency_limit=1, max_size=5).launch(
219
+ server_name="0.0.0.0",
220
+ server_port=7860,
221
+ show_error=True,
222
+ debug=False
223
+ )