Bluetti Prometheus Montior

server.rb

require 'bigdecimal'
require 'json'
require 'mqtt'
require 'prometheus_exporter'
require 'prometheus_exporter/server'

class MetricsServer
  def initialize(broker)
    @broker = broker

    @mutex = Mutex.new
    @stopped = false
    @integral_hours = {}
    @thread = nil
  end

  def start
    @server = PrometheusExporter::Server::WebServer.new({})
    @server.start

    register_metrics

    @thread = Thread.new(&method(:run))
    @thread.abort_on_exception = true

    self
  end

  def stop
    @mutex.synchronize { @stopped = true }
    @thread.join
    @server.stop
  end

  def stopped?
    @mutex.synchronize { @stopped }
  end

  private

  def run
    MQTT::Client.connect(@broker) do |c|
      c.get('bluetti/state/#') do |topic, message|
        break if stopped?

        match = /^bluetti\/state\/([^-]+)-(\d+)\/(.+)$/.match(topic)
        labels = { device_type: match[1], serial_number: match[2] }

        # puts "#{topic}: #{message}"

        case match[3]
        when 'dc_input_power'
          value = message.to_i
          @solar_power.observe(value, labels)
          @solar_power_wh.observe(integral_hours('solar_power', value), labels)
        when 'ac_input_power'
          value = message.to_i
          @grid_power.observe(value, labels)
          @grid_power_wh.observe(integral_hours('grid_power', value), labels)
        when 'ac_output_power'
          value = message.to_i
          @ac_output_power.observe(value, labels)
          @ac_output_power_wh.observe(integral_hours('ac_output_power', value), labels)
        when 'dc_output_power'
          value = message.to_i
          @dc_output_power.observe(value, labels)
          @dc_output_power_wh.observe(integral_hours('dc_output_power', value), labels)
        when 'internal_ac_voltage'
          @inverter_voltage.observe(BigDecimal(message), labels)
          @internal_ac_voltage.observe(BigDecimal(message), labels)
        when 'internal_ac_frequency'
          @inverter_frequency.observe(BigDecimal(message), labels)
          @internal_ac_frequency.observe(BigDecimal(message), labels)
        when 'ac_input_voltage'
          @grid_voltage.observe(BigDecimal(message), labels)
        when 'ac_input_frequency'
          @grid_frequency.observe(BigDecimal(message), labels)
        when 'ac_output_on'
          @ac_output_on.observe(message == 'ON' ? 1 : 0, labels)
        when 'dc_output_on'
          @dc_output_on.observe(message == 'ON' ? 1 : 0, labels)
        when 'internal_power_one'
          @internal_power_one.observe(message.to_i, labels)
        when 'internal_power_two'
          @internal_power_two.observe(message.to_i, labels)
        when 'internal_power_three'
          @internal_power_three.observe(message.to_i, labels)
        when 'internal_current_one'
          @internal_current_one.observe(BigDecimal(message), labels)
        when 'internal_current_two'
          @internal_current_two.observe(BigDecimal(message), labels)
        when 'internal_current_three'
          @internal_current_three.observe(BigDecimal(message), labels)
        when 'dc_input_voltage1'
          @dc_input_voltage.observe(BigDecimal(message), labels.merge({ input_num: 1 }))
        when 'dc_input_power1'
          @dc_input_power.observe(message.to_i, labels.merge({ input_num: 1 }))
        when 'dc_input_current1'
          @dc_input_current.observe(BigDecimal(message), labels.merge({ input_num: 1 }))
        when 'total_battery_percent'
          @total_battery_percent.observe(message.to_i, labels)
        when 'pack_details1'
          json = JSON.parse(message)
          pack_labels = labels.merge({ pack_num: 1 })
          @pack_battery_percent.observe(json['percent'], pack_labels)
          json['voltages'].each_with_index do |voltage, i|
            @cell_voltage.observe(voltage, pack_labels.merge({ cell_num: i + 1 }))
          end
        end
      end
    end
  end

  def register_metrics
    @solar_power = build_gauge('solar_power_watts', 'Current solar input')
    @solar_power_wh = build_counter('solar_power_wh', 'Total solar power generation')
    @grid_power = build_gauge('grid_power_watts', 'Current grid input')
    @grid_power_wh = build_counter('grid_power_wh', 'Total grid input')
    @ac_output_power = build_gauge('ac_output_power_watts', 'Current AC power output')
    @ac_output_power_wh = build_counter('ac_output_power_wh', 'Cumulative AC power output')
    @dc_output_power = build_gauge('dc_output_power_watts', 'Current DC power output')
    @dc_output_power_wh = build_counter('dc_output_power_wh', 'Cumulative DC power output')
    @inverter_voltage = build_gauge('inverter_voltage', 'The voltage of the AC inverter')
    @inverter_frequency = build_gauge('inverter_frequency', 'The AC frequency of the inverter')
    @grid_voltage = build_gauge('grid_voltage', 'The voltage of the grid')
    @grid_frequency = build_gauge('grid_frequency', 'The AC frequency of the grid')
    @total_battery_percent = build_gauge('total_battery_percent', 'Total battery percent')
    @pack_battery_percent = build_gauge('pack_battery_percent', 'Pack battery percent')
    @cell_voltage = build_gauge('cell_voltage', 'Voltage of a single cell in a pack')

    @internal_ac_voltage = build_gauge('internal_ac_voltage', 'AC voltage sensor')
    @internal_current_one = build_gauge('internal_current_one', 'Current sensor reading in amps')
    @internal_power_one = build_gauge('internal_power_one', 'Power sensor reading in watts')
    @internal_ac_frequency = build_gauge('internal_ac_frequency', 'AC frequency sensor reading in Hz')
    @internal_current_two = build_gauge('internal_current_two', 'Current sensor reading in amps')
    @internal_power_two = build_gauge('internal_power_two', 'Power sensor reading in watts')
    @internal_current_three = build_gauge('internal_current_three', 'Current sensor reading in amps')
    @internal_power_three = build_gauge('internal_power_three', 'Power sensor reading in watts')
    @dc_input_voltage = build_gauge('dc_input_voltage', 'Individual DC input voltage')
    @dc_input_power = build_gauge('dc_input_power', 'Individual DC input power in watts')
    @dc_input_current = build_gauge('dc_input_current', 'Individual DC input current in amps')
    @ac_output_on = build_gauge('ac_output_on', 'Whether AC output is on')
    @dc_output_on = build_gauge('dc_output_on', 'Whether DC output is on')
  end

  def integral_hours(field_name, new_value)
    @mutex.synchronize do
      now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      if @integral_hours.key?(field_name)
        last_time, last_value = @integral_hours[field_name]
        @integral_hours[field_name] = [now, new_value]

        if now - last_time > 120
          # Report no progress if we have a long dropout
          0
        else
          # Calculate the trapazoidal area - 1/2 * h * (d1 + d2)
          0.5 * (now - last_time) / 3600.0 * (last_value + new_value)
        end
      else
        @integral_hours[field_name] = [now, new_value]
        0
      end
    end
  end

  def build_gauge(name, hint = '')
    metric = PrometheusExporter::Metric::Gauge.new(name, hint)
    @server.collector.register_metric(metric)
    metric
  end

  def build_counter(name, hint = '')
    metric = PrometheusExporter::Metric::Counter.new(name, hint)
    @server.collector.register_metric(metric)
    metric
  end
end

# Set up logging
$stdout.sync = true

# Start up metrics server
metrics_server = MetricsServer.new(ARGV[0]).start

# Start up signal handlers
r, w = IO.pipe
main_thread = Thread.current
signal_handler = Thread.new do
  while (io = IO.select([r]))
    metrics_server.stop
    break
  end
  main_thread.run # Wake up from sleep
end
%w[INT TERM].each { |s| Signal.trap(s) { w.puts(s) } }

# Sleep forever
sleep