Source code for semi_cr.core.lab.utils.protocol_helpers

import os
import platform
import socket
import subprocess
import sys
import inspect
from pathlib import Path


def _run_git(repo: Path, args: list[str], strip=True) -> str | None:
    try:
        result = subprocess.check_output(
            ["git", *args],
            cwd=repo,
            text=True,
            stderr=subprocess.DEVNULL,
        )
        return result.strip() if strip else result
    except Exception:
        return None




def get_git_diff(repo_path: str | Path) -> str:
    repo = Path(repo_path).resolve()
    return _run_git(repo, ["diff"], strip=False) or ""





def collect_repo_snapshot(repo_path: str | Path) -> dict:

    repo = Path(repo_path).resolve()
    status = _run_git(repo, ["status", "--porcelain"])
    # diff = _run_git(repo, ["diff"], strip=False)

    status_files = [
        line.strip()
        for line in status.splitlines()
        if line.strip()
    ]

    return {
        "path": str(repo),
        "branch": _run_git(repo, ["rev-parse", "--abbrev-ref", "HEAD"]),  # stripped string
        "commit": _run_git(repo, ["rev-parse", "HEAD"]),  # stripped string
        "commit_short": _run_git(repo, ["rev-parse", "--short", "HEAD"]),
        "status": status_files,  # _run_git(repo, ["status", "--porcelain"], strip=False),
        # "diff": diff.splitlines() if diff else [], # _run_git(repo, ["diff"], strip=False),
        "diff_file": "git_diff.patch",
        "staged_diff": _run_git(repo, ["diff", "--staged"]),
        "is_dirty": _run_git(repo, ["status", "--porcelain"]) not in ("", None),
        "remote": _run_git(repo, ["remote", "get-url", "origin"]),
        }





def collect_machine_snapshot() -> dict:

    return {
            "hostname": socket.gethostname(),
            "platform": platform.platform(),
            "rdp_alias": os.getenv("RDP_ALIAS"),
            "python": sys.version,
            "executable": sys.executable,
            "cwd": str(Path.cwd()),
            "user": os.environ.get("USERNAME") or os.environ.get("USER"),
        }




def collect_runtime_info(repo_path: str | Path) -> dict:

    repo = Path(repo_path).resolve()

    return {
        "machine": collect_machine_snapshot(),
        "repo": collect_repo_snapshot(repo),
    }





def collect_function_snapshot(
        func,
        source_file: str = "protocol_function.py",
) -> dict:

    function_snapshot = {
        "function_name": func.__name__,
        "module": func.__module__,
        "source_file": source_file,
    }
    return function_snapshot





def get_function_source(func) -> str | None:

    try:
        source = inspect.getsource(func)
    except Exception:
        source = None

    return source