179 lines
8.2 KiB
Python
179 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Expose two live bcachefs options through Home Assistant MQTT discovery."""
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import threading
|
|
import time
|
|
import subprocess
|
|
|
|
from reconcile_metrics import FIELDS, read_metrics
|
|
import uuid as uuid_module
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
OPTIONS = {"reconcile_enabled": "Reconcile", "copygc_enabled": "Copy GC"}
|
|
|
|
|
|
class Bridge:
|
|
def __init__(self, client, uuid, sysfs=Path('/sys/fs/bcachefs'), discovery='homeassistant'):
|
|
self.client = client
|
|
self.uuid = str(uuid_module.UUID(uuid))
|
|
self.options = Path(sysfs) / self.uuid / 'options'
|
|
self.base = f'bcachefs/{self.uuid}'
|
|
self.discovery = discovery
|
|
self.lock = threading.RLock()
|
|
self.last = {}
|
|
self.reconcile_refresh = threading.Event()
|
|
client.on_connect = self.on_connect
|
|
client.on_message = self.on_message
|
|
client.will_set(f'{self.base}/availability', 'offline', qos=1, retain=True)
|
|
|
|
def publish(self, topic, payload):
|
|
return self.client.publish(topic, payload, qos=1, retain=True)
|
|
|
|
def announce(self):
|
|
for option, name in OPTIONS.items():
|
|
topic = f'{self.base}/{option}'
|
|
config = {
|
|
'name': name, 'unique_id': f'bcachefs_{self.uuid}_{option}',
|
|
'default_entity_id': f'switch.bcachefs_{option}',
|
|
'command_topic': f'{topic}/set', 'state_topic': f'{topic}/state',
|
|
'payload_on': '1', 'payload_off': '0', 'optimistic': False,
|
|
'qos': 1, 'retain': False,
|
|
'availability_mode': 'all',
|
|
'availability': [{'topic': f'{self.base}/availability'},
|
|
{'topic': f'{topic}/availability'}],
|
|
'device': {'identifiers': [f'bcachefs_{self.uuid}'],
|
|
'name': f'Bcachefs {socket.gethostname()}',
|
|
'manufacturer': 'bcachefs', 'model': 'Filesystem options'},
|
|
}
|
|
self.publish(f'{self.discovery}/switch/bcachefs_{self.uuid}/{option}/config', json.dumps(config))
|
|
|
|
for field in FIELDS:
|
|
config = {
|
|
'name': 'Reconcile ' + field.replace('_', ' '),
|
|
'unique_id': f'bcachefs_{self.uuid}_reconcile_{field}',
|
|
'default_entity_id': f'sensor.bcachefs_reconcile_{field}',
|
|
'state_topic': f'{self.base}/reconcile/state',
|
|
'value_template': '{{ value_json.' + field + ' | default(none) }}',
|
|
'device_class': 'data_size', 'state_class': 'measurement',
|
|
'unit_of_measurement': 'B', 'suggested_display_precision': 0,
|
|
# Keep sampling timestamps even while rounded byte counts are flat.
|
|
'force_update': True, 'expire_after': 90,
|
|
'availability_mode': 'all',
|
|
'availability': [{'topic': f'{self.base}/availability'},
|
|
{'topic': f'{self.base}/reconcile/availability'}],
|
|
'device': {'identifiers': [f'bcachefs_{self.uuid}']},
|
|
}
|
|
self.publish(f'{self.discovery}/sensor/bcachefs_{self.uuid}/reconcile_{field}/config', json.dumps(config))
|
|
|
|
def sample_reconcile(self, mount):
|
|
try:
|
|
if not self.options.is_dir():
|
|
raise OSError('Filesystem not mounted')
|
|
values = read_metrics(self.uuid, mount)
|
|
except (OSError, ValueError, subprocess.SubprocessError) as error:
|
|
logging.warning('Reconcile measurements unavailable: %s', error)
|
|
self.publish(f'{self.base}/reconcile/availability', 'offline')
|
|
return
|
|
self.publish(f'{self.base}/reconcile/state', json.dumps(values))
|
|
self.publish(f'{self.base}/reconcile/availability', 'online')
|
|
|
|
def sample(self, force=False):
|
|
with self.lock:
|
|
if not self.options.is_dir():
|
|
self.publish(f'{self.base}/reconcile/availability', 'offline')
|
|
for option in OPTIONS:
|
|
topic = f'{self.base}/{option}'
|
|
try:
|
|
value = (self.options / option).read_text().strip()
|
|
if value not in ('0', '1'):
|
|
raise ValueError(f'Unexpected option value: {value!r}')
|
|
except (OSError, ValueError):
|
|
value = None
|
|
if force or option not in self.last or self.last[option] != value:
|
|
if value is not None:
|
|
self.publish(f'{topic}/state', value)
|
|
self.publish(f'{topic}/availability', 'online' if value is not None else 'offline')
|
|
self.last[option] = value
|
|
|
|
def on_connect(self, client, userdata, flags, reason_code, properties):
|
|
if reason_code.is_failure:
|
|
logging.error('MQTT connection refused: %s', reason_code)
|
|
return
|
|
with self.lock:
|
|
self.publish(f'{self.base}/availability', 'offline')
|
|
self.publish(f'{self.base}/reconcile/availability', 'offline')
|
|
self.reconcile_refresh.set()
|
|
self.announce()
|
|
for option in OPTIONS:
|
|
client.subscribe(f'{self.base}/{option}/set', qos=1)
|
|
client.subscribe(f'{self.discovery}/status', qos=1)
|
|
self.sample(force=True)
|
|
self.publish(f'{self.base}/availability', 'online')
|
|
logging.info('Connected; discovery and live states published')
|
|
|
|
def on_message(self, client, userdata, message):
|
|
with self.lock:
|
|
if message.topic == f'{self.discovery}/status':
|
|
if message.payload == b'online':
|
|
self.announce()
|
|
self.sample(force=True)
|
|
self.reconcile_refresh.set()
|
|
return
|
|
option = next((name for name in OPTIONS if message.topic == f'{self.base}/{name}/set'), None)
|
|
if option is None or message.retain or message.payload not in (b'0', b'1'):
|
|
logging.warning('Ignoring invalid or retained command on %s', message.topic)
|
|
return
|
|
try:
|
|
# Never create a file if the filesystem disappeared between polling and the command.
|
|
fd = os.open(self.options / option, os.O_WRONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
|
|
with os.fdopen(fd, 'wb') as target:
|
|
target.write(message.payload + b'\n')
|
|
logging.info('Applied %s=%s', option, message.payload.decode())
|
|
except OSError as error:
|
|
logging.error('Could not set %s: %s', option, error)
|
|
self.sample(force=True)
|
|
|
|
|
|
def main():
|
|
logging.basicConfig(level=logging.INFO, format='%(levelname)s %(message)s')
|
|
uuid = os.environ['BCACHEFS_UUID']
|
|
interval = float(os.environ.get('POLL_INTERVAL', '2'))
|
|
if interval <= 0:
|
|
raise ValueError('POLL_INTERVAL must be positive')
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f'bcachefs-{uuid}', clean_session=True)
|
|
bridge = Bridge(client, uuid, discovery=os.environ.get('MQTT_DISCOVERY_PREFIX', 'homeassistant'))
|
|
if os.environ.get('MQTT_USERNAME'):
|
|
client.username_pw_set(os.environ['MQTT_USERNAME'], os.environ.get('MQTT_PASSWORD'))
|
|
if os.environ.get('MQTT_CA_FILE'):
|
|
client.tls_set(ca_certs=os.environ['MQTT_CA_FILE'])
|
|
client.reconnect_delay_set(min_delay=1, max_delay=30)
|
|
stopped = threading.Event()
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
signal.signal(sig, lambda *_: stopped.set())
|
|
client.connect_async(os.environ['MQTT_HOST'], int(os.environ.get('MQTT_PORT', '1883')), keepalive=15)
|
|
client.loop_start()
|
|
next_reconcile = 0.0
|
|
try:
|
|
while not stopped.wait(interval):
|
|
if client.is_connected():
|
|
bridge.sample()
|
|
if bridge.reconcile_refresh.is_set() or time.monotonic() >= next_reconcile:
|
|
bridge.reconcile_refresh.clear()
|
|
bridge.sample_reconcile(os.environ.get('FSPATH', '/mnt'))
|
|
next_reconcile = time.monotonic() + 30
|
|
finally:
|
|
if client.is_connected():
|
|
bridge.publish(f'{bridge.base}/availability', 'offline').wait_for_publish(timeout=3)
|
|
client.disconnect()
|
|
client.loop_stop()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|