Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,8 @@ version = "0.0.0"
dependencies = [
"discord.py",
"python-dotenv",
"requests"
"requests",
"pyserial"
]
# dynamic = []

Expand Down
84 changes: 84 additions & 0 deletions src/pytexbot/hardware.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
import os
import asyncio
import discord
from discord import app_commands
from dotenv import load_dotenv
import serial
import time

# 1. Load configuration from .env file
load_dotenv()
TOKEN = os.getenv('DISCORD_TOKEN')
SERIAL_PORT = os.getenv('SERIAL_PORT', 'COM5')
ALLOWED_CHANNEL_ID = os.getenv('ALLOWED_CHANNEL_ID') # NEW: Re-load from .env

# 2. Setup Serial Connection
try:
arduino = serial.Serial(SERIAL_PORT, 9600, timeout=1)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Feels weird to have this at the module global scope, where it is executed once, on import (and before anything starts up). Is this the right place for this, or should it be tucked into a function or method somewhere?

@dijital20 dijital20 Mar 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One pattern you could use is a global variable and a function return or init.

__ARDUINO = None  # Global variable for state, init to none

def get_arduino() -> serial.Serial | None:
    global __ARDUINO

    if not __ARDUINO:  # If we haven't initialize, try to init.
        try:
            __ARDUINO = serial.Serial(SERIAL_PORT, 9600, timeout=1)
        except Exception as e:  # Fail
            print(f"❌ Hardware Error on {SERIAL_PORT}: {e}")
        else:  # Success
            time.sleep(2)
            print(f"✅ Connected to Arduino on {SERIAL_PORT}!")

    return __ARDUINO  # Return what we have. If we failed, this could be None.

The benefit here is that the arduino instance isn't created until the first time it's called for, and then it is cached in the global scope. For each successive request, the cached copy is returned.

Lines like 42, 51 and 60 would need to change from:

    # Current
    arduino.write(b'W')

    # New
    get_arduino().write(b'W')

time.sleep(2)
print(f"✅ Connected to Arduino on {SERIAL_PORT}!")
except Exception as e:
print(f"❌ Hardware Error on {SERIAL_PORT}: {e}")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Assuming you don't do the above... do note that since you are handling the error, your script continues executing from here... but when you go call arduino in lines like 42, and you hit this error, you will crash at that point.

Consider either setting arduino to something like None and augmenting things to handle it, or adding raise after this line to re-raise the exception and crash here.


# 3. Discord Bot Logic
class HardwareBot(discord.Client):
def __init__(self):
super().__init__(intents=discord.Intents.default())
self.tree = app_commands.CommandTree(self)

async def setup_hook(self):
await self.tree.sync()

client = HardwareBot()

# --- COMMANDS WITH COOLDOWNS ---

@client.tree.command(name="wave", description="Send a long wave blink")
@app_commands.checks.cooldown(1, 10.0, key=lambda i: i.user.id) # 1 use every 10s
async def wave(interaction: discord.Interaction):
if ALLOWED_CHANNEL_ID and str(interaction.channel.id) != str(ALLOWED_CHANNEL_ID):
await interaction.response.send_message("❌ This command is not allowed in this channel.", ephemeral=True)
return
arduino.write(b'W')
await interaction.response.send_message(f"👋 {interaction.user.display_name} sent a wave!")

@client.tree.command(name="love", description="Send fast blinks of love")
@app_commands.checks.cooldown(1, 10.0, key=lambda i: i.user.id) # 1 use every 10s
async def love(interaction: discord.Interaction):
if ALLOWED_CHANNEL_ID and str(interaction.channel.id) != str(ALLOWED_CHANNEL_ID):
await interaction.response.send_message("❌ This command is not allowed in this channel.", ephemeral=True)
return
arduino.write(b'L')
await interaction.response.send_message(f"❤️ {interaction.user.display_name} is sending love!")

@client.tree.command(name="question", description="Send a pulse for a question")
@app_commands.checks.cooldown(1, 10.0, key=lambda i: i.user.id)
async def question(interaction: discord.Interaction):
if ALLOWED_CHANNEL_ID and str(interaction.channel.id) != str(ALLOWED_CHANNEL_ID):
await interaction.response.send_message("❌ This command is not allowed in this channel.", ephemeral=True)
return
arduino.write(b'Q')
await interaction.response.send_message(f"❓ {interaction.user.display_name} has a question!")

# --- GLOBAL ERROR HANDLER ---
# This one function handles the "Rate Limit" message for ALL commands above
@client.tree.error
async def on_app_command_error(interaction: discord.Interaction, error: app_commands.AppCommandError):
if isinstance(error, app_commands.CommandOnCooldown):
await interaction.response.send_message(
f"⏳ Slow down, {interaction.user.display_name}! Try again in {error.retry_after:.1f}s.",
ephemeral=True # Only the spammer sees this
)
else:
# Log other errors so you can see them in your terminal
print(f"Command Error: {error}")

def main():
# 4. Run the Bot using the hidden Token
if TOKEN:
client.run(TOKEN)
else:
print("❌ Error: No token found. Did you create the .env file?")

if __name__ == "__main__":
main()