885 lines
37 KiB
Python
885 lines
37 KiB
Python
#!/usr/bin/python3
|
||
|
||
import discord
|
||
from discord.ext import commands, tasks
|
||
from datetime import datetime
|
||
import urllib.request
|
||
import random
|
||
import os
|
||
import octoapi
|
||
import time
|
||
import configparser
|
||
import json
|
||
import requests
|
||
import asyncio
|
||
from bambu_connect import BambuClient
|
||
|
||
# Monkey-patch every dataclass in bambu-connect's models module to tolerate
|
||
# unknown fields. The Bambu printer firmware sends extra keys that the
|
||
# bambu-connect 0.3.1 models don't define (e.g. Upload.file_size,
|
||
# Online.ext), causing TypeErrors that crash the MQTT thread.
|
||
# Filtering kwargs to known fields before calling the original __init__
|
||
# fixes this for all nested classes at once.
|
||
import dataclasses as _dc
|
||
import inspect as _inspect
|
||
from bambu_connect.utils import models as _bambu_models
|
||
|
||
def _make_permissive(cls):
|
||
known = {f.name for f in _dc.fields(cls)}
|
||
orig = cls.__init__
|
||
def __init__(self, **kwargs):
|
||
orig(self, **{k: v for k, v in kwargs.items() if k in known})
|
||
cls.__init__ = __init__
|
||
|
||
for _name, _cls in _inspect.getmembers(_bambu_models, _inspect.isclass):
|
||
if _dc.is_dataclass(_cls):
|
||
_make_permissive(_cls)
|
||
|
||
#Record the time the bot started
|
||
start_time = time.time()
|
||
|
||
config = configparser.RawConfigParser()
|
||
configFilePath = r'bot_config.conf'
|
||
|
||
try:
|
||
config.read(configFilePath)
|
||
except:
|
||
print("Error loading config! Please check config file 'bot-config.conf'")
|
||
|
||
# Last error message variable used by $last_error command to display in discord
|
||
last_error_msg = "No errors have been recorded."
|
||
|
||
#
|
||
# Assign bot config from setting in config file
|
||
#
|
||
|
||
# Assign Bot core settings
|
||
DISCORD_AUTH_TOKEN = config.get('bot-config', 'discord_auth_token')
|
||
BOT_COMMAND_PREFIX = config.get('bot-config', 'bot_command_prefix')
|
||
|
||
# 3D Printer Setup - HomeAssistant URL and Headers & OctoPrint
|
||
HASS_TOKEN = config.get('bot-config', 'hass_token')
|
||
HASS_URL = config.get('bot-config', 'hass_url')
|
||
HASS_LIGHT = config.get('bot-config', 'hass_light')
|
||
hassURL = HASS_URL + "/api/states/" + HASS_LIGHT
|
||
hassHEADERS = {
|
||
'Authorization': f"Bearer {HASS_TOKEN}",
|
||
'content-type': 'application/json',
|
||
}
|
||
OCTOPRINT_IP_ADDRESS = config.get('bot-config', 'octoprint_ip_address')
|
||
|
||
# Bambu Lab Printer config (optional — feature is skipped if not set)
|
||
BAMBU_IP_ADDRESS = config.get('bot-config', 'bambu_ip_address', fallback=None)
|
||
BAMBU_ACCESS_CODE = config.get('bot-config', 'bambu_access_code', fallback=None)
|
||
BAMBU_SERIAL_NUMBER = config.get('bot-config', 'bambu_serial_number', fallback=None)
|
||
BAMBU_CAMERA_ENABLE = config.get('bot-config', 'bambu_camera_enable', fallback='false').lower() == 'true'
|
||
|
||
# Module-level cache for Bambu printer status, updated by the MQTT watch callback
|
||
bambu_status = None
|
||
bambu_client = None
|
||
_bot_loop = None # set in on_ready; used by the MQTT thread
|
||
_last_bambu_activity_update = 0.0
|
||
_bambu_was_printing = False
|
||
|
||
|
||
def _on_bambu_update(status):
|
||
"""Callback invoked by bambu-connect's MQTT thread with the latest PrinterStatus."""
|
||
global bambu_status, _last_bambu_activity_update, _bambu_was_printing
|
||
bambu_status = status
|
||
|
||
if _bot_loop is None:
|
||
return
|
||
|
||
now = time.time()
|
||
if status.gcode_state == "RUNNING":
|
||
_bambu_was_printing = True
|
||
# Throttle Discord presence updates to once every 30 seconds
|
||
if (now - _last_bambu_activity_update) >= 30:
|
||
_last_bambu_activity_update = now
|
||
asyncio.run_coroutine_threadsafe(_update_bambu_activity(status), _bot_loop)
|
||
elif _bambu_was_printing:
|
||
# State just left RUNNING (paused, finished, failed) — reset presence
|
||
_bambu_was_printing = False
|
||
asyncio.run_coroutine_threadsafe(_reset_bot_activity(), _bot_loop)
|
||
|
||
|
||
# Configure bot
|
||
intents = discord.Intents.default()
|
||
intents.message_content = True
|
||
intents.members = True
|
||
intents.voice_states = True
|
||
bot = commands.Bot(command_prefix=BOT_COMMAND_PREFIX, description='A derpy derp of a bot.', intents=intents)
|
||
bot.remove_command('help')
|
||
|
||
# Load opus once at startup rather than on every sound command
|
||
discord.opus.load_opus("libopus.so")
|
||
|
||
# Printing configuration details to console
|
||
print("=== 3D Printer Status config ===")
|
||
print(f" HomeAssistant URL: {hassURL}")
|
||
print(f" HomeAssistant URL HEADERS: {hassHEADERS}")
|
||
print(f" HomeAssistant Light: {HASS_LIGHT}")
|
||
print(f" OctoPrint IP Address: {OCTOPRINT_IP_ADDRESS}")
|
||
|
||
async def _update_bambu_activity(status):
|
||
remaining_mins = status.mc_remaining_time or 0
|
||
remaining_hours = remaining_mins // 60
|
||
remaining_mins_part = remaining_mins % 60
|
||
activity = discord.Activity(
|
||
name=f"Bambu @ {status.mc_percent}%, {remaining_hours}h{remaining_mins_part:02d}m to go",
|
||
type=discord.ActivityType.watching,
|
||
)
|
||
await bot.change_presence(activity=activity)
|
||
|
||
|
||
async def _capture_bambu_frame(filename):
|
||
"""Capture a single JPEG frame from the Bambu RTSPS stream via ffmpeg.
|
||
Returns True on success, raises on timeout or ffmpeg error."""
|
||
rtsp_url = f"rtsps://bblp:{BAMBU_ACCESS_CODE}@{BAMBU_IP_ADDRESS}/streaming/live/1"
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"ffmpeg", "-y",
|
||
"-rtsp_transport", "tcp",
|
||
"-i", rtsp_url,
|
||
"-vframes", "1",
|
||
"-q:v", "2",
|
||
filename,
|
||
stdout=asyncio.subprocess.DEVNULL,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
_, stderr = await asyncio.wait_for(proc.communicate(), timeout=15)
|
||
if proc.returncode != 0:
|
||
print(f" ! ffmpeg stderr: {stderr.decode(errors='replace').strip()}")
|
||
return proc.returncode == 0
|
||
|
||
|
||
async def _reset_bot_activity():
|
||
activity = discord.Activity(
|
||
name=f"your {BOT_COMMAND_PREFIX}commands",
|
||
type=discord.ActivityType.listening,
|
||
)
|
||
await bot.change_presence(activity=activity)
|
||
|
||
|
||
# Background task: poll OctoPrint every 2 minutes and update bot status if printing
|
||
@tasks.loop(seconds=120)
|
||
async def check_printer_status():
|
||
try:
|
||
print(">> Checking 3D Printer Status")
|
||
# Run the synchronous octoapi call in a thread so it doesn't block
|
||
# the event loop and starve Discord's heartbeat on slow/unreachable hosts
|
||
loop = asyncio.get_running_loop()
|
||
print_completion = round(await loop.run_in_executor(None, octoapi.get_completion), 2)
|
||
except Exception:
|
||
print_completion = "Unknown"
|
||
print(">>>> Nothing Printing.")
|
||
if print_completion != "Unknown" and print_completion < 100:
|
||
print(">>>> Something is printing, calling UpdateStatus")
|
||
await updateStatus()
|
||
else:
|
||
# Will hit this if the last print job completed and is sitting at 100%
|
||
print(">>>> Nothing Printing.")
|
||
|
||
|
||
# On Ready
|
||
@bot.event
|
||
async def on_ready():
|
||
print("\n:::: Config Loaded ::::\n")
|
||
print('Logged in as')
|
||
print(f" User: {bot.user.name}")
|
||
print(f" ID: {bot.user.id}")
|
||
print(f"\n\n")
|
||
print(f"Setting activity to 'Listenting to your {BOT_COMMAND_PREFIX}commands'")
|
||
activity = discord.Activity(name=f"your {BOT_COMMAND_PREFIX}commands",type=discord.ActivityType.listening)
|
||
await bot.change_presence(activity=activity)
|
||
|
||
print(">> Startup complete.")
|
||
|
||
# Clear any stale voice states left over from a previous session.
|
||
# Discord holds onto the bot's voice state across restarts; if not cleared,
|
||
# the first voice connect attempt gets a 4006 (session no longer valid).
|
||
print(">> Clearing stale voice states across all guilds")
|
||
for guild in bot.guilds:
|
||
try:
|
||
await guild.change_voice_state(channel=None)
|
||
except Exception as e:
|
||
print(f" ! Could not clear voice state for {guild.name}: {e}")
|
||
|
||
# Start the printer polling task (guard against double-start on reconnects)
|
||
if not check_printer_status.is_running():
|
||
check_printer_status.start()
|
||
|
||
# Store the running event loop so the Bambu MQTT thread can schedule coroutines
|
||
global _bot_loop
|
||
_bot_loop = asyncio.get_running_loop()
|
||
|
||
# Start the Bambu watch client if config is present
|
||
global bambu_client
|
||
if BAMBU_IP_ADDRESS and BAMBU_ACCESS_CODE and BAMBU_SERIAL_NUMBER:
|
||
if bambu_client is None:
|
||
print(">> Connecting to Bambu printer MQTT watch client")
|
||
bambu_client = BambuClient(BAMBU_IP_ADDRESS, BAMBU_ACCESS_CODE, BAMBU_SERIAL_NUMBER)
|
||
bambu_client.start_watch_client(_on_bambu_update)
|
||
print(">> Bambu watch client started")
|
||
else:
|
||
print(">> Bambu printer config not found, skipping Bambu watch client")
|
||
|
||
|
||
@bot.command()
|
||
async def greetings(ctx):
|
||
await ctx.send(":smiley: :wave: Hello there!")
|
||
|
||
|
||
@bot.event
|
||
async def on_command_error(ctx, error):
|
||
global last_error_msg
|
||
last_error_msg = error
|
||
if isinstance(error, commands.CommandOnCooldown):
|
||
seconds = round(error.retry_after,1)
|
||
print(f" - Error, user tried to use command while in cooldown (wait is: {seconds})")
|
||
await ctx.send(f"This command is ratelimited per user, please try again in {seconds}s")
|
||
elif isinstance(error, commands.CommandInvokeError):
|
||
print(" - ERROR: encountered 'CommandInvokeError'")
|
||
print(f" - Dumping error: {error}")
|
||
await ctx.send("Error, unable to complete your request.")
|
||
else:
|
||
print('Oopsie, I found an error...')
|
||
print(f"Error: {error}")
|
||
|
||
#
|
||
# Debug commands
|
||
#
|
||
@bot.command()
|
||
async def last_error(ctx):
|
||
await ctx.send(f"Last error recorded: {last_error_msg}")
|
||
|
||
#
|
||
# Giphy Commands
|
||
#
|
||
@bot.command()
|
||
async def cat(ctx):
|
||
await ctx.send("https://media.giphy.com/media/JIX9t2j0ZTN9S/giphy.gif")
|
||
|
||
@bot.command()
|
||
async def explain(ctx):
|
||
await ctx.send("https://media.giphy.com/media/A8NNZlVuA1LoY/giphy.gif")
|
||
|
||
@bot.command()
|
||
async def holyshit(ctx):
|
||
await ctx.send("https://media.giphy.com/media/QcGTswMKyBNgA/giphy.gif")
|
||
|
||
###
|
||
|
||
#
|
||
# Image Commands
|
||
#
|
||
@bot.command()
|
||
async def smudge(ctx):
|
||
print(" - Uploading Smudge to chat channel")
|
||
file = discord.File("./images/smudge.png", filename="smudge.png")
|
||
await ctx.send(file=file)
|
||
|
||
#
|
||
# Voice Commands
|
||
#
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=30.0, type=commands.BucketType.user)
|
||
async def bitch(ctx, member : discord.Member="NONE"):
|
||
datestring = datetime.now()
|
||
datestring = datestring.strftime("%m/%d/%Y-%H:%M:%S")
|
||
if ctx.message.channel.is_nsfw():
|
||
await play_sound(ctx, member, "./sounds/monkey_bitch.mp3", "$bitch")
|
||
else:
|
||
print(f"[{datestring}]: {ctx.message.author.display_name} called $bitch, but is not in a NSFW channel")
|
||
await ctx.send("This command is too explicit for you!")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def cowbell(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/More_cowbell.mp3", "$cowbell")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def boom(ctx, member : discord.Member="NONE"):
|
||
datestring = datetime.now()
|
||
datestring = datestring.strftime("%m/%d/%Y-%H:%M:%S")
|
||
if ctx.message.channel.is_nsfw():
|
||
await play_sound(ctx, member, "./sounds/BoomBitch.mp3", "$boom")
|
||
else:
|
||
print(f"[{datestring}]: {ctx.message.author.display_name} called $boom, but is not in a NSFW channel")
|
||
await ctx.send("This command is too explicit for you!")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def oops(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/Oops.mp3", "$oops")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def trololo(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/Trololo.mp3", "$trololo")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def leeroy(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/leeroy.mp3", "$leeroy")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def promoted(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/Promoted.mp3", "$promoted")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def wow(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/wow.mp3", "$wow")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def aww(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/Awww_Bitch.mp3", "$aww")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def eia(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/awesome.mp3", "$eia")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def yaw(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/yaw.mp3", "$yaw")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def heavy(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/heavy.mp3", "$heavy")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def iran(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/iran.mp3", "$iran")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def big(ctx, member : discord.Member="NONE"):
|
||
await play_sound(ctx, member, "./sounds/big.mp3", "$big")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def breee(ctx, member : discord.Member="NONE"):
|
||
await ctx.send(f"\m/ ( ﾟДﾟ)＜BREEEEE!! \m/")
|
||
await play_sound(ctx, member, "./sounds/wbbwb_metal.mp3", "$breee")
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def nice(ctx, member : discord.Member="NONE"):
|
||
await ctx.send(f"\m/ ( ﾟДﾟ)＜N!CE! \m/")
|
||
await play_sound(ctx, member, "./sounds/wbtbwb-nice.mp3", "$nice")
|
||
|
||
|
||
|
||
@bot.command()
|
||
@commands.cooldown(rate=1, per=10.0, type=commands.BucketType.user)
|
||
async def rs(ctx, member : discord.Member="NONE"):
|
||
datestring = datetime.now()
|
||
datestring = datestring.strftime("%m/%d/%Y-%H:%M:%S")
|
||
if ctx.message.channel.is_nsfw():
|
||
soundfiles = ["./sounds/leeroy.mp3","./sounds/monkey_bitch.mp3","./sounds/More_cowbell.mp3","./sounds/BoomBitch.mp3","./sounds/Promoted.mp3","./sounds/Trololo.mp3","./sounds/Oops.mp3"]
|
||
await play_sound(ctx, member, random.choice(soundfiles), "$rs")
|
||
else:
|
||
print(f"[{datestring}]: {ctx.message.author.display_name} called $rs, but is not in a NSFW channel")
|
||
await ctx.send("This command is too explicit for you!")
|
||
|
||
async def play_sound(ctx, member : discord.Member, soundFile, command):
|
||
|
||
datestring = datetime.now()
|
||
datestring = datestring.strftime("%m/%d/%Y-%H:%M:%S")
|
||
|
||
print(f"[{datestring}]: {ctx.message.author.display_name} called {command}")
|
||
|
||
# Need to check whether or not a user has been specified. If one has, we need to use the channel
|
||
# they are currently connected to instead the channel that the command issuer is in.
|
||
if member != "NONE":
|
||
try:
|
||
channel = member.voice.channel
|
||
except AttributeError:
|
||
await ctx.send(f"{member.display_name} is not in a voice channel!")
|
||
print(f" ! {member.display_name} is not in a voice channel")
|
||
return
|
||
print(f" - User specified: {member.display_name}")
|
||
print(f" - Destination channel: {channel}")
|
||
|
||
if member == "NONE":
|
||
try:
|
||
channel = ctx.message.author.voice.channel
|
||
except AttributeError:
|
||
await ctx.send("You are not connected to a voice channel.")
|
||
print("User is not connected to a voice channel")
|
||
return
|
||
|
||
voice: discord.VoiceClient = discord.utils.get(bot.voice_clients, guild=ctx.guild)
|
||
|
||
# Different join actions are needed depending on whether or not the bot is already in a
|
||
# voice channel. If already in a channel, wait until disconnected, then join the requested
|
||
# channel. If not already in a channel, then join requested channel immediately
|
||
if voice and voice.is_connected():
|
||
print(f" - Moving to channel '{channel}''")
|
||
await ctx.send(f"Moving to '{channel}' for just a moment")
|
||
await voice.move_to(channel)
|
||
|
||
else:
|
||
print(f" - Joining Channel '{channel}''")
|
||
await ctx.send(f"Joining '{channel}' for just a moment")
|
||
# reconnect=False disables discord.py's internal retry loop so it doesn't
|
||
# hammer the same invalid session repeatedly (which was causing "attempt 4").
|
||
voice = await channel.connect(reconnect=False)
|
||
|
||
print(" - Creating Player")
|
||
audio_source = discord.FFmpegPCMAudio(soundFile)
|
||
print(f" - Playing {soundFile}")
|
||
await asyncio.sleep(1)
|
||
voice.play(audio_source, after=None)
|
||
while voice.is_playing():
|
||
await asyncio.sleep(1)
|
||
print(f" - Disconnecting from '{channel}'")
|
||
await voice.disconnect()
|
||
###
|
||
|
||
#
|
||
# OctoPrint Commands
|
||
#
|
||
@bot.command()
|
||
async def printstat(ctx):
|
||
|
||
datestring = datetime.now()
|
||
datestring = datestring.strftime("%m/%d/%Y-%H:%M:%S")
|
||
|
||
print(f"[{datestring}]: {ctx.message.author.display_name} called $printstat")
|
||
|
||
# We haven't yet changed the state of the light, setting this to False
|
||
turned_on_light = False
|
||
|
||
# Try to get the status of the 3D printer light
|
||
print(" - Getting Printer Light Status")
|
||
loop = asyncio.get_running_loop()
|
||
work_lights_resp = await loop.run_in_executor(None, lambda: requests.get(hassURL, headers=hassHEADERS))
|
||
print(f" JSON Response from HomeAssistant: {work_lights_resp.text}")
|
||
|
||
work_lights = json.loads(work_lights_resp.text)
|
||
|
||
# Making sure that the API request succeeded. If it has, there will be a state attribute added to
|
||
# work_lights.
|
||
try:
|
||
work_lights["state"]
|
||
except KeyError:
|
||
print(" ! Unable to get light status from Homeassistant, notifying chat channel")
|
||
await ctx.send("Hmm, I can't tell if the light is on... oh well [Hass API error]")
|
||
|
||
# Create empty dict w/ state of unknown
|
||
work_lights = {'state': "unknown"}
|
||
|
||
# If the light is off, let's turn it on before we take a picture
|
||
if work_lights["state"] == 'off':
|
||
print(" - Lights are off, turning on for image capture")
|
||
turned_on_light = True
|
||
|
||
lightMessages = ["Woah, it's pretty dark in R0b3's Basement... Give me a couple seconds to turn on a light.", \
|
||
"Turning on a light, give me a moment...", \
|
||
"I can't see a thing, let me get a light...", \
|
||
"Where's that light switch?...", \
|
||
"Uh, the lights are off. Give me a sec...", \
|
||
"Replacing the light bulb, just a moment..."]
|
||
|
||
await ctx.send(f"{random.choice(lightMessages)}")
|
||
|
||
try:
|
||
# REST API call to home assistant to turn the light on.
|
||
hassAPIURL = HASS_URL + "/api/services/switch/turn_on"
|
||
payload = f'{{"entity_id": "{HASS_LIGHT}"}}'
|
||
await loop.run_in_executor(None, lambda: requests.post(hassAPIURL, headers=hassHEADERS, data=payload))
|
||
|
||
except Exception as e:
|
||
print(f" ! Unable to turn on light; ERROR: {e}")
|
||
await ctx.send("Sorry, I was unable to turn on the light.")
|
||
pass
|
||
|
||
await asyncio.sleep(3)
|
||
|
||
# Randomly generate a filename to save the image to
|
||
file_name = random.randrange(1,10000)
|
||
full_file_name = str(file_name) + '.jpg'
|
||
|
||
# Get an image from OctoPrint and save it
|
||
print(" - Getting image from Webcam")
|
||
await loop.run_in_executor(None, lambda: urllib.request.urlretrieve(
|
||
"http://%s:8080/?action=snapshot" % (OCTOPRINT_IP_ADDRESS), full_file_name))
|
||
|
||
# If we turned on the light, let's be nice and turn it back off
|
||
if turned_on_light:
|
||
print(" - Turning Light back Off")
|
||
try:
|
||
# REST API call to home assistant to turn the light off.
|
||
hassAPIURL = HASS_URL + "/api/services/switch/turn_off"
|
||
payload = f'{{"entity_id": "{HASS_LIGHT}"}}'
|
||
await loop.run_in_executor(None, lambda: requests.post(hassAPIURL, headers=hassHEADERS, data=payload))
|
||
|
||
except Exception as e:
|
||
print(f" ! Unable to turn on light; ERROR: {e}")
|
||
print(" ! Unable to turn off light")
|
||
pass
|
||
|
||
# Send the image to the chat channel
|
||
print(" - Uploading image to chat channel")
|
||
file = discord.File(full_file_name, filename=full_file_name)
|
||
await ctx.send("3D Printer Snapshot:", file=file)
|
||
|
||
# Remove the Image File now that it is no longer needed
|
||
print(" - Deleting temporary image file")
|
||
os.remove(full_file_name)
|
||
|
||
print(" - Connecting to OctoPrint API")
|
||
# Testing OctoPrint API Status. If printer is off, I won't receive JSON data.
|
||
try:
|
||
operational = await loop.run_in_executor(None, octoapi.get_printer_dict)
|
||
except Exception:
|
||
# Printer is Offline
|
||
operational = False
|
||
|
||
if operational:
|
||
print(" - 3D Printer is on and printing, getting details")
|
||
|
||
# Get the name of the file currently being printed
|
||
try:
|
||
print(" - Getting Filename")
|
||
print_filename = await loop.run_in_executor(None, octoapi.get_printFileName)
|
||
except Exception:
|
||
print_filename = ""
|
||
|
||
try:
|
||
print(" - Getting % Completion")
|
||
# Try to get the % of Completion
|
||
print_completion = round(await loop.run_in_executor(None, octoapi.get_completion), 2)
|
||
except Exception:
|
||
print_completion = "Unknown"
|
||
|
||
# Get the current print time in seconds
|
||
try:
|
||
print(" - Getting Seconds Elapsed")
|
||
print_seconds = await loop.run_in_executor(None, octoapi.get_printTime)
|
||
except Exception:
|
||
print("Unable to get Seconds data from OctoPrint")
|
||
print_seconds = 0
|
||
|
||
# Get the time remaining for current print job in seconds
|
||
try:
|
||
print(" - Getting Seconds Left")
|
||
print_secondsleft = await loop.run_in_executor(None, octoapi.get_printTimeLeft)
|
||
except Exception:
|
||
print(" ! Unable to get Seconds Left from OctoPrint")
|
||
print_secondsleft = 0
|
||
|
||
try:
|
||
print(" - Converting Seconds to Hours")
|
||
# Try to convert seconds to hours
|
||
print_hours = int(((print_seconds / 60) / 60))
|
||
except (TypeError, ValueError):
|
||
print_hours = "Unknown"
|
||
|
||
try:
|
||
print(" - Converting Seconds left to Hours left")
|
||
# Try to convert seconds left to hours left
|
||
print_hoursleft = int(((print_secondsleft / 60) / 60))
|
||
except (TypeError, ValueError):
|
||
print_hoursleft = "Unknown"
|
||
|
||
try:
|
||
print(" - Calculating Minutes Elapsed")
|
||
# Using the total number of minutes minus the total number of whole hours to get the minutes remaining
|
||
print_min = int(((print_seconds / 60) - (print_hours * 60)))
|
||
except (TypeError, ValueError):
|
||
print_min = "Unknown"
|
||
|
||
try:
|
||
print(" - Calculating Minutes Left")
|
||
# Same as above but for the minutes left in print job
|
||
print_minleft = int(((print_secondsleft / 60) - (print_hoursleft * 60)))
|
||
except (TypeError, ValueError):
|
||
print_minleft = "Unknown"
|
||
|
||
time_elapsed = "%s Hours %s Minutes" % (print_hours, print_min)
|
||
time_remaining = "%s Hours %s Minutes" % (print_hoursleft, print_minleft)
|
||
|
||
# operational dict already fetched above — use it directly instead of calling again
|
||
printer_bed = "%s°C" % (operational["temperature"]["bed"]["actual"])
|
||
printer_hotend = "%s°C" % (operational["temperature"]["tool0"]["actual"])
|
||
|
||
embed = discord.Embed(title="R0b3's 3D Printer Status", description="Current Job: " + print_filename, color=0xF5A623)
|
||
|
||
embed.add_field(name="Percent Complete: ", value=str(print_completion) + "%")
|
||
embed.add_field(name="Time Elapsed: ", value=time_elapsed)
|
||
embed.add_field(name="Time Remaining: ", value=time_remaining)
|
||
embed.add_field(name="Bed Temp.: ", value=printer_bed)
|
||
embed.add_field(name="Hotend Temp.: ", value=printer_hotend)
|
||
embed.add_field(name="Livestream URL: ",value="https://rickelobe.com/printer/?action=stream")
|
||
print(" - Sending details to chat channel")
|
||
await ctx.send(embed=embed)
|
||
|
||
## No longer need to update the bot status here since there's a check happening every 2 mins in on_ready
|
||
#activity = discord.Activity(name=f"3D Print @ {str(print_completion)}%",type=discord.ActivityType.watching)
|
||
#await bot.change_presence(activity=activity)
|
||
#await updateStatus() # No longer need to call this since it the bot now checks for a print job every 2 mins.
|
||
|
||
else:
|
||
print(" - 3D Printer is not on or is not printing")
|
||
embed = discord.Embed(title="Printer is Offline")
|
||
print(" - Notifying chat channel")
|
||
await ctx.send(embed=embed)
|
||
|
||
async def updateStatus():
|
||
|
||
# Initialize variables
|
||
# Unknowns is present to handle issues like print paused or cancelled
|
||
print_completion = 0
|
||
unknowns = 0
|
||
loop = asyncio.get_running_loop()
|
||
#While the print status is less than 100 % complete, keep updating
|
||
while print_completion < 100:
|
||
try:
|
||
# Try to get the % of Completion
|
||
print_completion = round(await loop.run_in_executor(None, octoapi.get_completion), 1)
|
||
except Exception:
|
||
print_completion = 999
|
||
|
||
try:
|
||
print_secondsleft = await loop.run_in_executor(None, octoapi.get_printTimeLeft)
|
||
except Exception:
|
||
print(" ! Unable to get Seconds Left from OctoPrint")
|
||
print_secondsleft = 0
|
||
|
||
try:
|
||
# Try to convert seconds left to hours left
|
||
print_hoursleft = int(((print_secondsleft / 60) / 60))
|
||
except (TypeError, ValueError):
|
||
print_hoursleft = "Unknown"
|
||
|
||
try:
|
||
# Same as above but for the minutes left in print job
|
||
print_minleft = int(((print_secondsleft / 60) - (print_hoursleft * 60)))
|
||
except (TypeError, ValueError):
|
||
print_minleft = "Unknown"
|
||
|
||
if print_minleft == "Unknown" or print_hoursleft == "Unknown":
|
||
time_left = "Unknown"
|
||
else:
|
||
time_left = f"{print_hoursleft:02d}:{print_minleft:02d}"
|
||
|
||
#Set the activity to the new percent complete value
|
||
if print_completion != 999:
|
||
unknowns = 0
|
||
print(f" - Print job is at {print_completion}%, with {time_left} to go")
|
||
activity = discord.Activity(name=f"3D Print @ {str(print_completion)}%, {time_left} to go",type=discord.ActivityType.watching)
|
||
else:
|
||
unknowns += 1
|
||
print(f" - Error getting print status ({unknowns}/5 Errors)")
|
||
activity = discord.Activity(name=f"3D Print @ ERROR",type=discord.ActivityType.watching)
|
||
|
||
await bot.change_presence(activity=activity)
|
||
|
||
if (unknowns >= 5):
|
||
print(" - Reached 5 consecutive errors, breaking from loop")
|
||
break
|
||
await asyncio.sleep(30)
|
||
|
||
# Flash the status @ 100% before changing back to listening to your commands, only if uknowns !5.
|
||
if unknowns != 5:
|
||
for x in range(20):
|
||
activity = discord.Activity(name="-",type=discord.ActivityType.watching)
|
||
await bot.change_presence(activity=activity)
|
||
await asyncio.sleep(1)
|
||
activity = discord.Activity(name=f"3D Print @ {str(print_completion)}%",type=discord.ActivityType.watching)
|
||
await bot.change_presence(activity=activity)
|
||
await asyncio.sleep(1)
|
||
|
||
print(" - Done watching 3D Printer status")
|
||
await asyncio.sleep(3)
|
||
activity = discord.Activity(name=f"your {BOT_COMMAND_PREFIX}commands",type=discord.ActivityType.listening)
|
||
await bot.change_presence(activity=activity)
|
||
|
||
#
|
||
# Bambu Lab Printer Commands
|
||
#
|
||
@bot.command()
|
||
async def bambustat(ctx):
|
||
|
||
datestring = datetime.now().strftime("%m/%d/%Y-%H:%M:%S")
|
||
print(f"[{datestring}]: {ctx.message.author.display_name} called $bambustat")
|
||
|
||
if bambu_status is None:
|
||
await ctx.send("Bambu printer status is not available yet. The printer may be offline or still connecting.")
|
||
return
|
||
|
||
status = bambu_status
|
||
|
||
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"
|
||
|
||
# mc_remaining_time is in minutes
|
||
remaining_mins = status.mc_remaining_time or 0
|
||
remaining_hours = remaining_mins // 60
|
||
remaining_mins_part = remaining_mins % 60
|
||
time_remaining = f"{remaining_hours}h {remaining_mins_part: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}"
|
||
|
||
embed = discord.Embed(
|
||
title="Bambu Lab Printer Status",
|
||
description=f"File: {gcode_file}",
|
||
color=0x00B4D8,
|
||
)
|
||
embed.add_field(name="State:", value=gcode_state, inline=False)
|
||
embed.add_field(name="Percent Complete:", value=f"{mc_percent}%")
|
||
embed.add_field(name="Time Remaining:", value=time_remaining)
|
||
embed.add_field(name="Layer:", value=layer_info)
|
||
embed.add_field(name="Nozzle Temp:", value=nozzle_temp)
|
||
embed.add_field(name="Bed Temp:", value=bed_temp)
|
||
embed.add_field(name="Chamber Temp:", value=chamber_temp)
|
||
|
||
# Try to grab a snapshot via ffmpeg RTSPS (only if enabled in config)
|
||
image_filename = None
|
||
if BAMBU_CAMERA_ENABLE and BAMBU_IP_ADDRESS and BAMBU_ACCESS_CODE:
|
||
image_filename = f"bambu_{random.randrange(1, 10000)}.jpg"
|
||
try:
|
||
print(" - Capturing Bambu camera frame via RTSPS")
|
||
success = await _capture_bambu_frame(image_filename)
|
||
if not success:
|
||
print(" ! ffmpeg exited with non-zero code, skipping image")
|
||
image_filename = None
|
||
except asyncio.TimeoutError:
|
||
print(" ! ffmpeg timed out capturing Bambu camera frame")
|
||
image_filename = None
|
||
except Exception as e:
|
||
print(f" ! Unable to capture Bambu camera frame: {e}")
|
||
image_filename = None
|
||
|
||
print(" - Sending Bambu details to chat channel")
|
||
if image_filename:
|
||
file = discord.File(image_filename, filename=image_filename)
|
||
await ctx.send(file=file, embed=embed)
|
||
os.remove(image_filename)
|
||
else:
|
||
await ctx.send(embed=embed)
|
||
|
||
|
||
@bot.command()
|
||
async def info(ctx):
|
||
embed = discord.Embed(title="R0b3BOT", description="Derpy Derp of a Bot :P", color=0xeee657)
|
||
|
||
# give info about you here
|
||
embed.add_field(name="Author", value="R0b3")
|
||
|
||
# Shows the number of servers the bot is member of.
|
||
embed.add_field(name="Server count", value=f"{len(bot.guilds)}")
|
||
|
||
# give users a link to invite thsi bot to their server
|
||
embed.add_field(name="Invite", value="[Invite link](https://discordapp.com/api/oauth2/authorize?client_id=416312716677087233&permissions=51264&scope=bot)")
|
||
|
||
# Bot Uptime
|
||
second = time.time() - start_time
|
||
minute, second = divmod(second, 60)
|
||
hour, minute = divmod(minute, 60)
|
||
day, hour = divmod(hour, 24)
|
||
week, day = divmod(day, 7)
|
||
week = int(week)
|
||
day = int(day)
|
||
hour = int(hour)
|
||
minute = int(minute)
|
||
second = int(second)
|
||
embed.add_field(name="Uptime [WW:DD:HH:MM:SS]:", value=f"{week:02d}:{day:02d}:{hour:02d}:{minute:02d}:{second:02d}", inline=False)
|
||
|
||
print("Getting git describe data to display as version in info embed")
|
||
bot_version = await run('git describe --tags')
|
||
embed.add_field(name="Version:", value=f"{bot_version}", inline=False)
|
||
embed.add_field(name="discord.py Version:", value=f"{discord.__version__}", inline=False)
|
||
|
||
# Gitlab link to source
|
||
embed.add_field(name="Source Code:", value="[R0b3Bot on Gitlab](https://gitlab.rickelobe.com/Bots/r0b3BOT)", inline=False)
|
||
await ctx.send(embed=embed)
|
||
|
||
#
|
||
#
|
||
#
|
||
#@bot.command()
|
||
#async def commands(ctx):
|
||
# # Redirect to the 'help' command
|
||
# await help(ctx)
|
||
|
||
@bot.command()
|
||
async def help(ctx):
|
||
embed = discord.Embed(title='Core & Debug Commands:', description="Core commands for info and debug", color=0xeee657)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}info", value="Gives a little info about the bot", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}help", value="Gives this message", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}last_error", value="Will display the real error message the bot has last encountered for additional debugging info")
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}greetings", value="Gives a nice greeting message", inline=False)
|
||
await ctx.send(embed=embed)
|
||
|
||
embed = discord.Embed(title="GIF Commands", description="Commands to call up a GIF image", color=0xeee657)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}explain", value="Displays Dalek EXPLAIN gif", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}holyshit", value="Displays Marty McFly HOLY SHIT gif", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}cat", value="Gives a cute cat gif to lighten up the mood.", inline=False)
|
||
await ctx.send(embed=embed)
|
||
|
||
embed = discord.Embed(title="Sound Commands", description="To play a sound in a chat channel, use one of the following:", color=0xeee657)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}aww @member", value="Plays Rick & Morty clip 'Awww bitch'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}bitch @member", value="Plays clip of BigEric420 saying 'maybe you shouldn't be such a bitch.mp3'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}big @member", value="Plays 'Wiz Khalifa clip'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}boom @member", value="Plays 'Boom Bitch' sound clip", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}breee @member", value="Plays 'BREEEEE' clip from a WBTBWB song", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}cowbell @member", value="Plays 'SNL More Cowbell' sound clip", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}eia @member", value="Plays 'Everything is Awesome' song clip", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}heavy @member", value="Plays 'Poppy ooo heavy'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}iran @member", value="Plays Trump 'I Ran'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}leeroy @member", value="Plays Leeeerrroooyyy Jenkins clip", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}nice @member", value="Plays clip from WBTBWB - Nice!", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}oops @member", value="Plays 'Oops.mp3'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}promoted @member", value="Plays Battlefield Friends 'PROMOTED!' sound clip", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}trololo @member", value="Plays a clip of Trololo song", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}wow @member", value="Plays clip of Owen Wilson saying 'Woooow'", inline=False)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}yaw @member", value="Plays clip of Busta Ryhmes 'Yaw Yaw Yaw'", inline=False)
|
||
await ctx.send(embed=embed)
|
||
|
||
embed = discord.Embed(title="A note about sound clips:", description="Sound clips are only played in voice channels. If a user is not specified when calling the command, the sound will played in the channel of that issuing user is currently joined to. When a username is specified, the bot will play the sound in the channel that user is currently in.")
|
||
await ctx.send(embed=embed)
|
||
|
||
embed = discord.Embed(title="3D Printer Status", color=0xeee657)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}printstat", value="Uploads a snapshot of Rich's OctoPrint 3D printer and current stats", inline=False)
|
||
await ctx.send(embed=embed)
|
||
|
||
embed = discord.Embed(title="Bambu Lab Printer Status", color=0x00B4D8)
|
||
embed.add_field(name=f"{BOT_COMMAND_PREFIX}bambustat", value="Displays the current status of the Bambu Lab printer (temps, progress, time remaining)", inline=False)
|
||
await ctx.send(embed=embed)
|
||
|
||
# Function to run shell command and return stdout or sterr
|
||
async def run(cmd):
|
||
proc = await asyncio.create_subprocess_shell(
|
||
cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE)
|
||
|
||
stdout, stderr = await proc.communicate()
|
||
|
||
print(f'[{cmd!r} exited with {proc.returncode}]')
|
||
if stdout:
|
||
#print(f'[stdout]\n{stdout.decode()}')
|
||
return stdout.decode()
|
||
if stderr:
|
||
#print(f'[stderr]\n{stderr.decode()}')
|
||
return stderr.decode()
|
||
|
||
bot.run(DISCORD_AUTH_TOKEN)
|