chore: add test-stream-input tool (#2202)

docs/admin-manual/troubleshooting.md

@@ -60,3 +60,25 @@ sudo -u libretime libretime-analyzer --config /etc/libretime/config.yml --log-le
 ```
 
 The `/var/log/nginx/libretime.error.log` file contains logs from the web server.
+
+## Test the stream inputs
+
+To test or debug your input streams, you can use the [`tools/test-stream-input.py`](https://github.com/libretime/libretime/blob/main/tools/test-stream-input.py) script to send a test sound to your stream inputs.
+
+To test the `main` input stream, you can run the following command:
+
+```bash
+./tools/test-stream-input.py \
+    --host radio.example.org \
+    --port 8001 \
+    --mount main \
+    --user source \
+    --password hackme
+
+# Or using the --url option
+./tools/test-stream-input.py --url source:hackme@radio.example.org:8001/main
+```
+
+If you are hitting `HTTP error 401 Unauthorized` on the main input stream, make sure that you have configured a user and password in the **Settings** > **Streams settings** page.
+
+If you are connected but don't hear anything streaming, make sure that the main input stream is connected and enabled. In addition, to automatically turn on/off the stream on connect/disconnect you can configure the input stream auto switch in the **Settings** > **Streams settings** page.

tools/test-stream-input.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+# pylint: disable=invalid-name
+
+import subprocess
+from argparse import (
+    ArgumentDefaultsHelpFormatter,
+    ArgumentParser,
+    RawDescriptionHelpFormatter,
+)
+from contextlib import suppress
+
+
+class ArgumentParserFormatter(
+    RawDescriptionHelpFormatter,
+    ArgumentDefaultsHelpFormatter,
+):
+    pass
+
+
+def run():
+    parser = ArgumentParser(
+        description="Send a sine wave sound to an icecast mount or liquidsoap input harbor.",
+        formatter_class=lambda prog: ArgumentParserFormatter(
+            prog, max_help_position=60
+        ),
+    )
+    parser.add_argument(
+        "--url",
+        metavar="<url>",
+        help="""Stream <url> (<user>:<password>@<host>:<port>/<mount>) to test. If
+                defined any other option will be ignored.""",
+    )
+    parser.add_argument(
+        "--host",
+        metavar="<host>",
+        help="Stream <host> used to build the stream url.",
+        default="localhost",
+    )
+    parser.add_argument(
+        "--port",
+        metavar="<port>",
+        help="Stream <port> used to build the stream url.",
+        default=8001,
+    )
+    parser.add_argument(
+        "--mount",
+        metavar="<mount>",
+        help="Stream <mount> used to build the stream url.",
+        default="main",
+    )
+    parser.add_argument(
+        "--user",
+        metavar="<user>",
+        help="Stream <user> used to build the stream url.",
+        default="source",
+    )
+    parser.add_argument(
+        "--password",
+        metavar="<password>",
+        help="Stream <password> used to build the stream url.",
+        default="hackme",
+    )
+
+    args = parser.parse_args()
+
+    stream_url = args.url
+    if stream_url is None:
+        stream_url = f"icecast://{args.user}:{args.password}@{args.host}:{args.port}/{args.mount}"
+
+    cmd = ["ffmpeg", "-hide_banner"]
+    cmd.extend(["-re"])
+    cmd.extend(["-f", "lavfi", "-i", "sine=frequency=1000"])
+    cmd.extend(["-ar", "48000", "-ac", "2"])
+    cmd.extend(["-f", "ogg"])
+    cmd.extend(["-content_type", "application/ogg"])
+    cmd.extend([stream_url])
+
+    print(" ".join(cmd))
+    with suppress(subprocess.CalledProcessError):
+        subprocess.run(cmd, check=True, text=True)
+
+
+if __name__ == "__main__":
+    run()