Feat: Frequent oputput method

Author: DirectiveAthena

src/AthenaTwitchBot/__init__.py

@@ -2,6 +2,7 @@
 # - Package Imports -
 # ----------------------------------------------------------------------------------------------------------------------
 from AthenaTwitchBot.decorators.command import commandmethod
+from AthenaTwitchBot.decorators.frequentoutput import frequentoutputmethod
 
 from AthenaTwitchBot.models.twitch_bot import TwitchBot
 from AthenaTwitchBot.models.twitch_bot_protocol import TwitchBotProtocol

src/AthenaTwitchBot/decorators/command.py

@@ -17,6 +17,7 @@ def wrapper(*args, **kwargs):
             return fnc(*args, **kwargs)
 
         # store attributes for later use by the bot
+        wrapper.is_command = True
         wrapper.command_name = name
 
         return wrapper

src/AthenaTwitchBot/decorators/frequentoutput.py

@@ -0,0 +1,31 @@
+# ----------------------------------------------------------------------------------------------------------------------
+# - Package Imports -
+# ----------------------------------------------------------------------------------------------------------------------
+# General Packages
+from __future__ import annotations
+
+# Custom Library
+
+# Custom Packages
+
+# ----------------------------------------------------------------------------------------------------------------------
+# - Code -
+# ----------------------------------------------------------------------------------------------------------------------
+def frequentoutputmethod(delay:int=3600): # defult is every hour
+    """
+    Create a method that runs every couple of seconds.
+    The delay parameter is defined in seconds
+    :param delay:
+    :return:
+    """
+
+    def decorator(fnc):
+        def wrapper(*args, **kwargs):
+            return fnc(*args, **kwargs)
+
+        # store attributes for later use by the bot
+        #   to be used by the protocol to assign it top an async call loop
+        wrapper.is_frequent_ouput = True
+        wrapper.delay = delay
+        return wrapper
+    return decorator

src/AthenaTwitchBot/models/twitch_bot.py

@@ -33,6 +33,7 @@ class TwitchBot:
 
     # noinspection PyDataclass
     commands:dict[str: Callable]=field(init=False)
+    frequent_outputs:list[tuple[Callable, int]]=field(init=False)
 
     # non init slots
 
@@ -42,13 +43,19 @@ class TwitchBot:
     def __new__(cls, *args, **kwargs):
         # Loop over own functions to see if any is decorated with the command setup
         cls.commands = {}
+        cls.frequent_outputs = []
 
         # create the actual instance
         #   Which is to be used in the commands tuple
         obj = super(TwitchBot, cls).__new__(cls,*args,**kwargs)
+
+        # loop over the bot's metods and parse the different methods
         for k,v in cls.__dict__.items():
-            if inspect.isfunction(v) and getattr(v, "command_name", False):
-                cls.commands[v.command_name] = (v, obj)
+            if inspect.isfunction(v):
+                if "is_command" in (attributes := [attribute for attribute in dir(v) if not attribute.startswith("__")]):
+                    cls.commands[v.command_name] = v
+                elif "is_frequent_ouput" in attributes:
+                    cls.frequent_outputs.append((v,v.delay))
 
         return obj
 

src/AthenaTwitchBot/models/twitch_bot_protocol.py

@@ -36,6 +36,9 @@ def __post_init__(self):
         else:
             raise NotImplementedError("This needs to be created")
 
+    # ----------------------------------------------------------------------------------------------------------------------
+    # - Protocol neseccary  -
+    # ----------------------------------------------------------------------------------------------------------------------
     def connection_made(self, transport: asyncio.transports.Transport) -> None:
         self.transport = transport
         # first write the password then the nickname else the connection will fail
@@ -44,26 +47,40 @@ def connection_made(self, transport: asyncio.transports.Transport) -> None:
         self.transport.write(messages.join(channel=self.bot.channel))
         self.transport.write(messages.request_tags)
 
+        # add frequent_ouput methods to the coroutine loop
+        loop = asyncio.get_running_loop()
+        for callback, delay in self.bot.frequent_outputs:
+            coro = loop.create_task(self.frequent_output_call(callback,delay))
+            asyncio.ensure_future(coro, loop=loop)
+
+    async def frequent_output_call(self, callback,delay:int):
+        context = TwitchMessageContext(
+            message=TwitchMessage(channel=f"#{self.bot.channel}"),
+            transport=self.transport
+        )
+        while True:
+            await asyncio.sleep(delay)
+            callback(
+                self=self.bot,
+                context=context)
+
     def data_received(self, data: bytearray) -> None:
         for message in data.split(b"\r\n"):
             match (twitch_message := self.message_constructor(message, bot_name=self.bot.nickname)):
                 # Keepalive messages : https://dev.twitch.tv/docs/irc#keepalive-messages
                 case TwitchMessagePing():
                     print(ForeNest.ForestGreen("PINGED BY TWITCH"))
-                    self.transport.write(pong_message := messages.pong(message=twitch_message.text))
-                    print(pong_message)
+                    self.transport.write(messages.pong(message=twitch_message.text))
 
                 # catch a message which starts with a command:
-                case TwitchMessage(text=user_message) if user_message.startswith(f"{self.bot.prefix}"):
-                    user_message:str
+                case TwitchMessage(text=str(user_message)) if user_message.startswith(f"{self.bot.prefix}"):
                     print(ForeNest.ForestGreen("COMMAND CAUGHT"))
                     try:
                         user_cmd = user_message.replace(f"{self.bot.prefix}", "")
                         # tuple unpacking because we have a callback
                         #   and the object instance where the command is placed in
-                        callback, orign_obj = self.bot.commands[user_cmd]
-                        callback(
-                            self=orign_obj,
+                        self.bot.commands[user_cmd](
+                            self=self.bot,
                             # Assign a context so the user doesn't need to write the transport messages themselves
                             #   A user opnly has to write the text
                             context=TwitchMessageContext(