Создание обработчика ИБП

Вы можете добавить в DCImanager 6 собственный обработчик ИБП. Для этого создайте код обработчика и загрузите его в платформу.

Подготовка окружения

Обработчик ИБП должен быть написан на языке Python. Рекомендуем использовать версию Python не ниже 3.9.

Вы можете создать обработчик на основе существующего проекта. Чтобы скопировать проект, подключитесь к серверу-локации DCImanager 6 по SSH и выполните команду:

docker cp eservice_handler:/opt/ispsystem/equip_handler ./

При создании обработчика могут быть полезны директории проекта:

/common — общие вспомогательные классы и функции;
/ups_common — вспомогательные классы и функции для работы с ИБП;
/ups_common/handlers — обработчики ИБП.

Требуемые библиотеки Python и их версии вы можете посмотреть в файле проекта requirements.txt. Чтобы установить нужные библиотеки, выполните команду:

pip3 install -r requirements.txt

Для проверки типов данных в проекте рекомендуем использовать анализатор mypy.

Создание обработчика

Класс для работы с ИБП

Класс ИБП наследуется от класса:

BaseSnmpUps — для устройств, работающих через протокол SNMP;
BaseUps — для остальных устройств.

Класс BaseSnmpUps содержит методы взаимодействия с ИБП по протоколу SNMP:

snmp_get — выполнить запрос чтения по определённому OID;
snmp_set — выполнить модифицирующий запрос по определённому OID;
snmp_walk — выполнить запрос, результатом которого будет список.

Пример описания класса

class ExampleHandlerSnmp(BaseSnmpUps):
    """Example handler class."""

    def __init__(self, ups_data: UpsSnmpData):
        """__init__.

        Args:
            ups_data (UpsSnmpData): ups connection data
        """
        super().__init__(ups_data)

Методы для работы с ИБП

Чтобы DCImanager 6 мог взаимодействовать с ИБП, переопределите методы класса BaseUps:

status — опрос ИБП;
port_up — включение порта;
port_down — выключение порта;
statistic — сбор статистики.

Служба работы с оборудованием использует не "сырые" json-данные, а их объектное представление. Экземляры класса UpsView — представление ИБП. Имеет следующие параметры:

load_percentage — нагрузка, %;
charge_time — заряд, мин;
input_power — выходная мощность, кВт;
output_power — потребляемая мощность, кВт.

Класс UpsData наследуется от базового класса ConnectionParams и определяет данные для доступа к ИБП.

Каждый файл-обработчик должен содержать функцию make_handler. Эта функция создаёт объект обработчика:

Пример функции

def make_handler(ups_data: UpsData) -> BaseSnmpUps:
    """
    Returns APC UPS handler object
    :param ups_data: Snmp UPS data object
    :return:
    """
    return ApcSnmp(ups_data)

Пример кода обработчика

Пример кода

"""APC UPS handler"""
 
from ups_common.base_snmp_ups import BaseSnmpUps
from ups_common.connection import UpsData
from ups_common.helper import UpsView
from common.base_snmp import SnmpResponse
from common.custom_exception import RequestError
from common.logger import get_logger
 
HANDLER_APC_SNMP_LOGGER_NAME = "apc_ups_snmp"
 
OID_MODEL = "1.3.6.1.4.1.318.1.1.1.1.1.1.0"
 
OIDS_LOAD_PERCENTAGE = [
    "1.3.6.1.4.1.318.1.1.1.4.2.3",
    "1.3.6.1.4.1.318.1.1.1.4.2.3.0",  # MGE Galaxy 300, Smart-UPS SRT 10000
]
 
# Time in snmp ticks format
OIDS_CHARGE_TIME = [
    "1.3.6.1.4.1.318.1.1.1.2.2.3",
    "1.3.6.1.4.1.318.1.1.1.2.2.3.0",  # MGE Galaxy 300, Smart-UPS SRT 10000
]
OID_INPUT_POWER = "1.3.6.1.4.1.318.1.1.1.3.2.12"  # Input power in watts
OID_INPUT_POWER_SMART_UPS_SRT_10K = "1.3.6.1.2.1.33.1.3.3.1.5.1"  # Input power in watts
 
OID_MAIN_INPUT_VOLTAGE_PHASE_1 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.3.1.1.1"
OID_MAIN_INPUT_VOLTAGE_PHASE_2 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.3.1.1.2"
OID_MAIN_INPUT_VOLTAGE_PHASE_3 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.3.1.1.3"
 
OID_BYPASS_INPUT_VOLTAGE_PHASE_1 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.3.2.1.1"
OID_BYPASS_INPUT_VOLTAGE_PHASE_2 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.3.2.1.2"
OID_BYPASS_INPUT_VOLTAGE_PHASE_3 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.3.2.1.3"
 
OID_MAIN_INPUT_CURRENT_PHASE_1 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.6.1.1.1"
OID_MAIN_INPUT_CURRENT_PHASE_2 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.6.1.1.2"
OID_MAIN_INPUT_CURRENT_PHASE_3 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.6.1.1.3"
 
OID_BYPASS_INPUT_CURRENT_PHASE_1 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.6.2.1.1"
OID_BYPASS_INPUT_CURRENT_PHASE_2 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.6.2.1.2"
OID_BYPASS_INPUT_CURRENT_PHASE_3 = "1.3.6.1.4.1.318.1.1.1.9.2.3.1.6.2.1.3"
 
 
# Output power in watts
OIDS_OUTPUT_POWER = [
    "1.3.6.1.4.1.318.1.1.1.4.2.8",
    "1.3.6.1.4.1.318.1.1.1.4.2.8.0",  # MGE Galaxy 300, Smart-UPS SRT 10000
]
 
 
def make_handler(ups_data: UpsData) -> BaseSnmpUps:
    """
    Returns APC UPS handler object
    :param ups_data: Snmp UPS data object
    :return:
    """
    return ApcSnmp(ups_data)
 
 
def get_power_value(voltage: float, current: float) -> float:
    """Returns kW power value from voltage + current(value in 0.1 amps)
 
    Formula: kWatts = (Voltage * Amperage) / 1000
    """
 
    return (voltage * (current / 10)) / 1000
 
 
class ApcSnmp(BaseSnmpUps):
    """Ups ApcSnmp work class"""
 
    def __init__(self, ups_data: UpsData):
        super().__init__(ups_data)
        self._logger = get_logger(HANDLER_APC_SNMP_LOGGER_NAME)
 
    def __snmp_get_first_available_oid(self, oids: list[str]) -> SnmpResponse:
        """Returns result of first successful `snmp_get` or raises `RequestError`"""
 
        err = RequestError("")
 
        for oid in oids:
            try:
                return self.snmp_get(oid)
            except RequestError as e:
                self._logger.warning(f"Failed to get oid {oid}: {e}")
                err = e
 
        self._logger.exception(err)
        raise err
 
    def __stats_load_percentage(self) -> int:
        """Returns power load percentage"""
 
        return self.__snmp_get_first_available_oid(OIDS_LOAD_PERCENTAGE).value_int
 
    def __stats_charge_time(self) -> int:
        """Returns charge time in minutes"""
 
        return int(self.__snmp_get_first_available_oid(OIDS_CHARGE_TIME).get_tick_to_int_value() / 60)
 
    def __stats_input_power_mge_galaxy_300(self) -> float:
        return sum(
            (
                get_power_value(
                    voltage=self.snmp_get(OID_MAIN_INPUT_VOLTAGE_PHASE_1).value_float,
                    current=self.snmp_get(OID_MAIN_INPUT_CURRENT_PHASE_1).value_float,
                ),
                get_power_value(
                    voltage=self.snmp_get(OID_MAIN_INPUT_VOLTAGE_PHASE_2).value_float,
                    current=self.snmp_get(OID_MAIN_INPUT_CURRENT_PHASE_2).value_float,
                ),
                get_power_value(
                    voltage=self.snmp_get(OID_MAIN_INPUT_VOLTAGE_PHASE_3).value_float,
                    current=self.snmp_get(OID_MAIN_INPUT_CURRENT_PHASE_3).value_float,
                ),
                get_power_value(
                    voltage=self.snmp_get(OID_BYPASS_INPUT_VOLTAGE_PHASE_1).value_float,
                    current=self.snmp_get(OID_BYPASS_INPUT_CURRENT_PHASE_1).value_float,
                ),
                get_power_value(
                    voltage=self.snmp_get(OID_BYPASS_INPUT_VOLTAGE_PHASE_2).value_float,
                    current=self.snmp_get(OID_BYPASS_INPUT_CURRENT_PHASE_2).value_float,
                ),
                get_power_value(
                    voltage=self.snmp_get(OID_BYPASS_INPUT_VOLTAGE_PHASE_3).value_float,
                    current=self.snmp_get(OID_BYPASS_INPUT_CURRENT_PHASE_3).value_float,
                ),
            )
        )
 
    def __stats_input_power(self) -> float:
        """Returns input power in Kwh"""
 
        ups_model = self.snmp_get(OID_MODEL).value
 
        if "MGE Galaxy 300" in ups_model:
            return self.__stats_input_power_mge_galaxy_300()
 
        if ups_model == "Smart-UPS SRT 10000":
            return self.snmp_get(OID_INPUT_POWER_SMART_UPS_SRT_10K).value_float / 1000
 
        return self.snmp_get(OID_INPUT_POWER).value_int / 1000
 
    def __stats_output_power(self) -> float:
        """Returns output power in Kwh"""
 
        return self.__snmp_get_first_available_oid(OIDS_OUTPUT_POWER).value_int / 1000
 
    def status(self) -> UpsView:
        """
        :return: UpsView
        """
        self._logger.info("Start to poll UPS APC device status")
 
        load_percentage = self.__stats_load_percentage()
        self._logger.info(f"Load percentage: '{load_percentage}'")
 
        charge_time = self.__stats_charge_time()
        self._logger.info(f"Charge time: '{charge_time}'")
 
        input_power = self.__stats_input_power()
        self._logger.info(f"Input power: '{input_power}'")
 
        output_power = self.__stats_output_power()
        self._logger.info(f"Output power: '{output_power}'")
 
        return UpsView(
            load_percentage=load_percentage,
            charge_time=charge_time,
            input_power=input_power,
            output_power=output_power,
        )

Загрузка обработчика в платформу

Чтобы загрузить обработчик в платформу:

Создайте директорию со следующей структурой:
```
handler_dir/
├── __init__.py
└── my_handler.py
```
Пояснения

handler_dir — имя директории. В платформе не должно быть директории обработчика с таким же именем.
__init__.py — файл инициализации. Если такого файла нет, создайте пустой файл с этим именем.
my_handler.py — файл обработчика
Создайте архив tar.gz с этой директорией:
```
tar -czvf custom_handler.tar.gz handler_dir
```
Пояснения к команде

custom_handler.tar.gz — имя архива
handler_dir –- имя созданной директории с обработчиком
Авторизуйтесь в DCImanager 6 с правами администратора:
```
curl -o- -k https://example.com/api/auth/v4/public/token \
    -H "isp-box-instance: true" \
    -d '{
        "email": "<admin_email>",
        "password": "<admin_pass>"
    }'
```
Пояснения к команде

example.com — доменное имя сервера c DCImanager 6
<admin_email> — email администратора DCImanager 6
<admin_pass> — пароль администратора DCImanager 6

В ответ придёт сообщение вида:
```
{"id":"24","token":"24-cee181d2-ccaa-4b64-a229-234aa7a25db6"}
```
Сохраните из полученного ответа значение параметра token — токена авторизации.
Создайте описание для обработчика:
```
curl -o- -k https://example.com/api/eservice/v3/custom_equipment \
    -H "isp-box-instance: true" \
    -H "x-xsrf-token: <token>" \
    -d '{
        "device_type": "<device>",
        "handler": "<internal_handler_name>",
        "name": "<handler_name>",
        "protocol": ["<handler_protocol>"],
        "features": []
    }'
```
Пояснения к команде
example.com — доменное имя сервера c DCImanager 6
<token> — токен авторизации
<device> — тип устройства. Возможные значения:
- switch — коммутатор;
- ups — ИБП;
- pdu — PDU;
- vpu — VPU
<internal_handler_name> — уникальное внутреннее имя обработчика
<handler_name> — имя обработчика для отображения в интерфейсе платформы
<handler_protocol> — протокол для работы с обработчиком. Например, snmp
features — параметры вывода информации в пользовательский интерфейс. Для ИБП дополнительных параметров не предусмотрено, поэтому единственным возможным значением будет пустой массив.
Ответ будет содержать id созданного разработчика. Сохраните это значение.
Пример ответа
```
{"id": 1}
```
Загрузите архив с обработчиком в платформу:
```
curl -o- -k https://example.com/api/eservice/v3/custom_equipment/<handler_id>/content \
    -H "isp-box-instance: true" \
    -H "x-xsrf-token: <token>" \
    -F "data=@custom_handler.tar.gz" \
    -F "handler_import=<import_path>"
```
Пояснения к команде

example.com — доменное имя сервера c DCImanager 6
<handler_id> — id обработчика
<token> — токен авторизации
custom_handler.tar.gz — имя архива с обработчиком
<import_path> — относительный путь для импорта. Например, если файл обработчика my_handler.py находится в директории handler_dir, то относительный путь будет handler_dir.my_handler

Вы также можете использовать эту команду для загрузки новых версий обработчика в платформу.

Может быть полезно

Связанные статьи: