From acd8786fd8a6f0e3acb2ca9e98dd1ebe14bb9961 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Mon, 6 Jul 2026 13:06:12 +0800 Subject: [PATCH] feat: add timing detail scan tools --- scripts/daily_transit_window_scan.py | 128 ++++++++++++++++++++++ scripts/jyotish_engine.py | 10 +- scripts/vimshottari_subperiod_timeline.py | 120 ++++++++++++++++++++ tests/test_timing_detail_tools.py | 78 +++++++++++++ 4 files changed, 335 insertions(+), 1 deletion(-) create mode 100644 scripts/daily_transit_window_scan.py create mode 100644 scripts/vimshottari_subperiod_timeline.py create mode 100644 tests/test_timing_detail_tools.py diff --git a/scripts/daily_transit_window_scan.py b/scripts/daily_transit_window_scan.py new file mode 100644 index 00000000..60d3aa7b --- /dev/null +++ b/scripts/daily_transit_window_scan.py @@ -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()) diff --git a/scripts/jyotish_engine.py b/scripts/jyotish_engine.py index 24163750..2f83deb3 100644 --- a/scripts/jyotish_engine.py +++ b/scripts/jyotish_engine.py @@ -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 diff --git a/scripts/vimshottari_subperiod_timeline.py b/scripts/vimshottari_subperiod_timeline.py new file mode 100644 index 00000000..0dcb09bc --- /dev/null +++ b/scripts/vimshottari_subperiod_timeline.py @@ -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()) diff --git a/tests/test_timing_detail_tools.py b/tests/test_timing_detail_tools.py new file mode 100644 index 00000000..118cabd0 --- /dev/null +++ b/tests/test_timing_detail_tools.py @@ -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)