zigbee2mqtt_ota.rb

#!/usr/bin/env ruby

require "bundler/inline"

gemfile do
  source "https://rubygems.org"

  gem "json"
  gem "mqtt-ccutrer", "~> 1.0"
  gem "activesupport", "~> 7.2", require: false
end

require "active_support"
require "active_support/duration"

update_topics = {}

uri = ARGV.first

def simple_name(topic)
  topic.sub(%r{^zigbee2mqtt/}, "")
end

MQTT::Client.connect(uri) do |c|
  c.subscribe("homeassistant/update/+/config")
  c.subscribe("homeassistant/update/+/+/config")
  c.unsubscribe("homeassistant/update/+/config")
  c.unsubscribe("homeassistant/update/+/+/config", wait_for_ack: true)
  until c.queue_empty?
    packet = c.get
    config = JSON.load(packet.payload)
    update_topics[config["state_topic"]] = config["payload_install"]
  end

  loop do
    next_available = updating = nil
    remaining = nil

    c.subscribe("zigbee2mqtt/+")
    c.unsubscribe("zigbee2mqtt/+", wait_for_ack: true)

    until c.queue_empty?
      packet = c.get
      state = JSON.load(packet.payload)
      update_data = state["update"]
      next unless update_data

      case update_data["state"]
      when "updating"
        remaining = update_data["remaining"]
        remaining_string = "; ~#{ActiveSupport::Duration.build(remaining).inspect} remaining" if remaining
        puts "#{simple_name(packet.topic)} is still updating (#{update_data["progress"]}% complete#{remaining_string})"
        updating = true
      when "available"
        next_available ||= packet.topic if update_topics.key?(packet.topic)
      end
    end

    if !next_available && !updating
      puts "Done!"
      break
    elsif !updating
      puts "Starting update of #{simple_name(next_available)}"
      c.publish("zigbee2mqtt/bridge/request/device/ota_update/update", update_topics[next_available], qos: 1)
      # it can take a while for the update to start
      sleep 25
    else
      puts "Waiting..."
    end
    sleep 5
  end
end