40 lines
1.6 KiB
Python
40 lines
1.6 KiB
Python
import asyncio
|
|
import time
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.plugins.base import ActionResult, BaseAction, Emitter
|
|
|
|
|
|
class PingAction(BaseAction):
|
|
name = "ping"
|
|
display_name = "Ping"
|
|
compatible_categories = [] # works for all device types
|
|
params_schema = []
|
|
|
|
async def execute(self, device: Device, obj: Object, params: dict, emit: Emitter) -> ActionResult:
|
|
await emit(f"Pinging {device.ip}...", "info")
|
|
start = time.monotonic()
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"ping", "-c", "1", "-W", "2", device.ip,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5)
|
|
duration = int((time.monotonic() - start) * 1000)
|
|
success = proc.returncode == 0
|
|
msg = f"{device.ip}: {'reachable' if success else 'unreachable'} ({duration}ms)"
|
|
await emit(msg, "info" if success else "warn")
|
|
return ActionResult(
|
|
success=success,
|
|
stdout=stdout.decode(errors="replace").strip(),
|
|
exit_code=proc.returncode or 0,
|
|
data={"ping_ms": duration if success else None},
|
|
)
|
|
except asyncio.TimeoutError:
|
|
await emit(f"{device.ip}: ping timed out", "error")
|
|
return ActionResult(success=False, error="Timeout")
|
|
except Exception as e:
|
|
await emit(f"{device.ip}: error — {e}", "error")
|
|
return ActionResult(success=False, error=str(e))
|