Wrapper around pymelcloud to generate prometheus metrics from Mitsubishi Ecodan ATW heat pump

melcloud_exporter.py

#!/usr/bin/python3

"""
Copyright 2025 Brian Candler

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:

1.  Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.

2.  Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS”
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
"""

import aiohttp
import asyncio
import io
import pymelcloud
import re
import sys

from melcloud_config import MELCLOUD_USERNAME, MELCLOUD_PASSWORD

PREFIX="mel_"

async def emit_metrics(token, session, devices, buf):
    class Metric:
        def __init__(self, name, type="gauge", help=None):
            self.name = PREFIX + name
            self.type = type
            self.help = help
        def __enter__(self):
            if self.type:
                print(f"# TYPE {self.name} {self.type}", file=buf)
            if self.help:
                print(f"# HELP {self.name} {self.help}", file=buf)
            return self
        def __exit__(self, exc_type, exc_val, exc_tb):
            pass
        def __call__(self, value, **labels):
            if not labels:
                print(f"{self.name} {value}", file=buf)
                return
            print("{}{{{}}} {}".format(self.name, ",".join(
                    ("%s=\"%s\"" % (k,re.sub(r'"', r'\\"',v)) for k,v in labels.items())
                ), value), file=buf)

    atw = devices[pymelcloud.DEVICE_TYPE_ATW]
    if not atw:
        return

    for device in atw:
        await device.update()

    with Metric("device_info") as m:
        for device in atw:
            m(1,
                device_name=device.name,
                device_type=device.device_type,
                mac=device.mac,
                serial=device.serial,
            )
    with Metric("has_error") as m:
        for device in atw:
            m(int(device.has_error), device_name=device.name)
    with Metric("error_code") as m:
        for device in atw:
            m(device.error_code, device_name=device.name)
    with Metric("holiday_mode") as m:
        for device in atw:
            m(int(device.holiday_mode), device_name=device.name)
    with Metric("status") as m:
        for device in atw:
            for status in pymelcloud.atw_device._STATE_LOOKUP.values():
                m(int(device.status == status), device_name=device.name, status=status)
    with Metric("wifi_signal") as m:
        for device in atw:
            m(device.wifi_signal, device_name=device.name)
    with Metric("outside_temperature") as m:
        for device in atw:
            m(device.outside_temperature, device_name=device.name, temp_unit=device.temp_unit)
    with Metric("tank_temperature") as m:
        for device in atw:
            m(device.tank_temperature, device_name=device.name, temp_unit=device.temp_unit)
    with Metric("target_tank_temperature") as m:
        for device in atw:
            m(device.target_tank_temperature, device_name=device.name, temp_unit=device.temp_unit)
    with Metric("room_temperature") as m:
        for device in atw:
            for zone in device.zones:
                m(zone.room_temperature, device_name=device.name, zone_name=zone.name, temp_unit=device.temp_unit)
    with Metric("target_temperature") as m:
        for device in atw:
            for zone in device.zones:
                m(zone.target_temperature, device_name=device.name, zone_name=zone.name, temp_unit=device.temp_unit)
    with Metric("flow_temperature") as m:
        for device in atw:
            for zone in device.zones:
                m(zone.flow_temperature, device_name=device.name, zone_name=zone.name, temp_unit=device.temp_unit)
    with Metric("target_flow_temperature") as m:
        for device in atw:
            for zone in device.zones:
                m(zone.target_flow_temperature, device_name=device.name, zone_name=zone.name, temp_unit=device.temp_unit)
    with Metric("return_temperature") as m:
        for device in atw:
            for zone in device.zones:
                m(zone.return_temperature, device_name=device.name, zone_name=zone.name, temp_unit=device.temp_unit)
    # https://github.com/zaklex/pymelcloud/issues/1
    with Metric("daily_energy_consumed") as m:
        for device in atw:
            for mode in ["Heating", "Cooling", "HotWater"]:
                m(device.get_device_prop("Daily%sEnergyConsumed" % mode), device_name=device.name, mode=mode)
    with Metric("daily_energy_produced") as m:
        for device in atw:
            for mode in ["Heating", "Cooling", "HotWater"]:
                m(device.get_device_prop("Daily%sEnergyProduced" % mode), device_name=device.name, mode=mode)
    with Metric("current_energy_consumed", "counter") as m:
        for device in atw:
            m(device.get_device_prop("CurrentEnergyConsumed"), device_name=device.name)
    with Metric("current_energy_produced", "counter") as m:
        for device in atw:
            m(device.get_device_prop("CurrentEnergyProduced"), device_name=device.name)

from aiohttp import web

session = None
token = None
devices = None

async def handle(request):
    global session, token, devices
    buf = io.StringIO()
    # Assume our token can be used forever. If it becomes invalid,
    # we'd like the exporter to crash and a new instance spawned by systemd
    # (but will it just start returning 500 errors instead?)
    if session is None:
        session = aiohttp.ClientSession()
    if token is None:
        token = await pymelcloud.login(MELCLOUD_USERNAME, MELCLOUD_PASSWORD, session=session)
    if devices is None:
        devices = await pymelcloud.get_devices(token, session=session)
    await emit_metrics(token, session, devices, buf)
    return web.Response(text=buf.getvalue())

def init_func(argv):
    app = web.Application()
    app.router.add_get("/metrics", handle)
    return app

if __name__ == '__main__':
    app = init_func(sys.argv)
    web.run_app(app, port=8998)