import divera.api.v1
from divera.api.v2 import alarms as _alarms
from divera.api.v2.pull import All as _All
from divera.models import Alarm as _Alarm
from . import PullModel as _PullModel
from .status import Status
[docs]class Alarm(
_Alarm,
):
[docs] @staticmethod
def get_all(
):
return _PullModel.get_all(
model=Alarm,
)
[docs] def get(
self,
obj=None,
object_id=None,
):
request = _alarms.Get(
obj=obj or (Alarm(id=int(object_id)) if object_id is not None else None) or self,
)
request.process_result = lambda result: Alarm(data=result['data'])
return request
[docs] def confirm(
self,
response_id: str,
response_text: str = None,
):
request = divera.api.v1.Confirm(
alarm_id=self.id,
status_id=response_id,
note=response_text,
)
return request
[docs] def read(
self,
obj=None,
object_id=None,
):
request = _alarms.Read(
obj=obj or (Alarm(id=int(object_id)) if object_id is not None else None) or self,
)
return request
[docs] @staticmethod
def get_response_options(
) -> _All:
"""
Return all the accepted answers for this event.
:return: An endpoint wrapper that returns divera.Status objects that can be used to reply to this alarm.
"""
request = divera.Status.get_all()
def process_result(
result,
):
ids = result['data']['cluster']['statussorting_alarm']
res = []
for id_ in ids:
res.append(
Status(data=result['data']['cluster']['status'][str(id_)]),
)
return res
request.process_result = process_result
return request