diagnostics.py

"""On-demand desktop performance tests. Every ArcPy read stays in its own process."""
import hashlib
import json
import os
import platform
import statistics
import time
from pathlib import Path

from history import SavedLists
from runtime import balanced_groups, run_readers


def fingerprint(project):
    """Compare all saved connection evidence, ignoring timings and item order."""
    value = {k: v for k, v in project.items() if k != "timings"}
    for key in ("rows", "connection_metadata"):
        if key in value:
            value[key] = sorted(value[key], key=lambda row: json.dumps(row, sort_keys=True))
    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=True).encode("utf-8")).hexdigest()


def same_file(first, second):
    return (isinstance(first, dict) and isinstance(second, dict)
            and "size_bytes" in first and "modified_ns" in first
            and all(first.get(k) == second.get(k) for k in ("size_bytes", "modified_ns")))


def project_measurements(projects):
    return [{"path": p["path"], "name": p["name"], "status": p["status"],
             "rows": len(p["rows"]), "matching_connections": p["tsmis_connections"],
             "fingerprint": fingerprint(p), "timings": p.get("timings", {})} for p in projects]


def compare_projects(baseline, projects):
    expected = {p["path"]: p for p in baseline}
    actual = {p["path"]: p for p in projects}
    missing = sorted(expected.keys() - actual.keys())
    unexpected = sorted(actual.keys() - expected.keys())
    different, changed = [], []
    for path in expected.keys() & actual.keys():
        if fingerprint(expected[path]) != fingerprint(actual[path]):
            different.append(path)
        initial = expected[path].get("timings", {}).get("file_before")
        timings = actual[path].get("timings", {})
        if not same_file(initial, timings.get("file_before")) or not same_file(initial, timings.get("file_after")):
            changed.append(path)
    duplicate_count = len(projects) - len(actual)
    return {"results_match": not (missing or unexpected or different or duplicate_count),
            "missing_projects": missing, "unexpected_projects": unexpected,
            "different_projects": sorted(different), "duplicate_projects": duplicate_count,
            "changed_or_unverified_files": sorted(changed)}


def run_pass(owner, python, request, baseline, entry):
    started = time.perf_counter()
    children = []
    def progress(children, snapshots):
        completed = sum(state["completed"] for state in snapshots)
        messages = [state["message"] for state in snapshots if state["running"]]
        with owner.lock:
            owner.state["test_progress"] = {"completed": completed, "total": len(baseline)}
            owner.state["message"] = entry["label"] + (" · " + " / ".join(messages) if messages else "")
    try:
        requests = [dict(request, _timing_pass=True, _project_paths=group) for group in balanced_groups(baseline, entry["readers"])]
        run_readers(owner, python, requests, children, progress)
    finally:
        projects = [p for child in children for p in (child.state["result"] or {}).get("projects", [])]
        entry.update(wall_seconds=time.perf_counter() - started,
                     complete=bool(children) and all((child.state["result"] or {}).get("complete") for child in children),
                     errors=[child.state["error"] for child in children if child.state["error"]],
                     workers=[dict((child.state["result"] or {}).get("timings", {}),
                                   last_read=(child.state["result"] or {}).get("last_read", {}),
                                   last_progress=(child.state["result"] or {}).get("last_progress", "")) for child in children],
                     projects=project_measurements(projects))
        if request.get("_probe_only"):
            entry["files"] = {path: stamp for child in children for path, stamp in (child.state["result"] or {}).get("file_probe", {}).items()}
            entry["warnings"] = [warning for child in children for warning in (child.state["result"] or {}).get("warnings", [])]
        else:
            entry.update(compare_projects(baseline, projects))
        if owner.cancelled.is_set():
            entry["complete"] = False
    return entry


def cache_probe(baseline, folder, file_probe):
    """Measure a scratch saved-list load and stat checks; do not enable incremental scans."""
    ordinary = {k: v for k, v in baseline.items() if k not in {"performance", "timings"} and not k.startswith("_")}
    ordinary["projects"] = [{k: v for k, v in p.items() if k not in {"timings", "connection_metadata"}} for p in baseline["projects"]]
    ordinary["diagnostic_scan"] = False
    store = SavedLists(Path(folder) / "cache-probe")
    started = time.perf_counter()
    store.save(ordinary)
    write_seconds = time.perf_counter() - started
    loads = []
    for _ in range(3):
        started = time.perf_counter()
        restored = store.load(baseline["root"])
        loads.append(time.perf_counter() - started)
        if restored["result"] != ordinary:
            raise ValueError("Saved-list round trip changed results")
    reusable = []
    for project in baseline["projects"]:
        timings = project.get("timings", {})
        stamp = timings.get("file_before")
        if (not project.get("read_issues") and not project.get("open_error")
                and same_file(stamp, timings.get("file_after")) and same_file(stamp, file_probe["files"].get(project["path"]))):
            reusable.append(project)
    return {"status": "measured", "write_seconds": write_seconds, "load_seconds": loads,
            "median_load_seconds": statistics.median(loads),
            "file_check_seconds": sum(worker.get("file_checks_seconds", 0) for worker in file_probe["workers"]),
            "potentially_reusable_projects": len(reusable), "projects_requiring_reread": len(baseline["projects"]) - len(reusable),
            "estimated_project_read_seconds_avoided": sum(p.get("timings", {}).get("total_seconds", 0) for p in reusable),
            "note": "Scratch cache round trip and size/modified-time checks only; no incremental scan was run. "
                    "Read-time savings are an estimate. Unchanged project files do not prove external connections are unchanged."}


def storage_context(root, projects):
    drive_type = "unknown"
    if os.name == "nt":
        import ctypes
        function = ctypes.windll.kernel32.GetDriveTypeW
        function.argtypes, function.restype = [ctypes.c_wchar_p], ctypes.c_uint
        drive_type = {2: "removable", 3: "local", 4: "network", 5: "optical", 6: "ramdisk"}.get(function(Path(root).anchor), "unknown")
    top_level = [p for p in projects if Path(p["path"]).parent == Path(root)]
    return {"drive_type": drive_type, "top_level_projects": len(top_level), "nested_projects": len(projects) - len(top_level),
            "initial_nested_read_seconds": sum(p.get("timings", {}).get("total_seconds", 0) for p in projects if p not in top_level),
            "offline_or_recall_files": sum(bool(p.get("timings", {}).get("file_before", {}).get("offline")
                                               or p.get("timings", {}).get("file_before", {}).get("recall_on_access")) for p in projects),
            "note": "Nested read time estimates work avoided by disabling Subfolders; it is not a separate scan. "
                    "Drive type and placeholder flags are hints. No files were moved and no local-versus-network copy benchmark was run."}


def run_tests(owner, python, request, folder):
    baseline = owner.state["result"]
    projects = baseline["projects"]
    report = baseline["performance"]
    report.update(machine={"os": platform.platform(), "logical_cpus": os.cpu_count()},
                  storage=storage_context(request["root"], projects),
                  notes=["Five reads: initial sequential, sequential, two readers, two readers, sequential. "
                         "Comparison passes use the same file manifest, include fresh interpreter startup, and exclude folder discovery.",
                         "Initial scan is not guaranteed cold. Later reads may benefit from ArcGIS, OS and service caches. "
                         "No caches are flushed. Two trials per mode indicate variability, not statistical certainty.",
                         "Every pass reads both connectionProperties and CIM, including joined sources and standalone tables. "
                         "Only saved metadata is inspected; no project is saved and no feature data is queried.",
                         "Matching results can include repeated read errors. Review unreadable projects before interpreting speed.",
                         "Diagnostic timings include metadata collection overhead. First-event time includes up to one polling interval."])
    report["passes"].append({"id": "initial", "label": "Initial sequential scan", "readers": 1,
                             "wall_seconds": baseline.get("timings", {}).get("wall_seconds", 0),
                             "complete": baseline.get("complete", False), "workers": [baseline.get("timings", {})],
                             "projects": project_measurements(projects)})
    if not projects:
        report.update(status="skipped", summary="No projects found. Speed comparisons were skipped.")
        return
    report["projects_with_read_issues"] = sum(bool(p.get("read_issues") or p.get("open_error")) for p in projects)
    schedule = [("sequential_1", 1), ("parallel_1", 2), ("parallel_2", 2), ("sequential_2", 1)]
    if len(projects) < 2:
        schedule = [("sequential_1", 1), ("sequential_2", 1)]
        report["parallel_skipped"] = "At least two projects are needed to compare concurrent readers."
    for index, (identifier, readers) in enumerate(schedule, 2):
        if owner.cancelled.is_set():
            return
        entry = {"id": identifier, "label": f"Test {index}/{len(schedule) + 1}: " + ("two readers" if readers == 2 else "sequential"),
                 "readers": readers, "complete": False}
        with owner.lock:
            report["passes"].append(entry)
        run_pass(owner, python, request, projects, entry)
        if owner.cancelled.is_set():
            return
        if not entry["complete"]:
            with owner.lock:
                owner.state["error"] = "A comparison reader failed. Previous saved list kept; save the diagnostic ZIP for details."
            return
    report["comparison"] = {"results_match": all(p["results_match"] for p in report["passes"][1:]),
                            "changed_or_unverified_files": sorted({path for p in report["passes"][1:] for path in p["changed_or_unverified_files"]})}
    # Network discovery/stat calls need the same cancellation and timeout as ArcPy.
    probe = {"label": "Checking folder and saved-list reuse", "readers": 1, "complete": False}
    report["file_probe"] = probe
    run_pass(owner, python, dict(request, _probe_only=True), projects, probe)
    if owner.cancelled.is_set():
        return
    final_files, warnings = probe.get("files", {}), probe.get("warnings", [])
    original_files = {p["path"] for p in projects}
    report["manifest_check"] = {
                                "added": sorted(set(final_files) - original_files),
                                "removed": sorted(original_files - set(final_files)), "warnings": warnings}
    if not probe["complete"] or set(final_files) != original_files or warnings:
        with owner.lock:
            owner.state["error"] = "The project folder changed or could not be fully checked. Previous saved list kept; save the diagnostic ZIP."
        return
    comparison = report["comparison"]
    comparison["changed_or_unverified_files"] = sorted(set(comparison["changed_or_unverified_files"]) | {
        p["path"] for p in projects if not same_file(p.get("timings", {}).get("file_before"), final_files.get(p["path"]))})
    try:
        report["cache_probe"] = cache_probe(baseline, folder, probe)
    except (OSError, ValueError, KeyError, TypeError) as exc:
        report["cache_probe"] = {"status": "failed", "error": type(exc).__name__}
        with owner.lock:
            owner.state["error"] = "The saved-list test failed. Previous saved list kept; save the diagnostic ZIP for details."
        return
    if not comparison["results_match"] or comparison["changed_or_unverified_files"]:
        with owner.lock:
            owner.state["error"] = "Test results changed or files could not be verified. Previous saved list kept; save the diagnostic ZIP."
        return
    sequential = [p["wall_seconds"] for p in report["passes"] if p["id"].startswith("sequential_")]
    parallel = [p["wall_seconds"] for p in report["passes"] if p["id"].startswith("parallel_")]
    comparison["sequential_seconds"] = sequential
    comparison["parallel_seconds"] = parallel
    if parallel:
        serial_median, parallel_median = statistics.median(sequential), statistics.median(parallel)
        comparison.update(sequential_median_seconds=serial_median, parallel_median_seconds=parallel_median,
                          speedup_ratio=serial_median / parallel_median if parallel_median else None)
        summary = f"Same results across all reads. Sequential: {serial_median:.1f}s; two readers: {parallel_median:.1f}s (medians)."
    else:
        summary = "Repeated reads returned the same results. Two-reader comparison needs at least two projects."
    if report["projects_with_read_issues"]:
        summary += " Some projects have read issues."
    with owner.lock:
        report.update(status="completed", summary=summary)
        owner.state["message"] = "Diagnostic test complete"


def report_text(result):
    report = result.get("performance", {})
    lines = ["TSMIS Project Inspector - performance diagnostics", "",
             "Status: " + report.get("status", "incomplete"), report.get("summary", ""), ""]
    entries = report.get("passes", []) + ([report["file_probe"]] if report.get("file_probe") else [])
    for entry in entries:
        lines.append(f"{entry['label']}: {entry.get('wall_seconds', 0):.3f}s; complete={entry.get('complete', False)}")
        if "results_match" in entry:
            lines.append("  Matches initial results: " + str(entry["results_match"]))
        for error in entry.get("errors", []):
            lines.append("  " + error)
        if not entry.get("complete"):
            for worker in entry.get("workers", []):
                stage = worker.get("last_read", {})
                detail = " · ".join(str(stage[k]) for k in ("stage", "project", "map", "layer") if stage.get(k))
                lines.append("  Last stage entered: " + (detail or worker.get("last_progress", "")))
    if result.get("error"):
        lines.extend(["", "Scan error: " + result["error"]])
        lines.append("Last progress: " + result.get("last_progress", ""))
        lines.append("Last read stage: " + json.dumps(result.get("last_read", {}), ensure_ascii=False))
    cache = report.get("cache_probe", {})
    if cache.get("status") == "measured":
        lines.extend(["", f"Scratch saved-list load (median): {cache['median_load_seconds']:.4f}s",
                      f"File checks: {cache['file_check_seconds']:.4f}s; potentially reusable: {cache['potentially_reusable_projects']} project(s).",
                      cache["note"]])
    lines.extend(["", "Initial scan: slowest projects"])
    for project in sorted(result.get("projects", []), key=lambda p: p.get("timings", {}).get("total_seconds", 0), reverse=True)[:10]:
        timings = project.get("timings", {})
        layers = sum(layer.get("total_seconds", 0) for layer in timings.get("layers", []))
        lines.append(f"  {timings.get('total_seconds', 0):.3f}s total; {timings.get('open_seconds', 0):.3f}s opening; {layers:.3f}s layers/tables - {project['path']}")
        if "release_seconds" in timings:
            lines.append(f"    Cleanup: {timings.get('release_seconds', 0):.3f}s releasing objects; {timings.get('gc_seconds', 0):.3f}s garbage collection.")
    lines.extend(["", *report.get("notes", []), report.get("storage", {}).get("note", ""), "",
                  "Detailed stage timings and comparison evidence: diagnostics.json. Timings in CSV: project-timings.csv and layer-timings.csv.",
                  "Normal Refresh uses up to two readers. This test does not enable incremental caching."])
    return "\n".join(lines) + "\n"


def write_report(archive, result):
    from core import csv_text
    report = result["performance"]
    passes = report.get("passes") or [{"label": "Initial scan (partial)", "readers": 1,
                                       "workers": [result.get("timings", {})],
                                       "projects": project_measurements(result.get("projects", []))}]
    if report.get("file_probe"):
        passes = passes + [report["file_probe"]]
    stages, projects, layers = [], [], []
    for entry in passes:
        for index, timings in enumerate(entry.get("workers", []), 1):
            stages.append(dict(timings, phase=entry["label"], reader=index, readers=entry["readers"],
                               phase_seconds=entry.get("wall_seconds"), complete=entry.get("complete", False)))
        for project in entry.get("projects", []):
            timings = project["timings"]
            project_layers = timings.get("layers", [])
            projects.append(dict(timings, phase=entry["label"], project=project["path"], status=project["status"],
                                 layer_count=len(project_layers), layer_seconds=sum(row.get("total_seconds", 0) for row in project_layers)))
            layers.extend(dict(row, phase=entry["label"], project=project["path"]) for row in project_layers)
    archive.writestr("performance.txt", report_text(result))
    archive.writestr("pass-timings.csv", csv_text(stages, [("phase", "Test"), ("reader", "Reader"), ("readers", "Concurrent readers"),
        ("complete", "Completed"), ("phase_seconds", "Entire test seconds"), ("wall_seconds", "Reader wall seconds"),
        ("first_event_seconds", "Launch to first event seconds"), ("discovery_seconds", "Folder discovery seconds"),
        ("file_checks_seconds", "Check source files seconds"),
        ("arcpy_import_seconds", "ArcPy import seconds"), ("arcgis_initialize_seconds", "ArcGIS initialization seconds"),
        ("projects_seconds", "All project reads seconds")]))
    archive.writestr("project-timings.csv", csv_text(projects, [("phase", "Test"), ("project", "Project path"), ("status", "Read status"),
        ("total_seconds", "Total seconds"), ("open_seconds", "Open project seconds"), ("list_maps_seconds", "List maps seconds"),
        ("list_items_seconds", "List layers and tables seconds"), ("layer_count", "Layers and tables read"),
        ("layer_seconds", "Read layers and tables seconds"), ("cleanup_seconds", "Cleanup seconds"),
        ("release_seconds", "Release project objects seconds"), ("gc_seconds", "Garbage collection seconds"),
        ("file_stat_seconds", "File checks seconds")]))
    archive.writestr("layer-timings.csv", csv_text(layers, [("phase", "Test"), ("project", "Project path"), ("map", "Map"),
        ("layer", "Layer / table"), ("kind", "Type"), ("total_seconds", "Total seconds"),
        ("connection_properties_seconds", "Connection properties seconds"), ("cim_definition_seconds", "Get CIM seconds"),
        ("cim_extract_seconds", "Extract CIM connections seconds"), ("redaction_seconds", "Redaction seconds"),
        ("interpret_seconds", "Interpret connections seconds"), ("data_source_seconds", "Fallback data source seconds")]))