590 lines
25 KiB
Python
590 lines
25 KiB
Python
#!/usr/bin/python3
|
|
"""
|
|
Matrix bot support for r0b3BOT.
|
|
|
|
Runs alongside the Discord bot in the same asyncio event loop (via asyncio.gather).
|
|
Enabled by adding a [matrix-config] section to bot_config.conf.
|
|
|
|
Supported commands (same prefix as Discord):
|
|
greetings, cat, explain, holyshit, smudge, printstat, bambustat, info, help, last_error
|
|
|
|
Voice/sound commands respond with a friendly Discord-only message.
|
|
Discord embeds are rendered as formatted HTML (org.matrix.custom.html).
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import os
|
|
import random
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
import octoapi
|
|
import requests
|
|
from nio import AsyncClient, AsyncClientConfig, JoinError, LoginError, MatrixRoom, MegolmEvent, RoomMessage, RoomMessageText, RoomUpgradeEvent, SyncError, UploadResponse
|
|
|
|
|
|
class MatrixBot:
|
|
def __init__(
|
|
self,
|
|
homeserver,
|
|
user_id,
|
|
password,
|
|
command_prefix,
|
|
room_ids,
|
|
get_bambu_status,
|
|
get_last_error,
|
|
hass_url,
|
|
hass_token,
|
|
hass_light,
|
|
octoprint_ip,
|
|
bambu_ip,
|
|
bambu_access_code,
|
|
bambu_camera_enable,
|
|
capture_bambu_frame=None,
|
|
store_path="./matrix_store/",
|
|
):
|
|
# Create the crypto store directory if it doesn't exist.
|
|
# matrix-nio[e2e] uses this to persist Olm/Megolm sessions and device keys
|
|
# across restarts so the bot can decrypt messages in E2E-encrypted rooms.
|
|
os.makedirs(store_path, exist_ok=True)
|
|
|
|
nio_config = AsyncClientConfig(store_sync_tokens=True)
|
|
self.client = AsyncClient(homeserver, user_id, store_path=store_path, config=nio_config)
|
|
self.prefix = command_prefix
|
|
self.room_ids = set(room_ids)
|
|
self.password = password
|
|
# Credentials file — persists access_token + device_id across restarts so we
|
|
# reuse the same Matrix device (and its Olm keys) rather than registering a new
|
|
# one on every bot startup.
|
|
self.credentials_file = os.path.join(store_path, "r0b3bot_credentials.json")
|
|
|
|
# Shared state — lambdas that read from r0b3bot.py module-level globals
|
|
self._get_bambu_status = get_bambu_status
|
|
self._get_last_error = get_last_error
|
|
|
|
# Configuration
|
|
self.hass_url = hass_url
|
|
self.hass_token = hass_token
|
|
self.hass_light = hass_light
|
|
self.octoprint_ip = octoprint_ip
|
|
self.bambu_ip = bambu_ip
|
|
self.bambu_access_code = bambu_access_code
|
|
self.bambu_camera_enable = bambu_camera_enable
|
|
self._capture_bambu_frame = capture_bambu_frame
|
|
self._start_time = time.time()
|
|
|
|
async def _login(self):
|
|
"""Authenticate with the homeserver.
|
|
|
|
Tries to restore a previously saved session first (same device_id, same Olm
|
|
keys). Falls back to a fresh password login if no credentials are saved or if
|
|
the saved token has expired. On a fresh login the credentials are persisted to
|
|
disk so the same device is reused on subsequent restarts — this is essential for
|
|
E2E encryption: the homeserver and other clients must be able to establish a
|
|
stable Olm session with a consistent device.
|
|
"""
|
|
if os.path.exists(self.credentials_file):
|
|
try:
|
|
with open(self.credentials_file) as f:
|
|
creds = json.load(f)
|
|
self.client.access_token = creds["access_token"]
|
|
self.client.user_id = creds["user_id"]
|
|
self.client.device_id = creds["device_id"]
|
|
self.client.load_store()
|
|
print(f">> [Matrix] Restored session: {creds['user_id']} / device {creds['device_id']}")
|
|
return True
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Could not restore session ({e}), doing fresh login")
|
|
# Remove the stale credentials so we don't retry a broken session
|
|
try:
|
|
os.remove(self.credentials_file)
|
|
except OSError:
|
|
pass
|
|
|
|
# Fresh password login — matrix-nio registers a new device and uploads Olm keys
|
|
resp = await self.client.login(self.password, device_name="r0b3bot")
|
|
if isinstance(resp, LoginError):
|
|
print(f" !! [Matrix] Login failed: {resp.message}")
|
|
print(f" !! [Matrix] Check matrix_password in bot_config.conf")
|
|
return False
|
|
|
|
print(f">> [Matrix] Logged in as {self.client.user_id} / device {self.client.device_id}")
|
|
self.client.load_store()
|
|
|
|
# Persist credentials for next restart
|
|
with open(self.credentials_file, "w") as f:
|
|
json.dump({
|
|
"access_token": self.client.access_token,
|
|
"user_id": self.client.user_id,
|
|
"device_id": self.client.device_id,
|
|
}, f, indent=2)
|
|
print(f">> [Matrix] Credentials saved to {self.credentials_file}")
|
|
return True
|
|
|
|
async def start(self):
|
|
print(">> [Matrix] Starting Matrix bot client")
|
|
|
|
if not await self._login():
|
|
print(" !! [Matrix] Matrix bot will NOT start.")
|
|
return
|
|
|
|
# Upload Olm one-time keys if this is the first run with a fresh Olm account.
|
|
# On subsequent restarts with a restored session this is a no-op.
|
|
if self.client.should_upload_keys:
|
|
print(">> [Matrix] Uploading Olm keys to homeserver")
|
|
await self.client.keys_upload()
|
|
|
|
# Initial sync with no timeout to consume backlog before registering callbacks,
|
|
# so we don't replay old messages on startup.
|
|
try:
|
|
resp = await self.client.sync(timeout=0)
|
|
if isinstance(resp, SyncError):
|
|
print(f" ! [Matrix] Initial sync error: {resp.message}")
|
|
else:
|
|
print(f">> [Matrix] Initial sync OK, next_batch={self.client.next_batch!r}")
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Initial sync exception: {e}")
|
|
|
|
# Join all configured rooms. The bot account must be a member of a room
|
|
# to receive its events in the sync stream — having the room ID in config
|
|
# alone is not sufficient.
|
|
for room_id in self.room_ids:
|
|
resp = await self.client.join(room_id)
|
|
if isinstance(resp, JoinError):
|
|
print(f" ! [Matrix] Failed to join {room_id}: {resp.message}")
|
|
else:
|
|
print(f">> [Matrix] Joined (or already in) room {room_id}")
|
|
|
|
self.client.add_event_callback(self._on_message, RoomMessageText)
|
|
self.client.add_event_callback(self._on_any_message, RoomMessage)
|
|
self.client.add_event_callback(self._on_encrypted, MegolmEvent)
|
|
self.client.add_event_callback(self._on_room_upgrade, RoomUpgradeEvent)
|
|
print(">> [Matrix] Listening for commands")
|
|
# Use a short poll timeout (5 s) to stay well under any reverse-proxy
|
|
# idle timeout. A 30 s long-poll through nginx/caddy will often get a
|
|
# 504 which matrix-nio can't parse, causing the "next_batch required"
|
|
# loop and silencing all callbacks. 5 s is safe for all common setups.
|
|
await self.client.sync_forever(timeout=5000)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Event handler
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def _on_any_message(self, room: MatrixRoom, event: RoomMessage):
|
|
"""DEBUG — fires for any m.room.message msgtype, not just m.text."""
|
|
print(f" [Matrix DEBUG RoomMessage] type={type(event).__name__} from {event.sender} in {room.room_id!r}")
|
|
|
|
async def _on_encrypted(self, room: MatrixRoom, event: MegolmEvent):
|
|
"""Fires for encrypted messages the bot can't yet decrypt (missing session key).
|
|
Normal on first run — the sender's client needs to establish an Olm session with
|
|
the bot's device and share the room key. Sending another message from Element
|
|
triggers this automatically. We also send a key request to speed it up."""
|
|
print(f" ! [Matrix] Could not decrypt message from {event.sender} in {room.room_id!r} "
|
|
f"(session {event.session_id!r} not yet received — send another message to trigger key sharing)")
|
|
try:
|
|
await self.client.request_room_key(event)
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Key request failed: {e}")
|
|
|
|
async def _on_room_upgrade(self, room: MatrixRoom, event: RoomUpgradeEvent):
|
|
"""Handle a room upgrade (tombstone). Join the replacement room and swap
|
|
the room ID in the active set so commands keep working without a restart.
|
|
The old room ID is left in bot_config.conf — update it to the new one."""
|
|
new_room_id = event.replacement_room
|
|
print(f">> [Matrix] Room {room.room_id} upgraded → {new_room_id}, joining replacement")
|
|
resp = await self.client.join(new_room_id)
|
|
if isinstance(resp, JoinError):
|
|
print(f" ! [Matrix] Failed to join upgraded room {new_room_id}: {resp.message}")
|
|
return
|
|
self.room_ids.discard(room.room_id)
|
|
self.room_ids.add(new_room_id)
|
|
print(f">> [Matrix] Now listening in {new_room_id}")
|
|
print(f" ! [Matrix] Update matrix_room_ids in bot_config.conf: replace {room.room_id} with {new_room_id}")
|
|
|
|
async def _on_message(self, room: MatrixRoom, event: RoomMessageText):
|
|
# DEBUG — remove once commands are confirmed working
|
|
print(f" [Matrix DEBUG] event from {event.sender} in {room.room_id!r} body={event.body!r}")
|
|
print(f" [Matrix DEBUG] configured room_ids={self.room_ids}")
|
|
|
|
# Ignore own messages
|
|
if event.sender == self.client.user_id:
|
|
print(" [Matrix DEBUG] ignored: own message")
|
|
return
|
|
# Only respond in configured rooms
|
|
if room.room_id not in self.room_ids:
|
|
print(f" [Matrix DEBUG] ignored: room_id not in configured list")
|
|
return
|
|
# Only respond to messages that start with the command prefix
|
|
if not event.body.startswith(self.prefix):
|
|
print(f" [Matrix DEBUG] ignored: no command prefix {self.prefix!r}")
|
|
return
|
|
|
|
body = event.body[len(self.prefix):]
|
|
parts = body.split()
|
|
if not parts:
|
|
return
|
|
|
|
command = parts[0].lower()
|
|
print(f">> [Matrix] {event.sender} in {room.room_id} called {self.prefix}{command}")
|
|
await self._dispatch(room.room_id, command)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Command dispatcher
|
|
# -------------------------------------------------------------------------
|
|
|
|
_SOUND_COMMANDS = {
|
|
"aww", "bitch", "big", "boom", "breee", "cowbell", "eia", "heavy",
|
|
"iran", "leeroy", "nice", "oops", "promoted", "rs", "trololo",
|
|
"wow", "yaw",
|
|
}
|
|
|
|
async def _dispatch(self, room_id, command):
|
|
handlers = {
|
|
"greetings": self._cmd_greetings,
|
|
"cat": self._cmd_cat,
|
|
"explain": self._cmd_explain,
|
|
"holyshit": self._cmd_holyshit,
|
|
"smudge": self._cmd_smudge,
|
|
"printstat": self._cmd_printstat,
|
|
"bambustat": self._cmd_bambustat,
|
|
"info": self._cmd_info,
|
|
"help": self._cmd_help,
|
|
"last_error": self._cmd_last_error,
|
|
}
|
|
if command in handlers:
|
|
await handlers[command](room_id)
|
|
elif command in self._SOUND_COMMANDS:
|
|
await self._send_text(
|
|
room_id,
|
|
"Sound commands are only available on Discord (voice channels)."
|
|
)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Helpers
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def _share_session(self, room_id):
|
|
"""Share the Megolm outbound group session with all current room members.
|
|
|
|
Must be called before every room_send() in encrypted rooms. This establishes
|
|
Olm sessions with each member's devices (claiming one-time keys as needed) and
|
|
sends them the Megolm session key so they can decrypt our messages.
|
|
|
|
ignore_unverified_devices=True lets the bot still send even when some devices
|
|
have unverifiable keys (e.g. stale bot devices with mismatched Olm keys).
|
|
"""
|
|
try:
|
|
await self.client.share_group_session(room_id, ignore_unverified_devices=True)
|
|
except Exception as e:
|
|
print(f" ! [Matrix] share_group_session for {room_id}: {e}")
|
|
|
|
async def _send_text(self, room_id, text, html=None):
|
|
await self._share_session(room_id)
|
|
content = {"msgtype": "m.text", "body": text}
|
|
if html:
|
|
content["format"] = "org.matrix.custom.html"
|
|
content["formatted_body"] = html
|
|
await self.client.room_send(room_id, "m.room.message", content)
|
|
|
|
async def _send_image(self, room_id, file_path):
|
|
await self._share_session(room_id)
|
|
with open(file_path, "rb") as f:
|
|
data = f.read()
|
|
|
|
# Encrypted rooms require the attachment itself to be encrypted before upload.
|
|
# matrix-nio handles this via encrypt=True on upload(), which returns the key
|
|
# info needed to build the m.image "file" field so clients can decrypt it.
|
|
room = self.client.rooms.get(room_id)
|
|
encrypt = room is not None and room.encrypted
|
|
|
|
resp, keys = await self.client.upload(
|
|
io.BytesIO(data),
|
|
content_type="image/jpeg",
|
|
filename=os.path.basename(file_path),
|
|
encrypt=encrypt,
|
|
)
|
|
if not isinstance(resp, UploadResponse):
|
|
print(f" ! [Matrix] Image upload failed: {resp}")
|
|
return
|
|
|
|
if encrypt and keys:
|
|
content = {
|
|
"msgtype": "m.image",
|
|
"body": os.path.basename(file_path),
|
|
"file": {"url": resp.content_uri, **keys},
|
|
}
|
|
else:
|
|
content = {
|
|
"msgtype": "m.image",
|
|
"url": resp.content_uri,
|
|
"body": os.path.basename(file_path),
|
|
"info": {"mimetype": "image/jpeg"},
|
|
}
|
|
await self.client.room_send(room_id, "m.room.message", content)
|
|
|
|
def _format_html(self, title, description, fields):
|
|
"""Render a status block as a heading + two-column table."""
|
|
rows = "".join(
|
|
f"<tr><td><strong>{name}</strong></td><td>{value}</td></tr>"
|
|
for name, value in fields
|
|
)
|
|
desc_html = f"<p>{description}</p>" if description else ""
|
|
return f"<h4>{title}</h4>{desc_html}<table>{rows}</table>"
|
|
|
|
def _format_text(self, title, description, fields):
|
|
"""Plain-text fallback — pad label column for alignment."""
|
|
max_len = max((len(name) for name, _ in fields), default=0)
|
|
parts = [title]
|
|
if description:
|
|
parts.append(description)
|
|
for name, value in fields:
|
|
parts.append(f" {name:<{max_len}} {value}")
|
|
return "\n".join(parts)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Command implementations
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def _cmd_greetings(self, room_id):
|
|
await self._send_text(room_id, "Hello there!")
|
|
|
|
async def _cmd_cat(self, room_id):
|
|
await self._send_text(room_id, "https://media.giphy.com/media/JIX9t2j0ZTN9S/giphy.gif")
|
|
|
|
async def _cmd_explain(self, room_id):
|
|
await self._send_text(room_id, "https://media.giphy.com/media/A8NNZlVuA1LoY/giphy.gif")
|
|
|
|
async def _cmd_holyshit(self, room_id):
|
|
await self._send_text(room_id, "https://media.giphy.com/media/QcGTswMKyBNgA/giphy.gif")
|
|
|
|
async def _cmd_smudge(self, room_id):
|
|
print(" - [Matrix] Uploading Smudge to Matrix room")
|
|
await self._send_image(room_id, "./images/smudge.png")
|
|
|
|
async def _cmd_last_error(self, room_id):
|
|
await self._send_text(room_id, f"Last error recorded: {self._get_last_error()}")
|
|
|
|
async def _cmd_info(self, room_id):
|
|
# Uptime
|
|
elapsed = time.time() - self._start_time
|
|
m, s = divmod(int(elapsed), 60)
|
|
h, m = divmod(m, 60)
|
|
d, h = divmod(h, 24)
|
|
w, d = divmod(d, 7)
|
|
uptime = f"{w:02d}:{d:02d}:{h:02d}:{m:02d}:{s:02d}"
|
|
|
|
# Git version tag
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"git", "describe", "--tags",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5)
|
|
bot_version = stdout.decode().strip() or "unknown"
|
|
except Exception:
|
|
bot_version = "unknown"
|
|
|
|
source_url = "https://gitlab.rickelobe.com/Bots/r0b3BOT"
|
|
fields = [
|
|
("Author", "R0b3"),
|
|
("Uptime [WW:DD:HH:MM:SS]", uptime),
|
|
("Version", bot_version),
|
|
("Python", sys.version.split()[0]),
|
|
("Source", f'<a href="{source_url}">{source_url}</a>'),
|
|
]
|
|
await self._send_text(
|
|
room_id,
|
|
self._format_text("r0b3BOT — Derpy Derp of a Bot", None, fields),
|
|
html=self._format_html("r0b3BOT — Derpy Derp of a Bot", None, fields),
|
|
)
|
|
|
|
async def _cmd_help(self, room_id):
|
|
p = self.prefix
|
|
commands = [
|
|
(f"{p}greetings", "Hello!"),
|
|
(f"{p}cat", "GIF"),
|
|
(f"{p}explain", "GIF"),
|
|
(f"{p}holyshit", "GIF"),
|
|
(f"{p}smudge", "Sends the Smudge image"),
|
|
(f"{p}printstat", "OctoPrint status + snapshot"),
|
|
(f"{p}bambustat", "Bambu Lab printer status + camera"),
|
|
(f"{p}info", "Bot info"),
|
|
(f"{p}last_error", "Last recorded error"),
|
|
]
|
|
rows = "".join(
|
|
f"<tr><td><code>{cmd}</code></td><td>{desc}</td></tr>"
|
|
for cmd, desc in commands
|
|
)
|
|
html = (
|
|
f"<h4>r0b3BOT Commands</h4>"
|
|
f"<table>{rows}</table>"
|
|
f"<p><em>Sound commands are Discord-only.</em></p>"
|
|
)
|
|
text_lines = [f" {cmd:<18} {desc}" for cmd, desc in commands]
|
|
text = "r0b3BOT Commands\n" + "\n".join(text_lines) + "\n (Sound commands are Discord-only)"
|
|
await self._send_text(room_id, text, html=html)
|
|
|
|
async def _cmd_printstat(self, room_id):
|
|
loop = asyncio.get_running_loop()
|
|
hass_headers = {
|
|
"Authorization": f"Bearer {self.hass_token}",
|
|
"content-type": "application/json",
|
|
}
|
|
hass_state_url = self.hass_url + "/api/states/" + self.hass_light
|
|
|
|
# Light control — same logic as the Discord version
|
|
turned_on_light = False
|
|
try:
|
|
resp = await loop.run_in_executor(
|
|
None, lambda: requests.get(hass_state_url, headers=hass_headers)
|
|
)
|
|
lights = resp.json()
|
|
if lights.get("state") == "off":
|
|
turned_on_light = True
|
|
await self._send_text(room_id, "Turning on a light, give me a moment...")
|
|
hass_on_url = self.hass_url + "/api/services/switch/turn_on"
|
|
payload = f'{{"entity_id": "{self.hass_light}"}}'
|
|
await loop.run_in_executor(
|
|
None, lambda: requests.post(hass_on_url, headers=hass_headers, data=payload)
|
|
)
|
|
await asyncio.sleep(3)
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Hass light check failed: {e}")
|
|
await self._send_text(room_id, "Hmm, I can't tell if the light is on... oh well [Hass API error]")
|
|
|
|
# Capture webcam snapshot
|
|
file_name = str(random.randrange(1, 10000)) + ".jpg"
|
|
try:
|
|
print(" - [Matrix] Getting image from Webcam")
|
|
await loop.run_in_executor(
|
|
None,
|
|
lambda: urllib.request.urlretrieve(
|
|
f"http://{self.octoprint_ip}:8080/?action=snapshot", file_name
|
|
),
|
|
)
|
|
await self._send_image(room_id, file_name)
|
|
os.remove(file_name)
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Webcam capture failed: {e}")
|
|
await self._send_text(room_id, f"Unable to capture snapshot: {e}")
|
|
|
|
# Turn the light back off if we turned it on
|
|
if turned_on_light:
|
|
try:
|
|
hass_off_url = self.hass_url + "/api/services/switch/turn_off"
|
|
payload = f'{{"entity_id": "{self.hass_light}"}}'
|
|
await loop.run_in_executor(
|
|
None, lambda: requests.post(hass_off_url, headers=hass_headers, data=payload)
|
|
)
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Unable to turn off light: {e}")
|
|
|
|
# Fetch printer stats from OctoPrint
|
|
try:
|
|
operational = await loop.run_in_executor(None, octoapi.get_printer_dict)
|
|
except Exception:
|
|
operational = False
|
|
|
|
if operational:
|
|
try:
|
|
print_filename = await loop.run_in_executor(None, octoapi.get_printFileName)
|
|
except Exception:
|
|
print_filename = ""
|
|
try:
|
|
print_completion = round(
|
|
await loop.run_in_executor(None, octoapi.get_completion), 2
|
|
)
|
|
except Exception:
|
|
print_completion = "Unknown"
|
|
try:
|
|
print_seconds = await loop.run_in_executor(None, octoapi.get_printTime)
|
|
print_secondsleft = await loop.run_in_executor(None, octoapi.get_printTimeLeft)
|
|
print_hours = int(print_seconds / 3600)
|
|
print_min = int((print_seconds / 60) - (print_hours * 60))
|
|
print_hoursleft = int(print_secondsleft / 3600)
|
|
print_minleft = int((print_secondsleft / 60) - (print_hoursleft * 60))
|
|
time_elapsed = f"{print_hours}h {print_min}m"
|
|
time_remaining = f"{print_hoursleft}h {print_minleft}m"
|
|
except Exception:
|
|
time_elapsed = time_remaining = "Unknown"
|
|
|
|
printer_bed = f"{operational['temperature']['bed']['actual']}°C"
|
|
printer_hotend = f"{operational['temperature']['tool0']['actual']}°C"
|
|
|
|
fields = [
|
|
("Progress", f"{print_completion}%"),
|
|
("Elapsed", time_elapsed),
|
|
("Remaining", time_remaining),
|
|
("Bed Temp", printer_bed),
|
|
("Hotend", printer_hotend),
|
|
]
|
|
await self._send_text(
|
|
room_id,
|
|
self._format_text("R0b3's 3D Printer Status", f"Current Job: {print_filename}", fields),
|
|
html=self._format_html("R0b3's 3D Printer Status", f"Current Job: {print_filename}", fields),
|
|
)
|
|
else:
|
|
await self._send_text(room_id, "Printer is Offline")
|
|
|
|
async def _cmd_bambustat(self, room_id):
|
|
status = self._get_bambu_status()
|
|
if status is None:
|
|
await self._send_text(
|
|
room_id,
|
|
"Bambu printer status is not available yet. The printer may be offline or still connecting.",
|
|
)
|
|
return
|
|
|
|
gcode_state = status.gcode_state or "Unknown"
|
|
gcode_file = status.gcode_file or "Unknown"
|
|
mc_percent = status.mc_percent if status.mc_percent is not None else "Unknown"
|
|
remaining_mins = status.mc_remaining_time or 0
|
|
time_remaining = f"{remaining_mins // 60}h {remaining_mins % 60:02d}m"
|
|
nozzle_temp = (
|
|
f"{status.nozzle_temper:.1f}°C" if status.nozzle_temper is not None else "Unknown"
|
|
)
|
|
bed_temp = (
|
|
f"{status.bed_temper:.1f}°C" if status.bed_temper is not None else "Unknown"
|
|
)
|
|
chamber_temp = (
|
|
f"{status.chamber_temper:.1f}°C" if status.chamber_temper is not None else "Unknown"
|
|
)
|
|
layer_info = "Unknown"
|
|
if status.layer_num is not None and status.total_layer_num is not None:
|
|
layer_info = f"{status.layer_num} / {status.total_layer_num}"
|
|
|
|
fields = [
|
|
("State", gcode_state),
|
|
("Progress", f"{mc_percent}%"),
|
|
("Remaining", time_remaining),
|
|
("Layer", layer_info),
|
|
("Nozzle", nozzle_temp),
|
|
("Bed", bed_temp),
|
|
("Chamber", chamber_temp),
|
|
]
|
|
await self._send_text(
|
|
room_id,
|
|
self._format_text("Bambu Lab Printer Status", f"File: {gcode_file}", fields),
|
|
html=self._format_html("Bambu Lab Printer Status", f"File: {gcode_file}", fields),
|
|
)
|
|
|
|
# Optional camera snapshot via ffmpeg RTSPS
|
|
if self.bambu_camera_enable and self._capture_bambu_frame:
|
|
image_filename = f"bambu_{random.randrange(1, 10000)}.jpg"
|
|
try:
|
|
print(" - [Matrix] Capturing Bambu camera frame via RTSPS")
|
|
success = await self._capture_bambu_frame(image_filename)
|
|
if success:
|
|
await self._send_image(room_id, image_filename)
|
|
os.remove(image_filename)
|
|
else:
|
|
print(" ! [Matrix] ffmpeg exited with non-zero code, skipping image")
|
|
except asyncio.TimeoutError:
|
|
print(" ! [Matrix] ffmpeg timed out capturing Bambu camera frame")
|
|
except Exception as e:
|
|
print(f" ! [Matrix] Unable to capture Bambu camera frame: {e}")
|