feat(playout): use liquidsoap version functions

- remove "packaging" package

playout/libretime_playout/liquidsoap/main.py

@@ -1,7 +1,6 @@
 """ Runs Airtime liquidsoap
 """
 import os
-import subprocess
 from pathlib import Path
 from typing import Optional
 
@@ -12,6 +11,7 @@ from libretime_shared.logging import level_from_name, setup_logger
 from loguru import logger
 
 from .entrypoint import generate_entrypoint
+from .version import get_liquidsoap_version
 
 
 @click.command(context_settings={"auto_envvar_prefix": DEFAULT_ENV_PREFIX})
@@ -25,14 +25,10 @@ def cli(log_level: int, log_filepath: Optional[Path]):
 
     generate_entrypoint(log_filepath)
 
-    # check liquidsoap version so we can run a scripts matching the liquidsoap minor version
+    version = get_liquidsoap_version()
-    liquidsoap_version = subprocess.check_output(
+
-        "liquidsoap 'print(liquidsoap.version) shutdown()'",
-        shell=True,
-        universal_newlines=True,
-    )[0:3]
     script_path = os.path.join(
-        os.path.dirname(__file__), liquidsoap_version, "ls_script.liq"
+        os.path.dirname(__file__), f"{version[0]}.{version[1]}", "ls_script.liq"
     )
     exec_args = [
         "/usr/bin/liquidsoap",
@@ -43,5 +39,5 @@ def cli(log_level: int, log_filepath: Optional[Path]):
     if log_level.is_debug():
         exec_args.append("--debug")
 
-    logger.debug(f"Liquidsoap {liquidsoap_version} using script: {script_path}")
+    logger.debug(f"Liquidsoap {version} using script: {script_path}")
     os.execl(*exec_args)

playout/libretime_playout/liquidsoap/version.py

@@ -0,0 +1,26 @@
+import re
+from subprocess import PIPE, run
+from typing import Tuple
+
+LIQUIDSOAP_VERSION_RE = re.compile(r"(?:Liquidsoap )?(\d+).(\d+).(\d+)")
+LIQUIDSOAP_MIN_VERSION = (1, 1, 1)
+
+
+def parse_liquidsoap_version(version: str) -> Tuple[int, int, int]:
+    match = LIQUIDSOAP_VERSION_RE.search(version)
+
+    if match is None:
+        return (0, 0, 0)
+    return (int(match.group(1)), int(match.group(2)), int(match.group(3)))
+
+
+def get_liquidsoap_version() -> Tuple[int, int, int]:
+    cmd = run(
+        ("liquidsoap", "--check", "print(liquidsoap.version) shutdown()"),
+        check=True,
+        stdout=PIPE,
+        stderr=PIPE,
+        universal_newlines=True,
+    )
+
+    return parse_liquidsoap_version(cmd.stdout)

playout/libretime_playout/main.py

@@ -2,7 +2,6 @@
 Python part of radio playout (pypo)
 """
 
-import re
 import signal
 import sys
 import telnetlib
@@ -11,7 +10,7 @@ from datetime import datetime
 from pathlib import Path
 from queue import Queue
 from threading import Lock
-from typing import Optional
+from typing import Optional, Tuple
 
 import click
 from libretime_api_client.version1 import AirtimeApiClient as ApiClient
@@ -20,8 +19,8 @@ from libretime_shared.config import DEFAULT_ENV_PREFIX
 from libretime_shared.logging import level_from_name, setup_logger
 from loguru import logger
 
-from . import pure
 from .config import CACHE_DIR, RECORD_DIR, Config
+from .liquidsoap.version import LIQUIDSOAP_MIN_VERSION, parse_liquidsoap_version
 from .listenerstat import ListenerStat
 from .pypofetch import PypoFetch
 from .pypofile import PypoFile
@@ -31,8 +30,6 @@ from .pypopush import PypoPush
 from .recorder import Recorder
 from .timeout import ls_timeout
 
-LIQUIDSOAP_MIN_VERSION = "1.1.1"
-
 
 class Global:
     def __init__(self, api_client):
@@ -51,7 +48,7 @@ def keyboardInterruptHandler(signum, frame):
 
 
 @ls_timeout
-def liquidsoap_get_info(telnet_lock, host, port):
+def liquidsoap_get_info(telnet_lock, host, port) -> Optional[Tuple[int, int, int]]:
     logger.debug("Checking to see if Liquidsoap is running")
     try:
         telnet_lock.acquire()
@@ -66,50 +63,38 @@ def liquidsoap_get_info(telnet_lock, host, port):
     finally:
         telnet_lock.release()
 
-    return get_liquidsoap_version(response)
+    return parse_liquidsoap_version(response)
-
-
-def get_liquidsoap_version(version_string):
-    m = re.match(r"Liquidsoap (\d+.\d+.\d+)", version_string)
-
-    if m:
-        return m.group(1)
-    else:
-        return None
 
 
 def liquidsoap_startup_test(telnet_lock, liquidsoap_host, liquidsoap_port):
 
-    liquidsoap_version_string = liquidsoap_get_info(
+    liquidsoap_version = liquidsoap_get_info(
         telnet_lock,
         liquidsoap_host,
         liquidsoap_port,
     )
-    while not liquidsoap_version_string:
+    while not liquidsoap_version:
-        logger.warning(
+        logger.warning("Liquidsoap doesn't appear to be running! Trying again later...")
-            "Liquidsoap doesn't appear to be running!, " + "Sleeping and trying again"
-        )
         time.sleep(1)
-        liquidsoap_version_string = liquidsoap_get_info(
+        liquidsoap_version = liquidsoap_get_info(
             telnet_lock,
             liquidsoap_host,
             liquidsoap_port,
         )
 
-    while pure.version_cmp(liquidsoap_version_string, LIQUIDSOAP_MIN_VERSION) < 0:
+    while not LIQUIDSOAP_MIN_VERSION <= liquidsoap_version:
         logger.warning(
-            "Liquidsoap is running but in incorrect version! "
+            f"Found invalid Liquidsoap version! "
-            + "Make sure you have at least Liquidsoap %s installed"
+            f"Liquidsoap<={LIQUIDSOAP_MIN_VERSION} is required!"
-            % LIQUIDSOAP_MIN_VERSION
         )
-        time.sleep(1)
+        time.sleep(5)
-        liquidsoap_version_string = liquidsoap_get_info(
+        liquidsoap_version = liquidsoap_get_info(
             telnet_lock,
             liquidsoap_host,
             liquidsoap_port,
         )
 
-    logger.info("Liquidsoap version string found %s" % liquidsoap_version_string)
+    logger.info(f"Liquidsoap version {liquidsoap_version}")
 
 
 @click.command(context_settings={"auto_envvar_prefix": DEFAULT_ENV_PREFIX})

playout/libretime_playout/pure.py

@@ -1,16 +1,3 @@
-from packaging.version import parse
-
-
-def version_cmp(version1, version2):
-    version1 = parse(version1)
-    version2 = parse(version2)
-    if version1 > version2:
-        return 1
-    if version1 == version2:
-        return 0
-    return -1
-
-
 def date_interval_to_seconds(interval):
     """
     Convert timedelta object into int representing the number of seconds. If

playout/libretime_playout/pypofetch.py

@@ -16,7 +16,6 @@ from libretime_api_client import version1 as v1_api_client
 from libretime_api_client import version2 as api_client
 from loguru import logger
 
-from . import pure
 from .config import CACHE_DIR, POLL_INTERVAL, Config
 from .schedule import get_schedule
 from .timeout import ls_timeout

playout/requirements.txt

@@ -4,7 +4,6 @@ amqplib
 defusedxml
 kombu
 mutagen
-packaging
 pytz
 requests
 typing-extensions

playout/setup.py

@@ -34,13 +34,13 @@ setup(
         "defusedxml",
         "kombu",
         "mutagen",
-        "packaging",
         "pytz",
         "requests",
         "typing-extensions",
     ],
     extras_require={
         "dev": [
+            "distro",
             f"libretime-api-client @ file://localhost{here.parent / 'api-client'}",
             f"libretime-shared @ file://localhost{here.parent / 'shared'}",
         ],

playout/tests/liquidsoap/version_test.py

@@ -0,0 +1,37 @@
+from os import getenv
+
+import distro
+import pytest
+
+from libretime_playout.liquidsoap.version import (
+    get_liquidsoap_version,
+    parse_liquidsoap_version,
+)
+
+
+@pytest.mark.parametrize(
+    "version, expected",
+    [
+        ("invalid data", (0, 0, 0)),
+        ("1.1.0", (1, 1, 0)),
+        ("1.4.4", (1, 4, 4)),
+        ("2.0.0", (2, 0, 0)),
+        ("Liquidsoap 1.1.0", (1, 1, 0)),
+        ("Liquidsoap 1.4.4", (1, 4, 4)),
+        ("Liquidsoap 2.0.0", (2, 0, 0)),
+    ],
+)
+def test_parse_liquidsoap_version(version, expected):
+    assert parse_liquidsoap_version(version) == expected
+
+
+@pytest.mark.skipif(getenv("CI") != "true", reason="requires liquidsoap")
+def test_get_liquidsoap_version():
+    LIQUIDSOAP_VERSION_MAP = {
+        "bionic": (1, 1, 1),
+        "buster": (1, 3, 3),
+        "focal": (1, 4, 1),
+        "bullseye": (1, 4, 3),
+        "jammy": (2, 0, 2),
+    }
+    assert get_liquidsoap_version() == LIQUIDSOAP_VERSION_MAP[distro.codename()]