92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""lock.py — Instancia unica del monitor (RF-22).
|
|
|
|
Usa un fichero `.lock` en el directorio de datos del monitor con el PID y la hora de
|
|
arranque de la instancia viva. Antes de arrancar, si hay un PID vivo, se aborta; si el
|
|
lock es de un proceso muerto (apagado/crash), se recupera.
|
|
|
|
Motivo: la tarea programada y el plugin de OpenCode pueden coincidir; este lock evita
|
|
que convivan dos monitores recogiendo a la vez.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from . import config
|
|
|
|
LOCK_PATH: Path = config.MONITOR_DIR / ".lock"
|
|
|
|
try:
|
|
import psutil # type: ignore
|
|
except Exception: # pragma: no cover - psutil esta disponible en el entorno
|
|
psutil = None
|
|
|
|
|
|
def pid_vivo(pid: int) -> bool:
|
|
"""Devuelve True si el PID corresponde a un proceso vivo (no zombie)."""
|
|
if pid <= 0:
|
|
return False
|
|
if psutil is not None:
|
|
try:
|
|
return psutil.pid_exists(pid) and psutil.Process(pid).status() != psutil.STATUS_ZOMBIE
|
|
except Exception:
|
|
return False
|
|
# Fallback sin psutil.
|
|
try:
|
|
os.kill(pid, 0)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _leer_pid() -> int | None:
|
|
"""Lee el PID del lock actual, o None si no se puede."""
|
|
try:
|
|
with open(LOCK_PATH, "r", encoding="utf-8") as f:
|
|
return int(json.load(f).get("pid", 0)) or None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _escribir_lock() -> None:
|
|
"""Escribe el lock con el PID y la hora de arranque actuales."""
|
|
LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(LOCK_PATH, "w", encoding="utf-8") as f:
|
|
json.dump({"pid": os.getpid(), "arranque": datetime.now().isoformat()}, f, indent=2)
|
|
|
|
|
|
def adquirir() -> bool:
|
|
"""Intenta adquirir el lock. Devuelve False si ya hay una instancia viva."""
|
|
propio = os.getpid()
|
|
if LOCK_PATH.exists():
|
|
pid = _leer_pid()
|
|
if pid and pid_vivo(pid):
|
|
# Viva (propia o ajena) -> no arrancar otra instancia.
|
|
return False
|
|
# Lock caducado (proceso muerto): se elimina y se reintenta.
|
|
try:
|
|
LOCK_PATH.unlink()
|
|
except OSError:
|
|
pass
|
|
_escribir_lock()
|
|
return True
|
|
|
|
|
|
def liberar() -> None:
|
|
"""Elimina el lock solo si pertenece a nuestro propio PID."""
|
|
try:
|
|
if LOCK_PATH.exists() and _leer_pid() == os.getpid():
|
|
LOCK_PATH.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def esta_bloqueado() -> bool:
|
|
"""True si hay una instancia viva (distinta o igual) ocupando el lock."""
|
|
pid = _leer_pid() if LOCK_PATH.exists() else None
|
|
return bool(pid and pid_vivo(pid))
|