"""The model: replace this file with yours. server.py calls run_model with the case the platform sent (only the inputs manifest.json declares) and reports what it returns. It runs on a thread of its own for each case, so it may compute for as long as it needs, and it may be `async def`. This one is the textbook scaled-distance law, as a worked example: the peak particle velocity at each receiver from the site's own law and the maximum instantaneous charge, PPV = K·(D/√Q)^−β, with its percentiles from the law's log-normal spread. It declines what it cannot answer and says why, which is what the platform checks a model for. """ import math # Below this scaled distance (m/√kg) the law says nothing: the near field. NEAR_FIELD = 2 def normal_quantile(p: float) -> float: """The standard normal quantile (Acklam's rational approximation, relative error below 1.2e-9), for the percentiles.""" a = [-39.69683028665376, 220.9460984245205, -275.9285104469687, 138.357751867269, -30.66479806614716, 2.506628277459239] b = [-54.47609879822406, 161.5858368580409, -155.6989798598866, 66.80131188771972, -13.28068155288572] c = [-0.007784894002430293, -0.3223964580411365, -2.400758277161838, -2.549732539343734, 4.374664141464968, 2.938163982698783] d = [0.007784695709041462, 0.3224671290700398, 2.445134137142996, 3.754408661907416] low = 0.02425 if p < low: q = math.sqrt(-2 * math.log(p)) return (((((c[0] * q + c[1]) * q + c[2]) * q + c[3]) * q + c[4]) * q + c[5]) / ((((d[0] * q + d[1]) * q + d[2]) * q + d[3]) * q + 1) if p > 1 - low: return -normal_quantile(1 - p) q = p - 0.5 r = q * q return ((((((a[0] * r + a[1]) * r + a[2]) * r + a[3]) * r + a[4]) * r + a[5]) * q) / (((((b[0] * r + b[1]) * r + b[2]) * r + b[3]) * r + b[4]) * r + 1) def run_model(case: dict) -> dict: answer = { "outputs": [], "unavailable": [], "warnings": [], "inputsUsed": ["receivers", "calibration.siteLaw", "design.maxInstantaneousCharge"], } request = case["request"] if "receiver_ppv" not in request["outputs"]: return answer named = request.get("receiverIds") receivers = [r for r in case.get("receivers") or [] if named is None or r["id"] in named] law = (case.get("calibration") or {}).get("siteLaw") charge = ((case.get("design") or {}).get("maxInstantaneousCharge") or {}).get("massKg") def decline(receiver_id, reason, detail): answer["unavailable"].append({"kind": "receiver_ppv", "receiverId": receiver_id, "reason": reason, "detail": detail}) if not receivers or not law or not (charge or 0) > 0: decline(None, "missing_input", "Needs receivers, the site law and the maximum instantaneous charge.") return answer if law["metric"] != "ppv" or law["scaling"] != "square_root": decline(None, "insufficient_calibration", "Needs a PPV law on square-root scaled distance.") return answer percentiles = [p for p in request.get("percentiles", [50, 95]) if 0 < p < 100 and p != 50] for receiver in receivers: distance = receiver.get("distanceM") if not (distance or 0) > 0: decline(receiver["id"], "missing_input", "The receiver has no resolved distance.") continue scaled_distance = distance / math.sqrt(charge) if scaled_distance < NEAR_FIELD: decline(receiver["id"], "outside_validity", "Inside the near field, where the law gives no figure.") continue median = law["k"] * scaled_distance ** -law["beta"] fitted = law.get("scaledDistance") or {} below = fitted.get("min") is not None and scaled_distance < fitted["min"] above = fitted.get("max") is not None and scaled_distance > fitted["max"] answer["outputs"].append( { "kind": "receiver_ppv", "receiverId": receiver["id"], "basis": law["basis"], "ppvMmS": { "median": median, "percentiles": [ {"p": p, "value": median * math.exp(law["sigmaLn"] * normal_quantile(p / 100))} for p in percentiles ], }, "scaledDistance": scaled_distance, "validity": ( {"status": "extrapolated", "code": "below_fitted_range" if below else "above_fitted_range"} if below or above else {"status": "within"} ), } ) answer["warnings"].append( { "code": "interval.approximate", "message": "The percentiles use the law’s residual spread alone, not the uncertainty of its fit.", } ) return answer