49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""
|
|
CameraGetStatusAction — queries Hikvision ISAPI for device info and system time.
|
|
"""
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.plugins.base import ActionResult, BaseAction, Emitter
|
|
from app.services import camera as camera_service
|
|
|
|
|
|
class CameraGetStatusAction(BaseAction):
|
|
name = "camera_get_status"
|
|
display_name = "Camera: Get Status"
|
|
compatible_categories = ["camera"]
|
|
params_schema = []
|
|
|
|
async def execute(
|
|
self, device: Device, obj: Object, params: dict, emit: Emitter, db=None
|
|
) -> ActionResult:
|
|
await emit(f"[{device.ip}] GET /ISAPI/System/deviceInfo", "info")
|
|
info_result = await camera_service.isapi_get(device, obj, "System/deviceInfo")
|
|
|
|
if not info_result.success:
|
|
await emit(f"[{device.ip}] Error: {info_result.error or info_result.status_code}", "error")
|
|
return ActionResult(
|
|
success=False,
|
|
stdout=info_result.body,
|
|
error=info_result.error or f"HTTP {info_result.status_code}",
|
|
)
|
|
|
|
await emit(f"[{device.ip}] GET /ISAPI/System/time", "info")
|
|
time_result = await camera_service.isapi_get(device, obj, "System/time")
|
|
|
|
data = {
|
|
"device_info": info_result.data,
|
|
"system_time": time_result.data if time_result.success else {},
|
|
}
|
|
|
|
model = info_result.data.get("model", "")
|
|
fw = info_result.data.get("firmwareVersion", "")
|
|
summary = f"model={model} fw={fw}"
|
|
await emit(f"[{device.ip}] {summary}", "info")
|
|
|
|
return ActionResult(
|
|
success=True,
|
|
stdout=info_result.body,
|
|
exit_code=0,
|
|
data=data,
|
|
)
|