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

feat: add timing detail scan tools

Browse Source
This commit is contained in:
732642856
2026-07-06 13:06:12 +08:00
parent 5509dd7a5b
commit acd8786fd8
4 changed files with 335 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/daily_transit_window_scan.py
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""Daily transit / optional double-transit PAC scanner."""
from __future__ import annotations
import argparse
import json
import subprocess
import sys
from datetime import date, timedelta
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
def _run(command: list[str], timeout: int = 90) -> dict[str, Any]:
completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, timeout=timeout, check=False)
if completed.returncode != 0:
return {"status": "error", "stderr": (completed.stderr or completed.stdout).strip()[:500]}
try:
return json.loads(completed.stdout)
except json.JSONDecodeError:
return {"status": "invalid_json", "stdout_excerpt": completed.stdout[:500]}
def _date_range(start: str, end: str) -> list[date]:
cursor = date.fromisoformat(start)
stop = date.fromisoformat(end)
days = []
while cursor <= stop:
days.append(cursor)
cursor += timedelta(days=1)
return days
def _transit(day: date, args: argparse.Namespace) -> dict[str, Any]:
return _run([
sys.executable,
"scripts/jyotish_engine.py",
"transit",
"--year",
str(day.year),
"--month",
str(day.month),
"--day",
str(day.day),
"--planet",
args.planets,
"--tz",
str(args.tz),
])
def _double_transit(day: date, args: argparse.Namespace) -> dict[str, Any]:
return _run([
sys.executable,
"scripts/jyotish_engine.py",
"double-transit-pac",
"--year",
str(args.year),
"--month",
str(args.month),
"--day",
str(args.day),
"--hour",
str(args.hour),
"--minute",
str(args.minute),
"--second",
str(args.second),
"--lat",
str(args.lat),
"--lon",
str(args.lon),
"--tz",
str(args.tz),
"--date",
day.isoformat(),
"--house",
str(args.house),
])
def build_report(args: argparse.Namespace) -> dict[str, Any]:
rows = []
for day in _date_range(args.start, args.end):
row: dict[str, Any] = {"date": day.isoformat(), "transit": _transit(day, args)}
if args.include_double_transit:
row["double_transit_pac"] = _double_transit(day, args)
rows.append(row)
return {
"scope": "daily_transit_window_scan",
"start": args.start,
"end": args.end,
"days": len(rows),
"planets": args.planets.split(","),
"include_double_transit": args.include_double_transit,
"rows": rows,
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--start", required=True)
parser.add_argument("--end", required=True)
parser.add_argument("--planets", default="Jupiter,Saturn,Rahu,Ketu")
parser.add_argument("--tz", type=float, default=0)
parser.add_argument("--include-double-transit", action="store_true")
parser.add_argument("--year", type=int)
parser.add_argument("--month", type=int)
parser.add_argument("--day", type=int)
parser.add_argument("--hour", type=int, default=12)
parser.add_argument("--minute", type=int, default=0)
parser.add_argument("--second", type=int, default=0)
parser.add_argument("--lat", type=float, default=0.0)
parser.add_argument("--lon", type=float, default=0.0)
parser.add_argument("--house", type=int, default=7)
args = parser.parse_args()
if args.include_double_transit and None in (args.year, args.month, args.day):
raise SystemExit("--include-double-transit requires --year --month --day")
print(json.dumps(build_report(args), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/jyotish_engine.py
+9 -1
View File
@@ -47,7 +47,15 @@ import importlib.util
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from typing import Dict, List
from tabulate import tabulate
try:
from tabulate import tabulate
except ModuleNotFoundError: # pragma: no cover - minimal environments
def tabulate(rows, headers=(), tablefmt=None):
lines = []
if headers:
lines.append(" | ".join(str(item) for item in headers))
lines.extend(" | ".join(str(item) for item in row) for row in rows)
return "\n".join(lines)
from life_stage_hook import generate_life_stage_hooks
from capability_evidence_pool import build_capability_evidence_pool_summary
from guided_topic_discovery import build_guided_topics
scripts/vimshottari_subperiod_timeline.py
+120
View File
@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""Expand Vimshottari MD/AD into PD or PrAD timeline from engine AD boundaries."""
from __future__ import annotations
import argparse
import json
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
DASHA_ORDER = ["Ketu", "Venus", "Sun", "Moon", "Mars", "Rahu", "Jupiter", "Saturn", "Mercury"]
DASHA_YEARS = {"Ketu": 7, "Venus": 20, "Sun": 6, "Moon": 10, "Mars": 7, "Rahu": 18, "Jupiter": 16, "Saturn": 19, "Mercury": 17}
def _parse_date(value: str) -> datetime:
return datetime.strptime(value, "%Y-%m-%d")
def _fmt(value: datetime) -> str:
return value.strftime("%Y-%m-%d")
def _ordered_lords(start_lord: str) -> list[str]:
index = DASHA_ORDER.index(start_lord)
return DASHA_ORDER[index:] + DASHA_ORDER[:index]
def _split_period(start: str, end: str, start_lord: str) -> list[dict[str, Any]]:
start_dt = _parse_date(start)
end_dt = _parse_date(end)
total_seconds = (end_dt - start_dt).total_seconds()
cursor = start_dt
periods: list[dict[str, Any]] = []
for lord in _ordered_lords(start_lord):
seconds = total_seconds * DASHA_YEARS[lord] / 120
next_cursor = cursor + timedelta(seconds=seconds)
periods.append({"lord": lord, "start": _fmt(cursor), "end": _fmt(next_cursor)})
cursor = next_cursor
periods[-1]["end"] = end
return periods
def _run_dasha(args: argparse.Namespace) -> dict[str, Any]:
command = [
sys.executable,
"scripts/jyotish_engine.py",
"dasha",
"--year",
str(args.year),
"--month",
str(args.month),
"--day",
str(args.day),
"--hour",
str(args.hour),
"--minute",
str(args.minute),
"--second",
str(args.second),
"--lat",
str(args.lat),
"--lon",
str(args.lon),
"--tz",
str(args.tz),
"--years",
str(args.years),
]
completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, timeout=120, check=False)
if completed.returncode != 0:
raise SystemExit(completed.stderr or completed.stdout)
return json.loads(completed.stdout)
def build_report(args: argparse.Namespace) -> dict[str, Any]:
dasha = _run_dasha(args)
md_reports = []
for md in dasha["timeline"]:
ad_reports = []
for ad in md.get("antardasha_timeline", []):
pd = _split_period(ad["start"], ad["end"], ad["lord"])
item: dict[str, Any] = {"lord": ad["lord"], "start": ad["start"], "end": ad["end"], "pratyantar": pd}
if args.depth == "prad":
for period in item["pratyantar"]:
period["sookshma"] = _split_period(period["start"], period["end"], period["lord"])
ad_reports.append(item)
md_reports.append({"lord": md["lord"], "start": md["start"], "end": md["end"], "antardasha": ad_reports})
return {
"scope": "vimshottari_subperiod_timeline",
"method": "standard proportional subdivision from jyotish_engine dasha AD boundaries",
"depth": args.depth,
"timeline": md_reports,
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--year", type=int, required=True)
parser.add_argument("--month", type=int, required=True)
parser.add_argument("--day", type=int, required=True)
parser.add_argument("--hour", type=int, required=True)
parser.add_argument("--minute", type=int, required=True)
parser.add_argument("--second", type=int, default=0)
parser.add_argument("--lat", type=float, required=True)
parser.add_argument("--lon", type=float, required=True)
parser.add_argument("--tz", type=float, default=0)
parser.add_argument("--years", type=int, default=45)
parser.add_argument("--depth", choices=["pd", "prad"], default="pd")
args = parser.parse_args()
print(json.dumps(build_report(args), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/test_timing_detail_tools.py
+78
View File
@@ -0,0 +1,78 @@
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_vimshottari_subperiod_timeline_expands_ad_to_pd() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vimshottari_subperiod_timeline.py",
"--year",
"2000",
"--month",
"1",
"--day",
"1",
"--hour",
"12",
"--minute",
"0",
"--lat",
"0.0",
"--lon",
"0.0",
"--tz",
"0",
"--years",
"30",
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=180,
check=False,
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
first_ad = report["timeline"][0]["antardasha"][0]
assert len(first_ad["pratyantar"]) == 9
assert first_ad["pratyantar"][0]["start"] == first_ad["start"]
assert first_ad["pratyantar"][-1]["end"] == first_ad["end"]
def test_daily_transit_window_scan_runs_one_day_json() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/daily_transit_window_scan.py",
"--start",
"2027-03-01",
"--end",
"2027-03-01",
"--planets",
"Jupiter,Saturn",
"--tz",
"0",
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["days"] == 1
assert report["rows"][0]["date"] == "2027-03-01"
planets = report["rows"][0]["transit"]["planets"]
if isinstance(planets, dict):
assert "Jupiter" in planets
else:
assert any(item.get("planet") == "Jupiter" for item in planets)