Added Live Streaming via ALSA Loopback Device

assets/installation.json

@@ -39,6 +39,7 @@
   "always": {
     "files": [
       "autorun/autorun.py",
+      "shared/alsa.py",
       "shared/cat.py",
       "shared/cat.jpg",
       "shared/handlers.py",
@@ -57,7 +58,8 @@
       "dlogger==1.0.4",
       "aiofiles==0.8.0",
       "aiohttp",
-      "morse-talk==0.2"
+      "morse-talk==0.2",
+      "pyalsaaudio==0.11.0"
     ],
     "binaries": [
       "bin/bw-autorun",

assets/latest.ver.txt

@@ -1 +1 @@
-2.0.0
+2.0.1

bin/bw-nandl

@@ -50,7 +50,7 @@ EDITOR="${EDITOR:-nano}"
 
 # supported handlers and commands
 VALID_PREFIXES=("l_onready" "l_onstart" "l_onstop" "s_onready" "s_onstart" "s_onstop" "s_onconnect" "s_ondisconnect" "s_onwsjoin" "s_onwsleave")
-VALID_COMMANDS=("start" "stop" "list" "upload" "dl" "handlers" "<" "help" "exit" "kick" "restart" "rm" "sync" "lf" "sstv")
+VALID_COMMANDS=("start" "stop" "list" "upload" "dl" "handlers" "<" "help" "exit" "kick" "rm" "sync" "lf" "sstv" "morse" "live")
 
 list_handlers() {
     echo ""

client/client.py

@@ -26,6 +26,7 @@
 
 # using this to access to the shared dir files
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+from shared.alsa import Alsa
 from shared.bw_custom import BWCustom
 from shared.cat import check
 from shared.http import BWHTTPFileClient
@@ -64,6 +65,9 @@ def __init__(self, server_host: str, ws_port: int, http_port: int, http_host: st
         self.broadcasting = False
         self.current_file = None
         self.broadcast_lock = asyncio.Lock() # using asyncio instead of thereading now
+        self.alsa = Alsa()
+        self.stream_task = None
+        self.stream_active = False
 
         # states
         self.running = False
@@ -186,6 +190,10 @@ async def _handle_server_msg(self, message: str):
                 await self._handle_start_broadcast(kwargs)
                 return
 
+            if command == Commands.STREAM_TOKEN:
+                await self._handle_stream_token(kwargs)
+                return
+
             if command == Commands.STOP:
                 await self._handle_stop_broadcast()
                 return
@@ -219,14 +227,9 @@ async def _handle_server_msg(self, message: str):
                 await self.stop()
                 return
 
-            if command == Commands.RESTART:
-                Log.info("Restart requested")
-                await self._handle_stop_broadcast()
-                response = ProtocolParser.build_response(Commands.OK, "Restart acknowledged")
-                await self.ws_client.send(response)
-                return
-
             Log.warning(f"Unknown command: {command}")
+            response = ProtocolParser.build_response(Commands.ERROR, f"Unknown command: {command}. Perhaps a protocol mismatch ?")
+            await self.ws_client.send(response)
 
         except Exception as e:
             Log.error(f"Error handling message: {e}")
@@ -386,6 +389,107 @@ async def _handle_start_broadcast(self, kwargs: dict):
 
         await self.ws_client.send(response)
 
+    async def _handle_stream_token(self, kwargs: dict):
+        token = kwargs.get('token')
+        rate = int(kwargs.get('rate', 48000))
+        channels = int(kwargs.get('channels', 2))
+
+        # Broadcast params
+        frequency = float(kwargs.get('frequency', 90.0))
+        ps = kwargs.get('ps', 'BotWave')
+        rt = kwargs.get('rt', 'Streaming')
+        pi = kwargs.get('pi', 'FFFF')
+
+        if not token:
+            error = ProtocolParser.build_response(Commands.ERROR, "Missing token")
+            await self.ws_client.send(error)
+            return
+
+        Log.broadcast(f"Received stream token (rate={rate}, channels={channels})")
+
+        started = await self._start_stream_broadcast(token, rate, channels, frequency, ps, rt, pi)
+
+        if isinstance(started, Exception):
+            response = ProtocolParser.build_response(Commands.ERROR, message=str(started))
+        else:
+            response = ProtocolParser.build_response(Commands.OK, "Stream broadcast started")
+
+        await self.ws_client.send(response)
+
+    async def _start_stream_broadcast(self, token, rate, channels, frequency, ps, rt, pi):
+        async def finished():
+            Log.info("Stream finished, stopping broadcast...")
+            await self._stop_broadcast()
+
+        async with self.broadcast_lock:
+            if self.broadcasting:
+                await self._stop_broadcast()
+
+            try:
+                self.piwave = PiWave(
+                    frequency=frequency,
+                    ps=ps,
+                    rt=rt,
+                    pi=pi,
+                    loop=False,
+                    backend="bw_custom",
+                    debug=False
+                )
+
+                stream_task = self.http_client.stream_pcm_generator(
+                    server_host=self.http_host,
+                    server_port=self.http_port,
+                    token=token,
+                    rate=rate,
+                    channels=channels,
+                    chunk_size=1024
+                )
+
+                self.stream_active = True
+
+                def sync_generator_wrapper():
+                    loop = asyncio.new_event_loop()
+                    try:
+                        async_gen = self.stream_task.__aiter__()
+                        while self.stream_active:
+                            try:
+                                chunk = loop.run_until_complete(
+                                    asyncio.wait_for(async_gen.__anext__(), timeout=5.0)
+                                )
+                                yield chunk
+                            except asyncio.TimeoutError:
+                                if not self.stream_active:
+                                    break
+                                continue
+                            except StopAsyncIteration:
+                                break
+                    except Exception as e:
+                        Log.error(f"Stream generator error: {e}")
+                    finally:
+                        loop.close()
+
+                self.broadcasting = True
+                self.current_file = f"stream:{token[:8]}"
+
+                self.stream_task = stream_task
+
+                self.piwave.play(
+                    sync_generator_wrapper(),
+                    sample_rate=rate,
+                    channels=channels,
+                    chunk_size=1024
+                )
+
+                self.piwave_monitor.start(self.piwave, finished, asyncio.get_event_loop())
+
+                Log.broadcast(f"Broadcasting stream on {frequency} MHz (rate={rate}, channels={channels})")
+                return True
+
+            except Exception as e:
+                Log.error(f"Stream broadcast error: {e}")
+                self.broadcasting = False
+                return e
+
     async def _delayed_broadcast(self, file_path, filename, frequency, ps, rt, pi, loop, delay):
         await asyncio.sleep(delay)
         started = await self._start_broadcast(file_path, filename, frequency, ps, rt, pi, loop)
@@ -438,6 +542,20 @@ async def _stop_broadcast(self):
         async with self.broadcast_lock:
             self.piwave_monitor.stop()
 
+            if self.stream_active:
+                self.stream_active = False
+                await asyncio.wait(0.2)
+
+            if self.stream_task:
+                try:
+                    await asyncio.sleep(0.1)
+                    await self.stream_task.aclose()
+                    Log.broadcast("Stream closed")
+                except Exception as e:
+                    Log.error(f"Error closing stream: {e}")
+                finally:
+                    self.stream_task = None
+
             if self.piwave:
                 try:
                     self.piwave.cleanup() # stops AND cleanups

local/local.md

@@ -61,6 +61,9 @@ Once the client is running, you can use the following commands:
 
 - `stop`: Stop the current broadcast.  
     - Usage: `botwave> stop`
+
+- `live`: Start a live broadcast.  
+    - Usage: `botwave> live [frequency] [loop] [ps] [rt] [pi]`
 
 - `sstv`: Start broadcasting an image converted to SSTV. For modes see [dnet/pySSTV](https://github.com/dnet/pySSTV/).  
     - Usage: `botwave> sstv <image path> [mode] [output wav name] [freq] [loop] [ps] [rt] [pi]`
@@ -80,7 +83,7 @@ Once the client is running, you can use the following commands:
 - `handlers`: List all handlers or commands in a specific handler file.  
     - Usage: `botwave> handlers [filename]`
 
-- `>`: Run a shell command on the main OS.  
+- `<`: Run a shell command on the main OS.  
     - Usage: `botwave> < <command>`
 
 - `help`: Display the help message.  

local/local.py

@@ -22,6 +22,7 @@
 
 # using this to access to the shared dir files
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+from shared.alsa import Alsa
 from shared.bw_custom import BWCustom
 from shared.cat import check
 from shared.handlers import HandlerExecutor
@@ -57,6 +58,7 @@ def __init__(self, upload_dir: str = "/opt/BotWave/uploads", handlers_dir: str =
         self.handlers_dir = handlers_dir
         self.handlers_executor = HandlerExecutor(handlers_dir, self._execute_command)
         self.piwave_monitor = PWM()
+        self.alsa = Alsa()
         self.ws_port = ws_port
         self.ws_server = None
         self.ws_clients = set()
@@ -101,6 +103,7 @@ def _execute_command(self, command: str):
                 return True
 
             cmd = cmd_parts[0].lower()
+
             if cmd == 'start':
                 if len(cmd_parts) < 2:
                     Log.error("Usage: start <file> [frequency] [loop] [ps] [rt] [pi]")
@@ -115,6 +118,17 @@ def _execute_command(self, command: str):
                 self.start_broadcast(file_path, frequency, ps, rt, pi, loop)
                 self.onstart_handlers()
                 return True
+
+            if cmd == 'live':
+                frequency = float(cmd_parts[1]) if len(cmd_parts) > 1 else 90.0
+                ps = cmd_parts[2] if len(cmd_parts) > 2 else "RADIOOOO"
+                rt = " ".join(cmd_parts[3:-1]) if len(cmd_parts) > 3 else "Broadcasting"
+                pi = cmd_parts[-1] if len(cmd_parts) > 4 else "FFFF"
+
+                self.start_live(frequency, ps, rt, pi)
+                self.onstart_handlers()
+                return True
+
 
             elif cmd == 'stop':
                 self.stop_broadcast()
@@ -342,7 +356,7 @@ def _download_reporthook(block_num, block_size, total_size):
         except Exception as e:
             Log.error(f"Download error: {str(e)}")
             return False
-
+        
     def start_broadcast(self, file_path: str, frequency: float = 90.0, ps: str = "RADIOOOO", rt: str = "Broadcasting", pi: str = "FFFF", loop: bool = False):
         def finished():
             Log.info("Playback finished, stopping broadcast...")
@@ -384,6 +398,53 @@ def finished():
             self.piwave = None
             return False
 
+    def start_live(self, frequency: float = 90.0, ps: str = "RADIOOOO", rt: str = "Broadcasting", pi: str = "FFFF"):
+        def finished():
+            Log.info("Playback finished, stopping broadcast...")
+            self.stop_broadcast()
+            self.onstop_handlers()
+
+        if not self.alsa.is_supported():
+            Log.alsa("Live broadcast is not supported on this installation.")
+            Log.alsa("Did you setup the ALSA loopback card correctly ?")
+            return False
+
+        if self.broadcasting:
+            self.stop_broadcast()
+
+        try:
+            self.piwave = PiWave(
+                frequency=frequency,
+                ps=ps,
+                rt=rt,
+                pi=pi,
+                backend="bw_custom",
+                debug=False
+            )
+
+            self.alsa.start()
+
+            self.current_file = "live_playback"
+            self.broadcasting = True
+            self.piwave.play(self.alsa.audio_generator(), sample_rate=self.alsa.rate, channels=self.alsa.rate, chunk_size=self.alsa.period_size)
+
+            self.piwave_monitor.start(self.piwave, finished)
+
+
+            Log.success(f"Live broadcast started on frequency {frequency} MHz")
+            Log.alsa("To play live, please set your output sound card (ALSA) to 'BotWave'.")
+            Log.alsa(f"We're expecting {self.alsa.rate}kHz on {self.alsa.channels} channels.")
+            return True
+
+        except Exception as e:
+            Log.error(f"Error starting broadcast: {e}")
+            self.alsa.stop()
+            self.broadcasting = False
+            self.current_file = None
+            self.piwave = None
+            return False
+
+
     def stop_broadcast(self):
         if not self.broadcasting:
             Log.warning("No broadcast is currently running")
@@ -398,6 +459,8 @@ def stop_broadcast(self):
             finally:
                 self.piwave = None
 
+        self.alsa.stop()
+
         self.broadcasting = False
         self.current_file = None
         return True
@@ -465,6 +528,11 @@ def display_help(self):
         Log.print("  Stop the current broadcast", 'white')
         Log.print("")
 
+        Log.print("live [freq] [ps] [rt] [pi]", 'bright_green')
+        Log.print("  Start a live audio broadcast", 'white')
+        Log.print("  Example: live", 'cyan')
+        Log.print("")
+
         Log.print("sstv <image_path> [mode] [output_wav] [frequency] [loop] [ps] [rt] [pi]", 'bright_green')
         Log.print("  Convert an image into a SSTV WAV file, and then broadcast it", 'white')
         Log.print("  Example: sstv /path/to/mycat.png Robot36 cat.wav 90 false PsPs Cutie FFFF", 'cyan')

misc_doc/cloud-install.sh

@@ -74,7 +74,7 @@ install_bore() {
 install_botwave() {
     log INFO "Installing BotWave server..."
 
-    curl -sSL https://botwave.dpip.lol/install | sudo bash -s server
+    curl -sSL https://botwave.dpip.lol/install | sudo bash -s -- server --no-alsa
 
     log INFO "BotWave server installed"
 }