# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: keyapis/telemetry_alerting/v1/keyapis_telemetry_alerting_events_v1.proto, keyapis/telemetry_alerting/v1/keyapis_telemetry_alerting_system_v1.proto, keyapis/telemetry_alerting/v1/keyapis_telemetry_alerting_trigger_v1.proto
# plugin: python-betterproto
from dataclasses import dataclass
from datetime import datetime
from typing import (
    TYPE_CHECKING,
    AsyncIterator,
    Dict,
    List,
    Optional,
)

import betterproto
import grpclib
from betterproto.grpc.grpclib_server import ServiceBase

if TYPE_CHECKING:
    import grpclib.server
    from betterproto.grpc.grpclib_client import MetadataLike
    from grpclib.metadata import Deadline


class TriggerAlertPagingOrderByType(betterproto.Enum):
    """Справочник типов значений сортировки"""

    ORDER_BY_TYPE_UNKNOWN = 0
    """Значение не указано"""

    RAISED_AT = 1
    """По дате возникновения"""

    TRIGGER_ID = 2
    """По идентификатору триггера"""

    TITLE = 3
    """По заголовку триггера"""

    DISABLED = 4
    """По флагу активности триггера"""

    SEVERITY = 5
    """По триггеру"""


class TriggerAlertPagingDirectionType(betterproto.Enum):
    """Справочник типов направлений сортировки"""

    DIRECTION_TYPE_UNKNOWN = 0
    """Значение не указано"""

    DESC = 1
    """От большего к меньшему"""

    ASC = 2
    """От меньшего к большему"""


class TriggerPagingOrderByType(betterproto.Enum):
    """Справочник типов значений сортировки"""

    ORDER_BY_TYPE_UNKNOWN = 0
    """Значение не указано"""

    ID = 1
    """По идентификатору"""

    TITLE = 2
    """По заголовку"""

    DISABLED = 3
    """По флагу активности"""

    SEVERITY = 4
    """По Severity"""


class TriggerPagingDirectionType(betterproto.Enum):
    """Справочник типов направлений сортировки"""

    DIRECTION_TYPE_UNKNOWN = 0
    """Значение не указано"""

    DESC = 1
    """От большего к меньшему"""

    ASC = 2
    """От меньшего к большему"""


class TriggerCriteriaType(betterproto.Enum):
    """
    Справочник типов критерия триггера. Задает параметры критерия, правила
    поиска проблемы и генерации алерта
    """

    CRITERIA_TYPE_UNKNOWN = 0
    """Значение не указано"""

    METRICS_COUNT = 1
    """Количество метрик меньше порога"""

    DEVICE_PERCENT = 2
    """Доля работающих устройств в группе меньше порога"""

    METRICS_VALUE_DEVIATION = 3
    """Превышение порога дисперсии значений показаний"""


class TriggerPeriodType(betterproto.Enum):
    """Справочник типов периодов контроля триггера"""

    PERIOD_TYPE_UNKNOWN = 0
    """Значение не указано"""

    DAY = 1
    """День"""

    MONTH = 2
    """Месяц"""


class TriggerSeverityType(betterproto.Enum):
    """Справочник типов уровней критичности"""

    SEVERITY_TYPE_UNKNOWN = 0
    """Значение не указано"""

    CRITICAL = 1
    """Проблема высшего приоритета. Максимальное оповещение"""

    WARNING = 2
    """Проблема среднего приоритета. Оповещение по внутренним каналам"""

    INFO = 3
    """
    Проблема не влияет на бизнес-пользователя. Отражается во внутренних
    интерфейсах
    """


class TriggerMetricType(betterproto.Enum):
    """Справочник типов энергоресурсов, собираемых ТУ"""

    METRIC_TYPE_UNKNOWN = 0
    """Значение не указано"""

    HOT_WATER = 1
    """Горячая вода"""

    COLD_WATER = 2
    """Холодная вода"""

    HEAT = 3
    """Теплоэнергия"""

    GAS = 4
    """Газ"""

    ELECTRICITY = 5
    """Электроэнергия"""


class TriggerDeviceType(betterproto.Enum):
    """Справочник типов ПУ"""

    DEVICE_TYPE_UNKNOWN = 0
    """Значение не указано"""

    HUB = 1
    """Коммуникационный модуль"""

    PULSE_COUNTER_RECORDER = 2
    """Счетчик импульсов - регистратор"""

    PULSE_COUNTER_RECORDER_LR = 3
    """Счетчик импульсов - регистратор LoRaWAN"""

    METERING_DEVICE = 4
    """Прибор учета расхода коммунальных услуг"""

    INTERFACE_CONVERTER = 5
    """Конвертер интерфейсов"""

    M_BUS_HUB = 6
    """M-BUS концентратор"""

    COMMUNICATION_MODULE_LR = 7
    """Модуль связи LoRaWAN"""

    BASE_LR = 8
    """Базовая станция LoRaWAN"""


class TriggerMetricGroupingType(betterproto.Enum):
    """Типы группировок метрик при вычислении критерия"""

    METRIC_GROUPING_TYPE_UNKNOWN = 0
    """Значение не указано"""

    PARENT_ID = 1
    """
    Признак группировки метрик по идентификаторам родительского
    устройства(идентификатор хаба, проксирующего прибор учета)
    """

    DEVICE_ID = 2
    """
    Признак группировки метрик по идентификатору устройства(устройство, с
    которого получена метрика)
    """


@dataclass(eq=False, repr=False)
class GetTriggerAlertDeviceListRequest(betterproto.Message):
    """Запрос списка устройств, вызвавших появление алерта"""

    id: int = betterproto.int32_field(1)
    """Идентификатор алерта"""

    filter: "TriggerAlertFilter" = betterproto.message_field(2)
    """Параметры фильтрации алерта"""

    paging: "TriggerAlertPaging" = betterproto.message_field(3, group="pagination")
    """Параметры пагинации"""


@dataclass(eq=False, repr=False)
class TriggerAlertPaging(betterproto.Message):
    """Параметры пагинации алертов"""

    order_by_type: "TriggerAlertPagingOrderByType" = betterproto.enum_field(1)
    """Тип значения сортировки. По умолчанию: ORDER_BY_TYPE_RAISED_AT"""

    direction_type: "TriggerAlertPagingDirectionType" = betterproto.enum_field(2)
    """Тип направления сортировки. По умолчанию: DESC"""

    limit: int = betterproto.int32_field(3)
    """
    Сколько элементов нужно получить за раз. Минимальное значение: 1.
    Максимальное значение: 100. По умолчанию: 20. Если значение 0 (не
    передано), то выставляем значение по умолчанию
    """

    offset: int = betterproto.int32_field(4)
    """Начальный сдвиг. По умолчанию: 0"""


@dataclass(eq=False, repr=False)
class AlertDevice(betterproto.Message):
    """Факты сработки алерта на устройствах"""

    alert_id: int = betterproto.int32_field(1)
    """Идентификатор алерта"""

    device_id: int = betterproto.int32_field(2)
    """Идентификатор устройства"""


@dataclass(eq=False, repr=False)
class GetTriggerAlertDeviceListResponse(betterproto.Message):
    """Ответ на запрос списка устройств, вызвавших появление алерта"""

    data: "AlertDevice" = betterproto.message_field(1, group="type")
    """Факт сработки триггера"""


@dataclass(eq=False, repr=False)
class GetTriggerRequest(betterproto.Message):
    """Запрос на чтение триггера"""

    id: int = betterproto.int32_field(1)
    """Идентификатор триггера"""


@dataclass(eq=False, repr=False)
class GetTriggerResponse(betterproto.Message):
    """Ответ на чтение триггера"""

    data: "Trigger" = betterproto.message_field(1, group="type")
    """Триггер"""


@dataclass(eq=False, repr=False)
class GetTriggerAlertListRequest(betterproto.Message):
    """Запрос получения списка алертов"""

    filter: "TriggerAlertFilter" = betterproto.message_field(1)
    """Параметры фильтрации"""

    paging: "TriggerAlertPaging" = betterproto.message_field(2, group="pagination")
    """Параметры пагинации"""


@dataclass(eq=False, repr=False)
class PostTriggerAlertDisableRequest(betterproto.Message):
    """Запрос на погашение алерта"""

    id: int = betterproto.int32_field(1)
    """Идентификатор алерта"""

    closing_reason: Optional[str] = betterproto.message_field(
        2, wraps=betterproto.TYPE_STRING
    )
    """
    Причина принудительного закрытия, заполняется в момент принудительного
    закрытия
    """


@dataclass(eq=False, repr=False)
class PostTriggerAlertDisableResponse(betterproto.Message):
    """Ответ на запрос погашение алерта"""

    pass


@dataclass(eq=False, repr=False)
class GetTriggerAlertCountRequest(betterproto.Message):
    """Запрос получения списка алертов"""

    filter: "TriggerAlertFilter" = betterproto.message_field(1)
    """Параметры фильтрации"""


@dataclass(eq=False, repr=False)
class GetTriggerListRequest(betterproto.Message):
    """Запрос получения списка триггеров"""

    filter: "TriggerFilter" = betterproto.message_field(1)
    """Параметры фильтрации"""

    paging: "TriggerPaging" = betterproto.message_field(2, group="pagination")
    """Параметры пагинации"""


@dataclass(eq=False, repr=False)
class TriggerPaging(betterproto.Message):
    """Параметры пагинации триггеров"""

    order_by_type: "TriggerPagingOrderByType" = betterproto.enum_field(1)
    """Тип значения сортировки. По умолчанию: SEVERITY"""

    direction_type: "TriggerPagingDirectionType" = betterproto.enum_field(2)
    """Тип направления сортировки. По умолчанию: ASC"""

    limit: int = betterproto.int32_field(3)
    """
    Сколько элементов нужно получить за раз. Минимальное значение: 1.
    Максимальное значение: 100. По умолчанию: 20. Если значение 0 (не
    передано), то выставляем значение по умолчанию
    """

    offset: int = betterproto.int32_field(4)
    """Начальный сдвиг. По умолчанию: 0"""


@dataclass(eq=False, repr=False)
class TriggerFilter(betterproto.Message):
    """
    Параметры фильтрации триггеров. При передаче массива в параметр фильтра
    элементы массива работают в выборке через ИЛИ. При передаче нескольких
    разных параметров фильтра они работают в выборке через И
    """

    severity_types: List["TriggerSeverityType"] = betterproto.enum_field(1)
    """По уровням критичности"""

    tags: List[str] = betterproto.string_field(2)
    """По тэгам"""

    metric_types: List["TriggerMetricType"] = betterproto.enum_field(3)
    """По типам метрик триггеров"""

    device_models: List[str] = betterproto.string_field(4)
    """По моделям устройств"""

    device_types: List["TriggerDeviceType"] = betterproto.enum_field(5)
    """По типам устройств"""

    metric_grouping_types: List["TriggerMetricGroupingType"] = betterproto.enum_field(6)
    """По типам группировки метрик"""

    is_disabled: Optional[bool] = betterproto.message_field(
        7, wraps=betterproto.TYPE_BOOL
    )
    """По признаку активности триггера"""


@dataclass(eq=False, repr=False)
class TriggerAlertFilter(betterproto.Message):
    """
    Параметры фильтрации алертов. При передаче массива в параметр фильтра
    элементы массива работают в выборке через ИЛИ. При передаче нескольких
    разных параметров фильтра они работают в выборке через И
    """

    trigger_severity_types: List["TriggerSeverityType"] = betterproto.enum_field(1)
    """По важности триггера"""

    fias_ids: List[str] = betterproto.string_field(2)
    """По ФИАСам"""

    tags: List[str] = betterproto.string_field(3)
    """По тэгам"""

    metric_types: List["TriggerMetricType"] = betterproto.enum_field(4)
    """По типам метрик триггера"""

    device_models: List[str] = betterproto.string_field(5)
    """По моделям устройств"""

    device_types: List["TriggerDeviceType"] = betterproto.enum_field(6)
    """По типам устройств"""

    device_parent_ids: List[int] = betterproto.int32_field(7)
    """По идентификаторам родительских устройств"""

    device_ids: List[int] = betterproto.int32_field(8)
    """По идентификаторам устройств"""

    is_disabled: Optional[bool] = betterproto.message_field(
        9, wraps=betterproto.TYPE_BOOL
    )
    """По признаку активности"""

    after_raised_at: datetime = betterproto.message_field(10)
    """От даты возникновения включительно (>=)"""

    before_raised_at: datetime = betterproto.message_field(11)
    """До даты возникновения (<)"""

    metric_grouping_types: List["TriggerMetricGroupingType"] = betterproto.enum_field(
        12
    )
    """По типам группировки метрик"""

    trigger_ids: List[int] = betterproto.int32_field(13)
    """По идентификаторам триггеров"""


@dataclass(eq=False, repr=False)
class Trigger(betterproto.Message):
    """Триггер"""

    id: int = betterproto.int32_field(1)
    """Идентификатор"""

    changed_at: datetime = betterproto.message_field(2)
    """
    Дата последнего изменения. Заполняется и обновляется сервером. Заполняется
    при создании и изменении. Является версией объекта
    """

    title: str = betterproto.string_field(3)
    """Название триггера"""

    criteria_type: "TriggerCriteriaType" = betterproto.enum_field(4)
    """Тип критерия триггера"""

    threshold: float = betterproto.float_field(5)
    """Пороговое значение триггера"""

    sliding_days: int = betterproto.int32_field(6, group="range")
    """
    Скользящее окно относительно текущего момента. Начало окна это 00:00:00.000
    UTC дня на sliding_days раньше текущего UTC дня. Конец окна - текущий
    момент
    """

    fixed_days_range: "TriggerFixedDaysOfMonth" = betterproto.message_field(
        7, group="range"
    )
    """Границы временного окна, в текущем или предыдущем месяце"""

    period_type: "TriggerPeriodType" = betterproto.enum_field(8)
    """Тип периода контроля"""

    severity_type: "TriggerSeverityType" = betterproto.enum_field(9)
    """
    Уровень критичности проблемы, которая описана критерием. Задает приоритет
    оповещения и реагирования
    """

    is_disabled: bool = betterproto.bool_field(10)
    """
    Флаг активности триггера. При погашении триггера гасятся и все его алерты
    """

    instruction: Optional[str] = betterproto.message_field(
        11, wraps=betterproto.TYPE_STRING
    )
    """Краткая инструкция по решению или эскалации проблемы"""

    metric_types: List["TriggerMetricType"] = betterproto.enum_field(12)
    """
    Фильтрация метрик по типам метрик. При нескольких элементах фильтрация
    срабатывает по значениям через ИЛИ
    """

    device_models: List[str] = betterproto.string_field(13)
    """
    Фильтрация метрик по значению моделей устройств. При нескольких элементах
    фильтрация срабатывает по значениям через ИЛИ
    """

    device_types: List["TriggerDeviceType"] = betterproto.enum_field(14)
    """
    Фильтрация метрик по значению типов устройств. При нескольких элементах
    фильтрация срабатывает по значениям через ИЛИ
    """

    metric_grouping_type: "TriggerMetricGroupingType" = betterproto.enum_field(15)
    """Тип группировки метрик при вычислении критерия"""

    success_threshold: int = betterproto.int32_field(16)
    """
    Количество отрицательных проверок критерия, после которого сервис отключает
    """

    tags: List[str] = betterproto.string_field(17)
    """Произвольные теги для фильтрации и упорядочивания"""

    is_manually_unclosable: bool = betterproto.bool_field(18)
    """Флаг невозможности погашения алертов этого триггера"""

    is_range_metric_expected_freq: bool = betterproto.bool_field(19)
    """
    Признак контроля триггером временного окна выборки метрик на основе поля
    индикатора metric_expected_freq. При значении true: поля
    range.sliding_days, range.fixed_days_range игнорируются при работе
    триггера. При значении true: временное окно выборки метрик при работе
    триггера работает как sliding_days=indicator.metric_expected_freq. При
    значении true: Для случаев indicator.metric_expected_freq=null индикатор
    игнорируется при работе триггера, алерт по нему не создается. При значении
    true: Для случаев indicator.metric_expected_freq<=0 индикатор игнорируется
    при работе триггера, алерт по нему не создается
    """


@dataclass(eq=False, repr=False)
class TriggerFixedDaysOfMonth(betterproto.Message):
    """
    Границы временного окна, в текущем или предыдущем месяце. Если end_day
    текущего месяца не наступил, то берем прошлый месяц
    """

    start_day: int = betterproto.int32_field(1)
    """
    Начало окна времени. Номер дня месяца, момент времени 00:00:00.000 UTC
    """

    end_day: int = betterproto.int32_field(2)
    """
    Конец окна времени. Номер дня месяца, момент времени 00:00:00.000 UTC
    """


@dataclass(eq=False, repr=False)
class TriggerAlert(betterproto.Message):
    """Определение проблемы, сгенерированной по триггеру"""

    id: int = betterproto.int32_field(1)
    """Идентификатор"""

    changed_at: datetime = betterproto.message_field(2)
    """
    Дата последнего изменения. Заполняется и обновляется сервером. Заполняется
    при создании и изменении. Является версией объекта
    """

    trigger_id: int = betterproto.int32_field(3)
    """Породивший триггер"""

    fias_id: str = betterproto.string_field(4)
    """ФИАС"""

    raised_at: datetime = betterproto.message_field(5)
    """Время возникновения алерта"""

    is_disabled: bool = betterproto.bool_field(6)
    """
    Признак погашенного алерта. Проставляется сервисом при
    no_alarm_checks_count >= success_threshold
    """

    no_alarm_checks_count: int = betterproto.int32_field(7)
    """
    Количество проверок триггера для атрибутов данного алерта, при которых не
    выполнился критерий триггера. Увеличивается у непогашенного алерта, если
    при очередной проверке нет срабатывания триггера с атрибутами данного
    алерта. При увеличении проставляется changed_at
    """

    trigger_metric_types: List["TriggerMetricType"] = betterproto.enum_field(8)
    """Типы метрик, по которым были отфильтрованы метрики"""

    device_models: List[str] = betterproto.string_field(9)
    """Модели устройств, по которым были отфильтрованы метрики"""

    trigger_device_types: List["TriggerDeviceType"] = betterproto.enum_field(10)
    """Типы устройств, по которым были отфильтрованы метрики"""

    device_parent_id: Optional[int] = betterproto.message_field(
        11, wraps=betterproto.TYPE_INT32
    )
    """
    Идентификаторы родительских устройств, по которому были сгруппированы
    метрики
    """

    closed_at: datetime = betterproto.message_field(12)
    """
    Время погашения алерта. Пустой или отсутствует у непогашенного активного
    алерта
    """

    is_manually_closed: bool = betterproto.bool_field(13)
    """Признак принудительного погашения алерта"""

    closing_reason: Optional[str] = betterproto.message_field(
        14, wraps=betterproto.TYPE_STRING
    )
    """
    Причина принудительного закрытия, заполняется в момент принудительного
    закрытия
    """

    trigger_metric_grouping_type: "TriggerMetricGroupingType" = betterproto.enum_field(
        15
    )
    """Группировка метрик при вычислении критерия"""


@dataclass(eq=False, repr=False)
class GetTriggerListResponse(betterproto.Message):
    """Ответ на запрос получения списка записей"""

    data: "Trigger" = betterproto.message_field(1, group="type")
    """Триггер"""


@dataclass(eq=False, repr=False)
class GetTriggerAlertListResponse(betterproto.Message):
    """Ответ на запрос получения списка записей"""

    data: "TriggerAlert" = betterproto.message_field(1, group="type")
    """Алерт"""


@dataclass(eq=False, repr=False)
class GetTriggerCountRequest(betterproto.Message):
    """Запрос получения количества записей"""

    filter: "TriggerFilter" = betterproto.message_field(1)
    """Параметры фильтрации"""


@dataclass(eq=False, repr=False)
class GetTriggerCountResponse(betterproto.Message):
    """Ответ на запрос получения количества записей"""

    data: int = betterproto.int32_field(1, group="type")
    """Всего записей"""


@dataclass(eq=False, repr=False)
class GetTriggerAlertCountResponse(betterproto.Message):
    """Ответ на запрос получения количества записей"""

    data: int = betterproto.int32_field(1, group="type")
    """Всего записей"""


@dataclass(eq=False, repr=False)
class PostTriggerRequest(betterproto.Message):
    """Запрос на создание триггера"""

    data: "Trigger" = betterproto.message_field(1)
    """Триггер, который хотим создать или обновить"""


@dataclass(eq=False, repr=False)
class PostTriggerResponse(betterproto.Message):
    """Ответ на запрос сохранения триггера"""

    data: "Trigger" = betterproto.message_field(1, group="type")
    """Триггер, созданный или обновленный"""


@dataclass(eq=False, repr=False)
class DeleteTriggerRequest(betterproto.Message):
    """Запрос на удаление триггера"""

    id: int = betterproto.int32_field(1)
    """Идентификатор триггера"""


@dataclass(eq=False, repr=False)
class DeleteTriggerResponse(betterproto.Message):
    """Ответ на запрос удаление триггера"""

    pass


@dataclass(eq=False, repr=False)
class DeleteTriggerAlertRequest(betterproto.Message):
    """Запрос на удаление алерта"""

    id: int = betterproto.int32_field(1)
    """Идентификатор алерта"""


@dataclass(eq=False, repr=False)
class DeleteTriggerAlertResponse(betterproto.Message):
    """Ответ на запрос удаление алерта"""

    pass


@dataclass(eq=False, repr=False)
class GetSystemStatusResponse(betterproto.Message):
    """Ответ на запрос проверки доступности сервиса"""

    pass


@dataclass(eq=False, repr=False)
class GetSystemStatusRequest(betterproto.Message):
    """Запрос проверки доступности сервиса"""

    pass


@dataclass(eq=False, repr=False)
class AlertingStartTask(betterproto.Message):
    """
    Сигнал на старт генерации алертов. Очередь
    key.telemetry_alerting.alerting_start_task
    """

    pass


class TriggerServiceStub(betterproto.ServiceStub):
    async def post_trigger(
        self,
        post_trigger_request: "PostTriggerRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PostTriggerResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/PostTrigger",
            post_trigger_request,
            PostTriggerResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def get_trigger(
        self,
        get_trigger_request: "GetTriggerRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "GetTriggerResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTrigger",
            get_trigger_request,
            GetTriggerResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def get_trigger_list(
        self,
        get_trigger_list_request: "GetTriggerListRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> AsyncIterator["GetTriggerListResponse"]:
        async for response in self._unary_stream(
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerList",
            get_trigger_list_request,
            GetTriggerListResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        ):
            yield response

    async def get_trigger_count(
        self,
        get_trigger_count_request: "GetTriggerCountRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "GetTriggerCountResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerCount",
            get_trigger_count_request,
            GetTriggerCountResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def delete_trigger(
        self,
        delete_trigger_request: "DeleteTriggerRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "DeleteTriggerResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/DeleteTrigger",
            delete_trigger_request,
            DeleteTriggerResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def get_trigger_alert_list(
        self,
        get_trigger_alert_list_request: "GetTriggerAlertListRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> AsyncIterator["GetTriggerAlertListResponse"]:
        async for response in self._unary_stream(
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerAlertList",
            get_trigger_alert_list_request,
            GetTriggerAlertListResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        ):
            yield response

    async def get_trigger_alert_count(
        self,
        get_trigger_alert_count_request: "GetTriggerAlertCountRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "GetTriggerAlertCountResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerAlertCount",
            get_trigger_alert_count_request,
            GetTriggerAlertCountResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def delete_trigger_alert(
        self,
        delete_trigger_alert_request: "DeleteTriggerAlertRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "DeleteTriggerAlertResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/DeleteTriggerAlert",
            delete_trigger_alert_request,
            DeleteTriggerAlertResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def post_trigger_alert_disable(
        self,
        post_trigger_alert_disable_request: "PostTriggerAlertDisableRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PostTriggerAlertDisableResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.TriggerService/PostTriggerAlertDisable",
            post_trigger_alert_disable_request,
            PostTriggerAlertDisableResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def get_trigger_alert_device_list(
        self,
        get_trigger_alert_device_list_request: "GetTriggerAlertDeviceListRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> AsyncIterator["GetTriggerAlertDeviceListResponse"]:
        async for response in self._unary_stream(
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerAlertDeviceList",
            get_trigger_alert_device_list_request,
            GetTriggerAlertDeviceListResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        ):
            yield response


class SystemServiceStub(betterproto.ServiceStub):
    async def get_system_status(
        self,
        get_system_status_request: "GetSystemStatusRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "GetSystemStatusResponse":
        return await self._unary_unary(
            "/keyapis.telemetry_alerting.v1.SystemService/GetSystemStatus",
            get_system_status_request,
            GetSystemStatusResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )


class TriggerServiceBase(ServiceBase):

    async def post_trigger(
        self, post_trigger_request: "PostTriggerRequest"
    ) -> "PostTriggerResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def get_trigger(
        self, get_trigger_request: "GetTriggerRequest"
    ) -> "GetTriggerResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def get_trigger_list(
        self, get_trigger_list_request: "GetTriggerListRequest"
    ) -> AsyncIterator["GetTriggerListResponse"]:
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def get_trigger_count(
        self, get_trigger_count_request: "GetTriggerCountRequest"
    ) -> "GetTriggerCountResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def delete_trigger(
        self, delete_trigger_request: "DeleteTriggerRequest"
    ) -> "DeleteTriggerResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def get_trigger_alert_list(
        self, get_trigger_alert_list_request: "GetTriggerAlertListRequest"
    ) -> AsyncIterator["GetTriggerAlertListResponse"]:
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def get_trigger_alert_count(
        self, get_trigger_alert_count_request: "GetTriggerAlertCountRequest"
    ) -> "GetTriggerAlertCountResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def delete_trigger_alert(
        self, delete_trigger_alert_request: "DeleteTriggerAlertRequest"
    ) -> "DeleteTriggerAlertResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def post_trigger_alert_disable(
        self, post_trigger_alert_disable_request: "PostTriggerAlertDisableRequest"
    ) -> "PostTriggerAlertDisableResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def get_trigger_alert_device_list(
        self, get_trigger_alert_device_list_request: "GetTriggerAlertDeviceListRequest"
    ) -> AsyncIterator["GetTriggerAlertDeviceListResponse"]:
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_post_trigger(
        self, stream: "grpclib.server.Stream[PostTriggerRequest, PostTriggerResponse]"
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_trigger(request)
        await stream.send_message(response)

    async def __rpc_get_trigger(
        self, stream: "grpclib.server.Stream[GetTriggerRequest, GetTriggerResponse]"
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_trigger(request)
        await stream.send_message(response)

    async def __rpc_get_trigger_list(
        self,
        stream: "grpclib.server.Stream[GetTriggerListRequest, GetTriggerListResponse]",
    ) -> None:
        request = await stream.recv_message()
        await self._call_rpc_handler_server_stream(
            self.get_trigger_list,
            stream,
            request,
        )

    async def __rpc_get_trigger_count(
        self,
        stream: "grpclib.server.Stream[GetTriggerCountRequest, GetTriggerCountResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_trigger_count(request)
        await stream.send_message(response)

    async def __rpc_delete_trigger(
        self,
        stream: "grpclib.server.Stream[DeleteTriggerRequest, DeleteTriggerResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.delete_trigger(request)
        await stream.send_message(response)

    async def __rpc_get_trigger_alert_list(
        self,
        stream: "grpclib.server.Stream[GetTriggerAlertListRequest, GetTriggerAlertListResponse]",
    ) -> None:
        request = await stream.recv_message()
        await self._call_rpc_handler_server_stream(
            self.get_trigger_alert_list,
            stream,
            request,
        )

    async def __rpc_get_trigger_alert_count(
        self,
        stream: "grpclib.server.Stream[GetTriggerAlertCountRequest, GetTriggerAlertCountResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_trigger_alert_count(request)
        await stream.send_message(response)

    async def __rpc_delete_trigger_alert(
        self,
        stream: "grpclib.server.Stream[DeleteTriggerAlertRequest, DeleteTriggerAlertResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.delete_trigger_alert(request)
        await stream.send_message(response)

    async def __rpc_post_trigger_alert_disable(
        self,
        stream: "grpclib.server.Stream[PostTriggerAlertDisableRequest, PostTriggerAlertDisableResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_trigger_alert_disable(request)
        await stream.send_message(response)

    async def __rpc_get_trigger_alert_device_list(
        self,
        stream: "grpclib.server.Stream[GetTriggerAlertDeviceListRequest, GetTriggerAlertDeviceListResponse]",
    ) -> None:
        request = await stream.recv_message()
        await self._call_rpc_handler_server_stream(
            self.get_trigger_alert_device_list,
            stream,
            request,
        )

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/keyapis.telemetry_alerting.v1.TriggerService/PostTrigger": grpclib.const.Handler(
                self.__rpc_post_trigger,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostTriggerRequest,
                PostTriggerResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTrigger": grpclib.const.Handler(
                self.__rpc_get_trigger,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetTriggerRequest,
                GetTriggerResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerList": grpclib.const.Handler(
                self.__rpc_get_trigger_list,
                grpclib.const.Cardinality.UNARY_STREAM,
                GetTriggerListRequest,
                GetTriggerListResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerCount": grpclib.const.Handler(
                self.__rpc_get_trigger_count,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetTriggerCountRequest,
                GetTriggerCountResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/DeleteTrigger": grpclib.const.Handler(
                self.__rpc_delete_trigger,
                grpclib.const.Cardinality.UNARY_UNARY,
                DeleteTriggerRequest,
                DeleteTriggerResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerAlertList": grpclib.const.Handler(
                self.__rpc_get_trigger_alert_list,
                grpclib.const.Cardinality.UNARY_STREAM,
                GetTriggerAlertListRequest,
                GetTriggerAlertListResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerAlertCount": grpclib.const.Handler(
                self.__rpc_get_trigger_alert_count,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetTriggerAlertCountRequest,
                GetTriggerAlertCountResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/DeleteTriggerAlert": grpclib.const.Handler(
                self.__rpc_delete_trigger_alert,
                grpclib.const.Cardinality.UNARY_UNARY,
                DeleteTriggerAlertRequest,
                DeleteTriggerAlertResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/PostTriggerAlertDisable": grpclib.const.Handler(
                self.__rpc_post_trigger_alert_disable,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostTriggerAlertDisableRequest,
                PostTriggerAlertDisableResponse,
            ),
            "/keyapis.telemetry_alerting.v1.TriggerService/GetTriggerAlertDeviceList": grpclib.const.Handler(
                self.__rpc_get_trigger_alert_device_list,
                grpclib.const.Cardinality.UNARY_STREAM,
                GetTriggerAlertDeviceListRequest,
                GetTriggerAlertDeviceListResponse,
            ),
        }


class SystemServiceBase(ServiceBase):

    async def get_system_status(
        self, get_system_status_request: "GetSystemStatusRequest"
    ) -> "GetSystemStatusResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_get_system_status(
        self,
        stream: "grpclib.server.Stream[GetSystemStatusRequest, GetSystemStatusResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_system_status(request)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/keyapis.telemetry_alerting.v1.SystemService/GetSystemStatus": grpclib.const.Handler(
                self.__rpc_get_system_status,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetSystemStatusRequest,
                GetSystemStatusResponse,
            ),
        }