#!/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"{name}{value}" for name, value in fields ) desc_html = f"

{description}

" if description else "" return f"

{title}

{desc_html}{rows}
" 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'{source_url}'), ] 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"{cmd}{desc}" for cmd, desc in commands ) html = ( f"

r0b3BOT Commands

" f"{rows}
" f"

Sound commands are Discord-only.

" ) 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}")