381 lines
15 KiB
Python
381 lines
15 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 os
|
|
import random
|
|
import urllib.request
|
|
|
|
import octoapi
|
|
import requests
|
|
from nio import AsyncClient, MatrixRoom, RoomMessageText, UploadResponse
|
|
|
|
|
|
class MatrixBot:
|
|
def __init__(
|
|
self,
|
|
homeserver,
|
|
user_id,
|
|
access_token,
|
|
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,
|
|
):
|
|
self.client = AsyncClient(homeserver, user_id)
|
|
self.client.access_token = access_token
|
|
self.client.user_id = user_id
|
|
self.prefix = command_prefix
|
|
self.room_ids = set(room_ids)
|
|
|
|
# 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
|
|
|
|
async def start(self):
|
|
print(">> [Matrix] Starting Matrix bot client")
|
|
# Initial sync with no timeout to consume current state before registering
|
|
# callbacks — prevents the bot from replaying old messages on startup.
|
|
await self.client.sync(timeout=0, full_state=True)
|
|
self.client.add_event_callback(self._on_message, RoomMessageText)
|
|
print(">> [Matrix] Sync complete, listening for commands")
|
|
await self.client.sync_forever(timeout=30000)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Event handler
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def _on_message(self, room: MatrixRoom, event: RoomMessageText):
|
|
# Ignore own messages
|
|
if event.sender == self.client.user_id:
|
|
return
|
|
# Only respond in configured rooms
|
|
if room.room_id not in self.room_ids:
|
|
return
|
|
# Only respond to messages that start with the command prefix
|
|
if not event.body.startswith(self.prefix):
|
|
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 _send_text(self, room_id, text, html=None):
|
|
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):
|
|
with open(file_path, "rb") as f:
|
|
data = f.read()
|
|
resp = await self.client.upload(data, content_type="image/jpeg")
|
|
if isinstance(resp, UploadResponse):
|
|
await self.client.room_send(room_id, "m.room.message", {
|
|
"msgtype": "m.image",
|
|
"url": resp.content_uri,
|
|
"body": os.path.basename(file_path),
|
|
"info": {"mimetype": "image/jpeg"},
|
|
})
|
|
else:
|
|
print(f" ! [Matrix] Image upload failed: {resp}")
|
|
|
|
def _format_html(self, title, description, fields):
|
|
"""Render a Discord-embed-style status block as Matrix HTML."""
|
|
parts = [f"<strong>{title}</strong>"]
|
|
if description:
|
|
parts.append(description)
|
|
for name, value in fields:
|
|
parts.append(f"<strong>{name}</strong> {value}")
|
|
return "<br>".join(parts)
|
|
|
|
def _format_text(self, title, description, fields):
|
|
"""Plain-text fallback for the same status block."""
|
|
parts = [title]
|
|
if description:
|
|
parts.append(description)
|
|
for name, value in fields:
|
|
parts.append(f"{name} {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):
|
|
await self._send_text(
|
|
room_id,
|
|
"r0b3BOT — a derpy derp of a bot.\n"
|
|
"Source: https://gitlab.rickelobe.com/Bots/r0b3BOT",
|
|
)
|
|
|
|
async def _cmd_help(self, room_id):
|
|
p = self.prefix
|
|
html_lines = [
|
|
"<strong>Commands (Matrix)</strong>",
|
|
f"<strong>{p}greetings</strong> — Hello!",
|
|
f"<strong>{p}cat</strong> / <strong>{p}explain</strong> / <strong>{p}holyshit</strong> — GIF commands",
|
|
f"<strong>{p}smudge</strong> — Sends the Smudge image",
|
|
f"<strong>{p}printstat</strong> — OctoPrint 3D printer status + snapshot",
|
|
f"<strong>{p}bambustat</strong> — Bambu Lab printer status",
|
|
f"<strong>{p}info</strong> — Bot info",
|
|
f"<strong>{p}last_error</strong> — Last recorded error",
|
|
f"<em>Sound commands are Discord-only (voice channels).</em>",
|
|
]
|
|
# Strip HTML tags for the plain-text body
|
|
import re
|
|
text_lines = [re.sub(r"<[^>]+>", "", l) for l in html_lines]
|
|
await self._send_text(
|
|
room_id,
|
|
"\n".join(text_lines),
|
|
html="<br>".join(html_lines),
|
|
)
|
|
|
|
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 = [
|
|
("Percent Complete:", f"{print_completion}%"),
|
|
("Time Elapsed:", time_elapsed),
|
|
("Time Remaining:", time_remaining),
|
|
("Bed Temp:", printer_bed),
|
|
("Hotend Temp:", 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),
|
|
("Percent Complete:", f"{mc_percent}%"),
|
|
("Time Remaining:", time_remaining),
|
|
("Layer:", layer_info),
|
|
("Nozzle Temp:", nozzle_temp),
|
|
("Bed Temp:", bed_temp),
|
|
("Chamber Temp:", 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 — reuse the same function from r0b3bot
|
|
if self.bambu_camera_enable and self.bambu_ip and self.bambu_access_code:
|
|
from r0b3bot import _capture_bambu_frame
|
|
image_filename = f"bambu_{random.randrange(1, 10000)}.jpg"
|
|
try:
|
|
print(" - [Matrix] Capturing Bambu camera frame via RTSPS")
|
|
success = await _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}")
|