root/Jyotisha
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Recast candidate vargas for active rectification

Browse Source
This commit is contained in:
732642856
2026-07-16 19:29:20 +08:00
parent c059a3ec4f
commit ba275139e4
4 changed files with 148 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/active_rectification_questions.py
+98 -8
View File
@@ -31,7 +31,16 @@ def _parse_time(value: str) -> datetime:
return datetime.strptime(value, "%Y-%m-%d %H:%M")
def _candidate_scan(center: datetime, uncertainty_minutes: int, step_minutes: int) -> dict[str, Any]:
def _candidate_scan(
center: datetime,
uncertainty_minutes: int,
step_minutes: int,
*,
lat: float | None = None,
lon: float | None = None,
tz: float | None = None,
ayanamsa: str = "lahiri",
) -> dict[str, Any]:
start = center - timedelta(minutes=uncertainty_minutes)
end = center + timedelta(minutes=uncertainty_minutes)
total_minutes = int((end - start).total_seconds() // 60)
@@ -46,12 +55,22 @@ def _candidate_scan(center: datetime, uncertainty_minutes: int, step_minutes: in
cluster = "late_candidate_cluster"
else:
cluster = "middle_candidate_cluster"
samples.append({
sample = {
"time": candidate.strftime("%Y-%m-%d %H:%M"),
"offset_minutes": offset,
"cluster": cluster,
"sensitivity_flags": _sensitivity_flags(abs(offset)),
})
}
recast = _candidate_recast(candidate, lat=lat, lon=lon, tz=tz, ayanamsa=ayanamsa)
if recast:
sample.update(recast)
samples.append(sample)
has_true_recast = all("varga_lagna" in sample for sample in samples)
computed_layers = ["time_range", "candidate_cluster", "question_sensitivity_map"]
blocked_layers = ["true_varga_recast", "true_kp_cusp_recast", "true_arudha_recast"]
if has_true_recast:
computed_layers.extend(["true_varga_recast", "true_arudha_recast"])
blocked_layers = ["true_kp_cusp_recast"]
return {
"start": start.strftime("%Y-%m-%d %H:%M"),
"end": end.strftime("%Y-%m-%d %H:%M"),
@@ -62,9 +81,9 @@ def _candidate_scan(center: datetime, uncertainty_minutes: int, step_minutes: in
"sensitivity_summary": {
"method": "range_bucket_scan_v1",
"high_value_layers": ["D9", "D10", "D24", "D30", "D60", "UL", "A7", "A10", "KP_cusp"],
"computed_layers": ["time_range", "candidate_cluster", "question_sensitivity_map"],
"blocked_layers": ["true_varga_recast", "true_kp_cusp_recast", "true_arudha_recast"],
"boundary": "This is a candidate-question scan. True chart-difference recast is the next gate.",
"computed_layers": computed_layers,
"blocked_layers": blocked_layers,
"boundary": "KP cusp recast remains blocked until a validated KP cusp engine is wired into this workflow.",
},
}
@@ -78,7 +97,70 @@ def _sensitivity_flags(abs_offset_minutes: int) -> list[str]:
return flags
def build_questionnaire(birth_time: str, uncertainty_minutes: int = 30, step_minutes: int = 1) -> dict[str, Any]:
def _candidate_recast(
candidate: datetime,
*,
lat: float | None,
lon: float | None,
tz: float | None,
ayanamsa: str,
) -> dict[str, Any] | None:
if lat is None or lon is None or tz is None:
return None
import domain_calculation_service
import jaimini
import varga
chart = domain_calculation_service.compute_chart({
"year": candidate.year,
"month": candidate.month,
"day": candidate.day,
"hour": candidate.hour,
"minute": candidate.minute,
"second": candidate.second,
"lat": lat,
"lon": lon,
"tz": tz,
"ayanamsa": ayanamsa,
})
planet_lons = {
name: data["lon"]
for name, data in chart.get("planets", {}).items()
if name in {"Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu"}
}
asc_lon = chart["ascendant"]["lon"]
vargas = varga.calc_all_vargas(planet_lons, asc_lon, divisions=[9, 10, 24, 30, 60])
arudha = jaimini.calc_arudha_padas(int(asc_lon // 30), planet_lons)
padas = arudha.get("padas", {})
upapada = arudha.get("upapada", {})
return {
"ascendant": {
"lon": round(asc_lon, 6),
"sign": chart["ascendant"].get("sign"),
"degree_in_sign": chart["ascendant"].get("degree_in_sign"),
},
"varga_lagna": {
key: value.get("Ascendant", {})
for key, value in vargas.items()
},
"arudha": {
"A7": padas.get("A7", {}),
"A10": padas.get("A10", {}),
"UL": upapada,
},
}
def build_questionnaire(
birth_time: str,
uncertainty_minutes: int = 30,
step_minutes: int = 1,
*,
lat: float | None = None,
lon: float | None = None,
tz: float | None = None,
ayanamsa: str = "lahiri",
) -> dict[str, Any]:
questions = []
for qid, round_id, domain, sensitivity, window, prompt, yes_bias, no_bias in QUESTION_TEMPLATES:
questions.append({
@@ -99,7 +181,15 @@ def build_questionnaire(birth_time: str, uncertainty_minutes: int = 30, step_min
return {
"scope": "active_birth_time_rectification_questionnaire",
"schema_version": 1,
"candidate_scan": _candidate_scan(_parse_time(birth_time), uncertainty_minutes, step_minutes),
"candidate_scan": _candidate_scan(
_parse_time(birth_time),
uncertainty_minutes,
step_minutes,
lat=lat,
lon=lon,
tz=tz,
ayanamsa=ayanamsa,
),
"workflow": [
"candidate_time_scan",
"varga_arudha_kp_sensitivity_diff",
scripts/jyotish_api_server.py
+12
View File
@@ -6228,12 +6228,24 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
step_minutes = self._get_int(body, 'step_minutes', 1)
if not 1 <= step_minutes <= 30:
raise BadRequest('step_minutes must be between 1 and 30')
lat = lon = tz = None
if any(key in body for key in ('lat', 'lon', 'tz')):
lat = self._get_float(body, 'lat', 0, -90, 90)
lon = self._get_float(body, 'lon', 0, -180, 180)
tz = self._get_float(body, 'tz', 0, -14, 14)
ayanamsa = body.get('ayanamsa', 'lahiri')
if not isinstance(ayanamsa, str):
raise BadRequest('ayanamsa must be a string')
try:
module = _load_local_module('active_rectification_questions')
result = module.build_questionnaire(
birth_time.strip(),
uncertainty_minutes=uncertainty_minutes,
step_minutes=step_minutes,
lat=lat,
lon=lon,
tz=tz,
ayanamsa=ayanamsa,
)
except ValueError as e:
raise BadRequest('birth_time must be YYYY-MM-DD HH:MM') from e
tests/test_active_rectification_api.py
+17
View File
@@ -36,6 +36,23 @@ def test_active_rectification_questions_api_builds_choice_workflow() -> None:
assert "dynamic_candidate_cluster_scoring" in result["workflow"]
def test_active_rectification_questions_api_accepts_location_for_true_recast() -> None:
result = _handler()._compute_active_rectification_questions(
{
"birth_time": "1993-04-17 14:49",
"uncertainty_minutes": 30,
"lat": 36.683333,
"lon": 114.35,
"tz": 8,
}
)
summary = result["candidate_scan"]["sensitivity_summary"]
assert "true_varga_recast" in summary["computed_layers"]
assert "true_arudha_recast" in summary["computed_layers"]
assert "true_varga_recast" not in summary["blocked_layers"]
def test_active_rectification_score_api_returns_rankings_and_next_questions() -> None:
questionnaire = _handler()._compute_active_rectification_questions(
{"birth_time": "1993-04-17 14:49", "uncertainty_minutes": 30}
tests/test_active_rectification_questions.py
+21
View File
@@ -44,3 +44,24 @@ def test_active_rectification_scores_answers_and_selects_next_round() -> None:
assert scored["next_round_questions"]
assert scored["candidate_cluster_rankings"][0]["score"] > scored["candidate_cluster_rankings"][-1]["score"]
assert "final rectification requires scoring answers against actual candidate chart differences" in scored["boundary"]
def test_active_rectification_recasts_candidate_vargas_when_location_is_available() -> None:
report = build_questionnaire(
"1993-04-17 14:49",
uncertainty_minutes=30,
lat=36.683333,
lon=114.35,
tz=8,
)
summary = report["candidate_scan"]["sensitivity_summary"]
assert "true_varga_recast" in summary["computed_layers"]
assert "true_arudha_recast" in summary["computed_layers"]
assert "true_kp_cusp_recast" in summary["blocked_layers"]
assert "true_varga_recast" not in summary["blocked_layers"]
sample = report["candidate_scan"]["samples"][1]
assert sample["varga_lagna"]["D9_Navamsa"]["sign"]
assert sample["varga_lagna"]["D10_Dasamsa"]["sign"]
assert sample["arudha"]["A7"]["sign"]
assert sample["arudha"]["A10"]["sign"]
assert sample["arudha"]["UL"]["sign"]