Creating a UPS handler

You can add a custom UPS handler to DCImanager 6. To do this, create the handler code and upload it to the platform.

Preparing the Environment

The UPS handler must be written in Python. We recommend using Python version 3.9 or higher.

You can create a handler based on an existing project. To copy the project, connect to the DCImanager 6 location server via SSH and run the command:

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

The following project directories may be useful when creating a handler:

/common — common helper classes and functions;
/ups_common — helper classes and functions for working with UPS;
/ups_common/handlers — UPS handlers.

You can see the required Python libraries and their versions in the requirements.txt project file. To install the necessary libraries, run the command:

pip3 install -r requirements.txt

To check data types in the project, we recommend using the mypy analyzer.

Creating a handler

Class for working with UPS

The UPS class inherits from the class:

BaseSnmpUps — for devices operating via SNMP;
BaseUps — for other devices.

The BaseSnmpUps class contains methods for interacting with the UPS via SNMP:

snmp_get — perform a read request by a specific OID;
snmp_set — perform a modifying request by a specific OID;
snmp_walk — perform a request that returns a list.

Example of class description

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

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

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

Methods for working with UPS

For DCImanager 6 to interact with the UPS, override the methods of the BaseUps class:

status — UPS polling;
port_up — enable port;
port_down — disable port;
statistic — collect statistics.

The equipment service uses not raw json data, but their object representation. Instances of the UpsView class are the UPS representation. It has the following parameters:

load_percentage — load, %;
charge_time — charge time, min;
input_power — output power, kW;
output_power — power consumption, kW.

The UpsData class inherits from the base class ConnectionParams and defines the data for accessing the UPS.

Each handler file must contain the function make_handler. This function creates the handler object:

Example function

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

Example handler code

Example code

"""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,
        )

Uploading the handler to the platform

To upload the handler to the platform:

Create a directory with the following structure:
```
handler_dir/
├── __init__.py
└── my_handler.py
```
Explanations

handler_dir — the name of the directory. There must not be a handler directory with the same name in the platform.
__init__.py — initialization file. If there is no such file, create an empty file with this name.
my_handler.py — the handler file
Create a tar.gz archive with this directory:
```
tar -czvf custom_handler.tar.gz handler_dir
```
Command explanations

custom_handler.tar.gz — archive name
handler_dir –- name of the created directory with the handler
Authenticate in DCImanager 6 with administrator rights:
```
curl -o- -k https://example.com/api/auth/v4/public/token \
    -H "isp-box-instance: true" \
    -d '{
        "email": "<admin_email>",
        "password": "<admin_pass>"
    }'
```
Command explanations

example.com — domain name of the DCImanager 6 server
<admin_email> — DCImanager 6 administrator email
<admin_pass> — DCImanager 6 administrator password

The response will be of the form:
```
{"id":"24","token":"24-cee181d2-ccaa-4b64-a229-234aa7a25db6"}
```
Save the value of the token parameter from the received response — the authorization token.
Create a description for the handler:
```
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": []
    }'
```
Command explanations
example.com — domain name of the DCImanager 6 server
<token> — authorization token
<device> — device type. Possible values:
- switch — switch;
- ups — UPS;
- pdu — PDU;
- vpu — VPU
<internal_handler_name> — unique internal handler name
<handler_name> — handler name to display in the platform interface
<handler_protocol> — protocol for working with the handler. For example, snmp
features — parameters for displaying information in the user interface. For UPS, no additional parameters are provided, so the only possible value is an empty array.
The response will contain the id of the created handler. Save this value.
Example response
```
{"id": 1}
```
Upload the archive with the handler to the platform:
```
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>"
```
Command explanations

example.com — domain name of the DCImanager 6 server
<handler_id> — handler id
<token> — authorization token
custom_handler.tar.gz — name of the archive with the handler
<import_path> — relative import path. For example, if the handler file my_handler.py is in the handler_dir directory, the relative path would be handler_dir.my_handler

You can also use this command to upload new versions of the handler to the platform.

Useful tips

Related articles: