Conectar uma máquina

# Todos os SOs
pip install psutil requests

# GPU NVIDIA (Windows/Linux)
pip install nvidia-ml-py3

# Windows — temperatura via OpenHardwareMonitor
pip install wmi pywin32

# macOS — temperatura (escolha um)
brew install osx-cpu-temp        # Macs Intel
gem install iStats               # Intel e Apple Silicon

"""
Testes de Stress Agent — coleta métricas locais e envia ao painel central.
Detecta automaticamente Windows / Linux / macOS e usa o método de
temperatura adequado para cada plataforma.

Pré-requisitos:
    pip install psutil requests
    # opcional (GPU NVIDIA):
    pip install nvidia-ml-py3
    # opcional (temperatura no macOS):
    #   gem install iStats        (mais simples, requer Ruby)

Como rodar:
    1) Edite as variáveis CONFIG abaixo (URL e API_KEY)
    2) python stressops_agent.py

Em paralelo, rode sua ferramenta de carga:
    Windows: AIDA64, OCCT, Prime95, FurMark
    Linux:   stress-ng, s-tui, glmark2
    macOS:   stress-ng (brew), Cinebench, Geekbench
"""
import os, re, time, socket, platform, subprocess, uuid
import psutil, requests

# ===== CONFIG =====
INGEST_URL = "https://SEU-DOMINIO/api/public/ingest"
API_KEY    = "COLE_AQUI_A_MESMA_CHAVE_INGEST_API_KEY"
INTERVAL_S = 5                 # frequência de envio (segundos)
TEST_NAME  = os.environ.get("TEST_NAME", "")  # ex: "stress_cpu_30min"
STATUS     = os.environ.get("TEST_STATUS", "running")  # idle|running|passed|failed
# ==================

SYSTEM     = platform.system()           # 'Windows' | 'Linux' | 'Darwin'
IS_WIN     = SYSTEM == "Windows"
IS_LINUX   = SYSTEM == "Linux"
IS_MAC     = SYSTEM == "Darwin"
MACHINE_ID = f"{socket.gethostname()}-{hex(uuid.getnode())[-6:]}"
HOSTNAME   = socket.gethostname()
OS_INFO    = f"{'macOS' if IS_MAC else SYSTEM} {platform.release()} ({platform.machine()})"

# GPU opcional (NVIDIA)
try:
    import pynvml
    pynvml.nvmlInit()
    _gpu = pynvml.nvmlDeviceGetHandleByIndex(0)
except Exception:
    _gpu = None

def read_gpu():
    if not _gpu: return {}
    try:
        util = pynvml.nvmlDeviceGetUtilizationRates(_gpu)
        temp = pynvml.nvmlDeviceGetTemperature(_gpu, pynvml.NVML_TEMPERATURE_GPU)
        mem  = pynvml.nvmlDeviceGetMemoryInfo(_gpu)
        return {"usage": util.gpu, "temp": temp, "vram_used_mb": mem.used // (1024*1024)}
    except Exception:
        return {}

# ---------- Temperatura por SO ----------
def _temp_linux():
    try:
        temps = psutil.sensors_temperatures() or {}
        for key in ("coretemp","k10temp","cpu_thermal","acpitz"):
            if key in temps and temps[key]:
                return temps[key][0].current
    except Exception:
        pass
    return None

def _temp_windows():
    # Tenta OpenHardwareMonitor/LibreHardwareMonitor via WMI (precisa estar rodando).
    try:
        import wmi  # pip install wmi pywin32
        w = wmi.WMI(namespace="root\\OpenHardwareMonitor")
        for s in w.Sensor():
            if s.SensorType == "Temperature" and "CPU" in s.Name:
                return float(s.Value)
    except Exception:
        pass
    # Fallback: psutil (raro funcionar em Windows nativo)
    return _temp_linux()

def _temp_macos():
    # 1) iStats (gem install iStats) — mais confiável
    try:
        out = subprocess.check_output(["istats","cpu","temp","--value-only"],
                                      stderr=subprocess.DEVNULL, timeout=3).decode().strip()
        return float(out)
    except Exception:
        pass
    # 2) osx-cpu-temp (brew install osx-cpu-temp) — Macs Intel
    try:
        out = subprocess.check_output(["osx-cpu-temp"], stderr=subprocess.DEVNULL, timeout=3).decode()
        m = re.search(r"([0-9]+\.[0-9]+)", out)
        if m: return float(m.group(1))
    except Exception:
        pass
    # 3) powermetrics (Apple Silicon) — requer sudo, então só tenta sem prompt
    try:
        out = subprocess.check_output(
            ["sudo","-n","powermetrics","--samplers","smc","-n","1","-i","200"],
            stderr=subprocess.DEVNULL, timeout=4).decode()
        m = re.search(r"CPU die temperature:\s*([0-9.]+)", out)
        if m: return float(m.group(1))
    except Exception:
        pass
    return None

def read_cpu_temp():
    if IS_MAC:   return _temp_macos()
    if IS_WIN:   return _temp_windows()
    return _temp_linux()

def read_battery():
    try:
        b = psutil.sensors_battery()
        if b: return {"charge_pct": round(b.percent, 1)}
    except Exception: pass
    return {}

_last_disk = psutil.disk_io_counters() or None
_last_t = time.time()
def read_disk_io():
    global _last_disk, _last_t
    now = psutil.disk_io_counters()
    t = time.time()
    if not now or not _last_disk:
        _last_disk, _last_t = now, t
        return None
    dt = max(0.001, t - _last_t)
    mbps = ((now.read_bytes + now.write_bytes) - (_last_disk.read_bytes + _last_disk.write_bytes)) / dt / (1024*1024)
    _last_disk, _last_t = now, t
    return round(mbps, 1)

def collect():
    cpu_pct = psutil.cpu_percent(interval=None)
    freq = psutil.cpu_freq()
    return {
        "machine_id": MACHINE_ID,
        "hostname": HOSTNAME,
        "os_info": OS_INFO,
        "test_name": TEST_NAME or None,
        "status": STATUS,
        "cpu":  {"usage": cpu_pct, "temp": read_cpu_temp(), "clock_mhz": freq.current if freq else None},
        "gpu":  read_gpu(),
        "ram":  {"used_pct": psutil.virtual_memory().percent, "errors": 0},
        "disk": {"io_mbps": read_disk_io(), "smart_ok": True},
        "battery": read_battery(),
    }

def main():
    print(f"[Testes de Stress] {OS_INFO}")
    print(f"[Testes de Stress] Enviando como {MACHINE_ID} para {INGEST_URL} a cada {INTERVAL_S}s")
    psutil.cpu_percent(interval=None)  # warm-up
    while True:
        try:
            payload = collect()
            r = requests.post(INGEST_URL, json=payload,
                              headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                              timeout=10)
            if r.status_code != 200:
                print(f"[!] {r.status_code} {r.text[:200]}")
        except Exception as e:
            print(f"[!] erro: {e}")
        time.sleep(INTERVAL_S)

if __name__ == "__main__":
    main()

# Terminal 1 — coleta e envia
python stressops_agent.py

# Opcional: nomear o teste em curso
set TEST_NAME=stress_cpu_30min   # Windows (CMD)
$env:TEST_NAME="stress_cpu_30min" # PowerShell
export TEST_NAME=stress_cpu_30min  # Linux/Mac

# Terminal 2 — ferramenta de carga

# Windows:  AIDA64, OCCT, Prime95, FurMark (GPU)

# Linux:
sudo apt install stress-ng
stress-ng --cpu 0 --cpu-method matrixprod --timeout 30m

# macOS:
brew install stress-ng
stress-ng --cpu 0 --timeout 30m
# (ou rode Cinebench / Geekbench pela GUI)