syslog

log95.py

@@ -0,0 +1,67 @@
+from enum import Enum
+import datetime, sys, os, io
+from typing import LiteralString, TextIO
+import syslog as sl
+
+try:
+    import colorama # type: ignore
+    colorama.init()
+except ModuleNotFoundError:
+    print("log95: colorama is not installed.")
+
+class log95Levels(Enum):
+    DEBUG = 0
+    VERBOSE = 1
+    INFO = 2
+    WARN = 3
+    ERROR = 4
+    CRITICAL_ERROR = 5
+
+def level_to_syslog(level: log95Levels):
+    match level:
+        case log95Levels.DEBUG: return sl.LOG_DEBUG
+        case log95Levels.VERBOSE: return sl.LOG_DEBUG
+        case log95Levels.INFO: return sl.LOG_INFO
+        case log95Levels.WARN: return sl.LOG_WARNING
+        case log95Levels.ERROR: return sl.LOG_ERR
+        case log95Levels.CRITICAL_ERROR: return sl.LOG_CRIT
+        case _: return sl.LOG_NOTICE
+
+class SyslogTextIO(io.TextIOWrapper):
+    def syslog(self, level: log95Levels, message: str): sl.syslog(level_to_syslog(level), message)
+
+class log95:
+    def __init__(self, tag : str="...", level: log95Levels = log95Levels.CRITICAL_ERROR, output: TextIO | io.TextIOWrapper | SyslogTextIO = sys.stdout) -> None:
+        self.tag = str(tag)
+        self.level = int(level.value)
+        self.output = output
+    def log(self, level: log95Levels, *args:str, seperator=" ") -> None:
+        if level.value > self.level: return
+        we_have_color = "colorama" in sys.modules
+        def level_to_str(_level: log95Levels, _color: bool) -> LiteralString | str:
+            if _color:
+                match _level:
+                    case log95Levels.VERBOSE: return f"{colorama.Fore.LIGHTWHITE_EX}VERBOSE{colorama.Fore.RESET}"
+                    case log95Levels.CRITICAL_ERROR: return f"{colorama.Fore.RED}CRITICAL{colorama.Fore.RESET}"
+                    case log95Levels.ERROR: return f"{colorama.Fore.LIGHTRED_EX}ERROR{colorama.Fore.RESET}"
+                    case log95Levels.WARN: return f"{colorama.Fore.YELLOW}WARN{colorama.Fore.RESET}"
+                    case log95Levels.INFO: return f"{colorama.Fore.BLUE}INFO{colorama.Fore.RESET}"
+                    case _: return _level.name
+            else:
+                match _level:
+                    case log95Levels.CRITICAL_ERROR: return "CRITICAL"
+                    case _: return _level.name
+        self.output.write(f"[{self.tag}] ({level_to_str(level, we_have_color)}) @ ({datetime.datetime.now().strftime('%d.%m.%Y %H:%M:%S.%f')}) - {seperator.join(args)}{os.linesep}")
+        if isinstance(self.output, SyslogTextIO): self.output.syslog(level, seperator.join(args))
+    def debug(self, *args:str, seperator=" ") -> None:
+        self.log(log95Levels.DEBUG, *args, seperator)
+    def verbose(self, *args:str, seperator=" ") -> None:
+        self.log(log95Levels.VERBOSE, *args, seperator)
+    def critical_error(self, *args:str, seperator=" ") -> None:
+        self.log(log95Levels.CRITICAL_ERROR, *args, seperator)
+    def error(self, *args:str, seperator=" ") -> None:
+        self.log(log95Levels.ERROR, *args, seperator)
+    def warning(self, *args:str, seperator=" ") -> None:
+        self.log(log95Levels.WARN, *args, seperator)
+    def info(self, *args:str, seperator=" ") -> None:
+        self.log(log95Levels.INFO, *args, seperator)

modules/ffmpeg_procman.py

@@ -50,11 +50,10 @@ class ProcessManager(ABC_ProcessManager):
                 except subprocess.TimeoutExpired: process.process.terminate()
             self.processes.clear()
     def test(self) -> bool:
-        return True
-        proc = Popen(["ffplay"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        proc = subprocess.Popen(["ffplay"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 
         start = time.monotonic()
-        while proc.poll() is None and (time.monotonic() - start) < 1: time.sleep(0.01)
+        while proc.poll() is None and (time.monotonic() - start) < 10: time.sleep(0.01)
 
         if proc.poll() is None: proc.kill()
         return proc.poll() not in (None, 127)

pyproject.toml

@@ -4,11 +4,11 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "radio-tools"
-version = "0.1"
-dependencies = ["log95", "unidecode", "libcache"]
+version = "0.2"
+dependencies = ["unidecode", "libcache"]
 
 [tool.setuptools]
-py-modules = ["radioPlaylist", "radioPlayer", "tinytag", "rds_codec"]
+py-modules = ["radioPlaylist", "radioPlayer", "tinytag", "rds_codec", "log95"]
 packages = ["modules"]
 
 [project.scripts]

radioPlayer.py

@@ -1,10 +1,10 @@
 #!/usr/bin/env python3
+from _io import _WrappedBuffer
 import os, importlib.util, importlib.machinery, types
 import sys, signal, time, traceback, io
 import concurrent.futures
 from modules import *
 from threading import Lock
-import syslog
 
 def prefetch(path):
     if os.name == "posix":
@@ -279,7 +279,11 @@ class RadioPlayer:
             traceback.print_exc(file=self.logger.output)
             raise
 
-class RotatingLog(io.TextIOWrapper):
+import syslog
+class RotatingLog(log95.SyslogTextIO):
+    def __init__(self, buffer: _WrappedBuffer, encoding: str | None = None, errors: str | None = None, newline: str | None = None, line_buffering: bool = False, write_through: bool = False) -> None:
+        super().__init__(buffer, encoding, errors, newline, line_buffering, write_through)
+        syslog.openlog("player")
     def write(self, *args, **kwargs) -> int:
         if self.tell() > 2_000_000:
             self.flush()
@@ -298,7 +302,9 @@ def main():
             signal.signal(signal.SIGINT, core.handle_sigint)
             core.loop()
         except SystemExit:
-            try: core.shutdown()
+            try: 
+                core.shutdown()
+                syslog.closelog()
             except BaseException: traceback.print_exc(file=f)
             raise
 

requirements.txt

@@ -1,3 +1,2 @@
 libcache
-log95
 unidecode