Feat: Outputs are created while connection_made function

Author: DirectiveAthena

src/AthenaTwitchBot/functions/launch.py

@@ -22,7 +22,7 @@ def launch(
         *, # after this, keywords only
         bot:TwitchBot=None,
         protocol_factory:Callable=None,
-        outputs:list[AbstractOutput]=None,
+        outputs:list[type[AbstractOutput]]=None,
         console_enabled:bool=True,
         irc_connection:bool=True,
         irc_host:str='irc.chat.twitch.tv',
@@ -32,14 +32,14 @@ def launch(
 ):
     if irc_connection:
         if outputs is None:
-            outputs=[OutputTwitch()]
+            outputs=[OutputTwitch]
         # thanks for fikotta for pointing me to use isinstance here
-        if not any(isinstance(o, OutputTwitch) for o in outputs):
+        if not any(issubclass(o, OutputTwitch) for o in outputs):
             # always make sure OutputTwitch callbacks are the first to be made
-            outputs.insert(0,OutputTwitch())
-        if not any(isinstance(o, OutputConsole) for o in outputs) and console_enabled:
+            outputs.insert(0,OutputTwitch)
+        if not any(issubclass(o, OutputConsole) for o in outputs) and console_enabled:
             # placement of the Console does not matter
-            outputs.append(OutputConsole())
+            outputs.append(OutputConsole)
 
         # a bot always has to be defined
         if bot is None or not isinstance(bot, TwitchBot):

src/AthenaTwitchBot/models/outputs/abstract_output.py

@@ -4,7 +4,6 @@
 # General Packages
 from __future__ import annotations
 from abc import ABC, abstractmethod
-import asyncio
 
 # Custom Library
 
@@ -17,26 +16,29 @@
 # ----------------------------------------------------------------------------------------------------------------------
 class AbstractOutput(ABC):
 
+    def __init__(self, **kwargs):
+        pass
+
     @abstractmethod
-    async def connection_made(self, bot:TwitchBot, transport: asyncio.transports.Transport,**kwargs):
+    async def connection_made(self, bot:TwitchBot, **kwargs):
         """Output of data when the connection to the Twitch servers is established"""
 
     @abstractmethod
-    async def connection_ping(self, transport: asyncio.transports.Transport, ping_response:list[str],**kwargs):
+    async def connection_ping(self, ping_response:list[str],**kwargs):
         """Output response to a ping sent by the Twitch servers"""
 
     @abstractmethod
     async def undefined(self,text:str, **kwargs):
         """Output response to anything that wasn't caught correctly"""
 
     @abstractmethod
-    async def write(self,transport: asyncio.transports.Transport,context:TwitchContext, **kwargs):
+    async def write(self, context:TwitchContext, **kwargs):
         """Output which context is not a reply, but simply write out to the chat"""
 
     @abstractmethod
-    async def reply(self,transport: asyncio.transports.Transport,context:TwitchContext, **kwargs):
+    async def reply(self, context:TwitchContext, **kwargs):
         """Output which context is a reply to a user"""
 
     @abstractmethod
-    async def scheduled_task(self, transport: asyncio.transports.Transport, context:TwitchContext, **kwargs):
+    async def scheduled_task(self, context:TwitchContext, **kwargs):
         """Automated output after a task has been run"""

src/AthenaTwitchBot/models/outputs/output_twitch.py

@@ -20,38 +20,43 @@
 # - Code -
 # ----------------------------------------------------------------------------------------------------------------------
 class OutputTwitch(AbstractOutput):
-    async def connection_made(self, bot:TwitchBot, transport: asyncio.transports.Transport,**kwargs):
-        transport.write(format_message(f"{PASS}{bot.oauth_token}"))
-        transport.write(format_message(f"{NICK} {bot.nickname}"))
-        transport.write(format_message(f"{JOIN} {','.join(str(c) for c in bot.channels)}"))
-        transport.write(format_message(REQUEST_TAGS))
+    transport:asyncio.transports.Transport
 
-    async def connection_ping(self, transport: asyncio.transports.Transport, ping_response:list[str], **kwargs):
-        transport.write(format_message(f"{PONG} {' '.join(ping_response)}"))
+    def __init__(self,transport: asyncio.transports.Transport, **kwargs):
+        self.transport = transport
+
+    async def connection_made(self, bot:TwitchBot,**kwargs):
+        self.transport.write(format_message(f"{PASS}{bot.oauth_token}"))
+        self.transport.write(format_message(f"{NICK} {bot.nickname}"))
+        self.transport.write(format_message(f"{JOIN} {','.join(str(c) for c in bot.channels)}"))
+        self.transport.write(format_message(REQUEST_TAGS))
+
+    async def connection_ping(self, ping_response:list[str], **kwargs):
+        self.transport.write(format_message(f"{PONG} {' '.join(ping_response)}"))
 
     async def undefined(self,**kwargs):
         pass # don't answer to something that is undefined
 
-    async def write(self, transport: asyncio.transports.Transport, context:TwitchContext, **kwargs):
+    async def write(self, context:TwitchContext, **kwargs):
         if context.output_text is not None:
-            transport.write(
+            self.transport.write(
                 format_message(
                     f"{PRIVMSG} {context.channel} :{context.output_text}"
             ))
 
-    async def reply(self, transport: asyncio.transports.Transport, context:TwitchContext, **kwargs):
+    async def reply(self, context:TwitchContext, **kwargs):
         if context.output_text is not None:
             if context.message_tags.message_id != EMPTY_STR:
-                transport.write(
+                self.transport.write(
                     format_message(
                         f"@reply-parent-msg-id={context.message_tags.message_id} {PRIVMSG} {context.channel} :{context.output_text}"
                 ))
             else:
-                transport.write(
+                self.transport.write(
                     format_message(
                         f"{PRIVMSG} {context.channel} :{context.output_text}"
                     ))
 
 
-    async def scheduled_task(self,transport: asyncio.transports.Transport, context:TwitchContext, **kwargs):
+    async def scheduled_task(self, context:TwitchContext, **kwargs):
         pass # is handled by the "write" output

src/AthenaTwitchBot/models/twitch_bot_protocol.py

@@ -33,10 +33,10 @@
 @dataclass(kw_only=True, slots=True, eq=False, order=False)
 class TwitchBotProtocol(asyncio.Protocol):
     bot:TwitchBot
-    outputs:list[AbstractOutput]
+    outputs:list[type[AbstractOutput]]
 
     # non init slots
-    transport:asyncio.transports.Transport = field(init=False)
+    outputs_initialized:list[AbstractOutput] =field(init=False)
     message_constructor:Callable = field(init=False)
     loop:asyncio.AbstractEventLoop = field(init=False)
     PREFIX_FULL:str = field(init=False)
@@ -70,8 +70,8 @@ async def scheduled_task_coro(self, tsk:TwitchBotMethod, channel:TwitchChannel):
     def output_handler(self,callback:OUTPUT_CALLBACKS, **kwargs):
         asyncio.ensure_future(
             asyncio.gather(
-                *(callback(output=output, transport=self.transport, **kwargs)
-                for output in self.outputs)
+                *(callback(output=output, **kwargs)
+                for output in self.outputs_initialized)
             ),
             loop=self.loop
         )
@@ -80,7 +80,8 @@ def output_handler(self,callback:OUTPUT_CALLBACKS, **kwargs):
     # - Protocol necessary  -
     # ------------------------------------------------------------------------------------------------------------------
     def connection_made(self, transport: asyncio.transports.Transport) -> None:
-        self.transport = transport
+        self.outputs_initialized = [output(transport=transport) for output in self.outputs]
+
         # first write the password then the nickname else the connection will fail
         self.output_handler(
             callback=output_connection_made,