102 lines
4.7 KiB
Python
102 lines
4.7 KiB
Python
# Throwaway smoke test: validate the conversion math + log parsers in server.py
|
|
# without needing FastAPI installed (FastAPI is stubbed before the import).
|
|
import sys, types
|
|
|
|
def _install_stub():
|
|
fa = types.ModuleType("fastapi")
|
|
class FastAPI:
|
|
def __init__(self, *a, **k): pass
|
|
def add_middleware(self, *a, **k): pass
|
|
def get(self, *a, **k): return (lambda f: f)
|
|
def post(self, *a, **k): return (lambda f: f)
|
|
def websocket(self, *a, **k): return (lambda f: f)
|
|
def mount(self, *a, **k): pass
|
|
class WebSocket: pass
|
|
class WebSocketDisconnect(Exception): pass
|
|
class HTTPException(Exception):
|
|
def __init__(self, status_code=None, detail=None):
|
|
self.status_code, self.detail = status_code, detail
|
|
super().__init__(detail)
|
|
fa.FastAPI, fa.WebSocket, fa.WebSocketDisconnect, fa.HTTPException = \
|
|
FastAPI, WebSocket, WebSocketDisconnect, HTTPException
|
|
resp = types.ModuleType("fastapi.responses")
|
|
class HTMLResponse: pass
|
|
class FileResponse: pass
|
|
resp.HTMLResponse, resp.FileResponse = HTMLResponse, FileResponse
|
|
mid = types.ModuleType("fastapi.middleware")
|
|
cors = types.ModuleType("fastapi.middleware.cors")
|
|
class CORSMiddleware: pass
|
|
cors.CORSMiddleware, mid.cors = CORSMiddleware, cors
|
|
sf = types.ModuleType("fastapi.staticfiles")
|
|
class StaticFiles:
|
|
def __init__(self, *a, **k): pass
|
|
sf.StaticFiles = StaticFiles
|
|
sys.modules.update({"fastapi": fa, "fastapi.responses": resp,
|
|
"fastapi.middleware": mid, "fastapi.middleware.cors": cors,
|
|
"fastapi.staticfiles": sf})
|
|
|
|
_install_stub()
|
|
import server # noqa: E401 (imported after stubbing fastapi)
|
|
|
|
# A real periodic sample captured on 21/09/2025 (21092025.txt, line 4)
|
|
line = "[2025-09-21 10:41:00] X:53037,Y:0,C:219838,V:4699627,P:7601002,S:19,er:1500"
|
|
|
|
# --- regex must match BOTH the bare and the legacy 'LOG-to prefix ---
|
|
m = server.RE_SAMPLE.match(line)
|
|
assert m, "RE_SAMPLE failed to match a standard sample line"
|
|
m2 = server.RE_SAMPLE.match("LOG:" + line)
|
|
assert m2, "RE_SAMPLE failed to match a LOG:-prefixed sample line"
|
|
pm = server.RE_PARAM.match("LOG:[2025-09-21 10:42:09] Az_deviation=5.00")
|
|
assert pm and abs(float(pm.group("val")) - 5.0) < 1e-9, "RE_PARAM failed"
|
|
|
|
h = server.hub
|
|
h.azdev = 5.0
|
|
|
|
# --- independent reference using the article / parse_log.py formulas ---
|
|
ADC_MAX = 8388607.0; VREF_MV = 2048.0; VREF_V = 2.048
|
|
GAIN_SHUNT = 20.0; R_SHUNT = 0.075; FF = 0.82; VOC_DIV = 2.0
|
|
p, v, c, x, y = (int(m.group(k)) for k in ("p", "v", "c", "x", "y"))
|
|
azdev = 5.0
|
|
voc_raw = v / ADC_MAX * VREF_V * VOC_DIV
|
|
isc_raw = (c / ADC_MAX * (VREF_MV / GAIN_SHUNT)) / 1000.0 / R_SHUNT
|
|
gpoa_20 = round(p / ADC_MAX * (VREF_MV / 2.0), 1)
|
|
voc_exp = round(voc_raw, 3)
|
|
isc_exp = round(isc_raw * 1000, 2)
|
|
pcpv_exp = round(voc_raw * isc_raw * FF * 1000, 2)
|
|
x_base = x - azdev / 90.0 * y
|
|
el_exp = round(90.0 - (x_base - 53000) / 53000 * 80.0, 2)
|
|
az_exp = round(180.0 + (y - 53000) / 53000 * 80.0, 2)
|
|
|
|
h.pyrano = 2.0
|
|
pt20 = h.on_sample(m.groupdict())
|
|
for got, exp, tag in [(pt20["gpoa"], gpoa_20, "G_POA"),
|
|
(pt20["voc"], voc_exp, "Voc"),
|
|
(pt20["isc_ma"], isc_exp, "Isc"),
|
|
(pt20["pcpv_mw"], pcpv_exp, "P_CPV"),
|
|
(pt20["az"], az_exp, "az"),
|
|
(pt20["el"], el_exp, "el")]:
|
|
assert got == exp or abs(got - exp) < 1e-9, (tag, got, exp)
|
|
print("OLD 2.0 mV/Wm2 : G_POA=%6.1f W/m2 Voc=%5.3f V Isc=%7.2f mA P_CPV=%6.2f mW az=%6.2f el=%6.2f"
|
|
% (pt20["gpoa"], pt20["voc"], pt20["isc_ma"], pt20["pcpv_mw"], pt20["az"], pt20["el"]))
|
|
|
|
h.pyrano = 0.2
|
|
pt02 = h.on_sample(m.groupdict())
|
|
print("NEW 0.2 mV/Wm2 : G_POA=%6.1f W/m2 (ratio vs 2.0 = %.3fx)"
|
|
% (pt02["gpoa"], pt02["gpoa"] / pt20["gpoa"]))
|
|
assert abs(pt02["gpoa"] / pt20["gpoa"] - 10.0) < 0.05, "0.2 must be 10x the 2.0 irradiance"
|
|
|
|
# az_dev change must re-derive elevation when the Y axis is non-zero.
|
|
# (The 10:41 sample has Y:0, so it cannot exercise the az_dev coupling;
|
|
# use a SunPos-style sample with a non-zero Y instead.)
|
|
d = {"ts": "2025-09-21 10:43:00", "x": "62299", "y": "9444",
|
|
"c": "200000", "v": "4600000", "p": "7600000", "er": "1500"}
|
|
h.azdev = 5.0
|
|
h.on_sample(d)
|
|
el_at_5 = h.samples[-1]["el"]
|
|
h.azdev = 0.0
|
|
h._recompute_position()
|
|
el_at_0 = h.samples[-1]["el"]
|
|
assert el_at_5 != el_at_0, "az_dev change should alter the elevation estimate when y!=0"
|
|
print("az_dev recompute: el(azdev=5)=%.2f el(azdev=0)=%.2f (differ as expected)" % (el_at_5, el_at_0))
|
|
|
|
print("OK: unit conversion, geometry and log parsers are all consistent.") |