Commit mess instead of stashing

AudioControl.exe.spec

@@ -0,0 +1,44 @@
+# -*- mode: python ; coding: utf-8 -*-
+
+
+block_cipher = None
+
+
+a = Analysis(
+    ['main.py'],
+    pathex=[],
+    binaries=[],
+    datas=[],
+    hiddenimports=[],
+    hookspath=[],
+    hooksconfig={},
+    runtime_hooks=[],
+    excludes=[],
+    win_no_prefer_redirects=False,
+    win_private_assemblies=False,
+    cipher=block_cipher,
+    noarchive=False,
+)
+pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
+
+exe = EXE(
+    pyz,
+    a.scripts,
+    a.binaries,
+    a.zipfiles,
+    a.datas,
+    [],
+    name='AudioControl.exe',
+    debug=False,
+    bootloader_ignore_signals=False,
+    strip=False,
+    upx=True,
+    upx_exclude=[],
+    runtime_tmpdir=None,
+    console=True,
+    disable_windowed_traceback=False,
+    argv_emulation=False,
+    target_arch=None,
+    codesign_identity=None,
+    entitlements_file=None,
+)

AudioControl.md

@@ -0,0 +1,44 @@
+# AudioControl
+A tool to remotely control audio volume of different things. It includes audio sessions in windows.
+Easy extendable via plugin system.
+
+## Architecture
+`SessionProvider` provides unified interface to audio sessions.
+
+`Controller` dispatches events from `SessionProvider`s to `TransportProvider`s and backward.
+
+`TransportProvider` allows controller to communicate with clients using common interface
+
+### Providers and Providers manager
+A `SessionProvider` represent a source and way to control audio sessions.
+There can be, for example, a `SessionProvider` for control over windows default mixer,  
+`SessionProvider` for internal audio control of a decent software (i.e. an audio player).
+
+`ProvidersManager` is piece that control both session and transport providers and allow `Controller` control it  
+---------
+Plugins must subclass and implement abstract methods of class `PluginABC`.
+In order to be loadable, file with class must have same name as plugin class inside this file
+(i.e. if you have plugin class called `MyPlugin` then you need to call file with plugin `MyPlugin`).
+Let's take a look to methods you should define:
+- `def __init__(self, plugin_system: 'PluginSystem')` - Constructor of a plugin will be called with one 
+argument - instance of a `PluginSystem`.
+Plugin should save given instance and use it to pass callbacks to plugin system.
+- `def tick(self) -> None` (optional) - If you wish to implement your plugin in single thread style, the `PluginSystem` 
+promises to call this method periodically (every 20ms let's say) in main thread, so you can perform some actions there.
+- `def handle_new_state(self, new_state: AudioSessionState) -> None` - The `PluginSystem` calls this method with new state
+when a client change something in a session. You should apply new state properties to specified session.
+- `def stop(self) -> None` (optional) - This method gets called when the application shutting down.
+
+`PluginSystem` Provides next interface for plugins:
+- `def handle_session_change`(self, session: AudioSessionState) -> None` - Plugins should call it and pass new state
+when they have new state. In order to create session plugins should simply pass new session state here.
+- `def remove_session(self, plugin_id: int, session_id: int) -> None` - Plugins should call this method when they
+want to remove a session.
+
+Note: When a plugin applies new state through `handle_new_state` call, the plugin must supply updated state to 
+`handle_session_change` of `PluginSystem`.
+
+### Controller
+
+### Transport
+Transport is a way how to communicate with clients. Transports are loadable plugins

AudioController.py

@@ -41,6 +41,12 @@ class PerSessionCallbacks(AudioSessionEvents):
     def on_session_disconnected(self, disconnect_reason, disconnect_reason_id):
         self.audio_controller.on_session_disconnected(self.pid, disconnect_reason, disconnect_reason_id)
 
+    def __str__(self) -> str:
+        return f'{self.__class__.__name__} - {id(self)} - {getattr(self, "pid", None)}'
+
+    def __repr__(self):
+        return self.__str__()
+
 
 class SessionCreateCallback(AudioSessionNotification):
     def __init__(self, audio_controller: 'AudioController'):
@@ -254,7 +260,7 @@ class AudioController:
                 self.set_volume(event.PID, event.volume)
 
     def start_blocking(self):
-        # self.perform_discover()
+        self.perform_discover()
         logger.debug(f'Starting blocking')
         self.view.start()
         while self.running:

AudioSessionState.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass, field
+
+Unchanged = type('Unchanged', tuple(), dict())
+unchanged = Unchanged()
+
+
+@dataclass
+class AudioSessionState:
+    """Represent audio session """
+    plugin_id: int  # Unique identifier of a plugin
+    id: int  # Unique identifier of a session within plugin
+    is_active: bool = unchanged
+    is_muted: bool = unchanged
+    name: str = unchanged  # Name of a session (i.e. name of a process)
+    volume: int = unchanged  # Volume in range(0, 101)

NetworkTransport.py

@@ -14,7 +14,7 @@ class NetworkTransport(TransportABC):
         self.view_rcv_callback = rcv_callback
 
         self._sock = socket.socket()
-        self._sock.bind(('localhost', 54683))
+        self._sock.bind(('0.0.0.0', 54683))
         self._sock.listen(100)
         self._sock.setblocking(False)
         self._selector.register(self._sock, selectors.EVENT_READ, self._accept)

PluginABC.py

@@ -0,0 +1,23 @@
+import typing
+from abc import ABC, abstractmethod
+
+from AudioSessionState import AudioSessionState
+
+if typing.TYPE_CHECKING:
+    from PluginSystem import PluginSystem
+
+
+class PluginABC(ABC):
+    @abstractmethod
+    def __init__(self, plugin_system: 'PluginSystem', plugin_id: int):
+        raise NotImplementedError
+
+    @abstractmethod
+    def handle_new_state(self, new_state: AudioSessionState) -> None:
+        raise NotImplementedError
+
+    def tick(self) -> None:
+        return
+
+    def stop(self) -> None:
+        return

PluginSystem.py

@@ -0,0 +1,80 @@
+import os
+import importlib
+import typing
+from time import sleep
+
+from loguru import logger
+
+from PluginABC import PluginABC
+from AudioSessionState import AudioSessionState
+
+PLUGINS_DIR: typing.Final = 'plugins'
+
+
+class PluginSystem:
+    """Class performs communication with plugins"""
+
+    def __init__(self):
+        self._id_counter = 0
+        self._plugins: dict[int, PluginABC] = dict()
+        self.keep_ticking = True
+
+    def shutdown(self, *args):
+        self.keep_ticking = False
+
+    def _plugin_load(self, plugin_path: typing.AnyStr, plugin_name: str):
+        try:
+            module = importlib.machinery.SourceFileLoader(plugin_name, plugin_path).load_module()
+            cls: type[PluginABC] = getattr(module, plugin_name, None)
+            constructor = getattr(cls, '__init__', None)
+            if not callable(constructor):
+                raise AttributeError(f'No __init__ method in {plugin_name} plugin')
+
+            instance = cls(self, self._id_counter)
+            if not isinstance(instance, PluginABC):
+                raise ValueError('Plugin class must be instance of PluginABC')
+
+            self._plugins[self._id_counter] = instance
+            self._id_counter += 1
+
+        except Exception:
+            logger.opt(exception=True).warning(f'Failed to load plugin {plugin_path!r} - {plugin_name!r}')
+
+    def plugins_load(self):
+        if len(self._plugins) > 0:
+            logger.warning(f"Can't load plugins twice")
+            return
+
+        for file_name in sorted(os.listdir(PLUGINS_DIR)):
+            if file_name.endswith('.py') and not file_name[0] in ['.', '_']:
+                path = os.path.join(PLUGINS_DIR, file_name)
+                plugin_name = file_name[:-3]
+                self._plugin_load(path, plugin_name)
+
+    def _plugins_stop(self):
+        for plugin in self._plugins.values():
+            plugin.stop()
+
+    def handle_session_change(self, session: AudioSessionState) -> None:
+        logger.debug(session)
+
+    def remove_session(self, plugin_id: int, session_id: int) -> None:
+        logger.debug(f'{plugin_id} - {session_id}')
+
+    def ticking(self) -> None:
+        while self.keep_ticking:
+            sleep(0.1)
+            for plugin in self._plugins.values():
+                plugin.tick()
+
+        self._plugins_stop()
+
+
+if __name__ == '__main__':
+    ps = PluginSystem()
+    import signal
+    signal.signal(signal.SIGINT, ps.shutdown)
+    signal.signal(signal.SIGTERM, ps.shutdown)
+
+    ps.plugins_load()
+    ps.ticking()

main.py

@@ -1,47 +1,52 @@
-import sys; sys.coinit_flags = 0  # noqa
+# import sys; sys.coinit_flags = 0  # noqa
 from loguru import logger
-import logging
-import signal
-
-
-class InterceptHandler(logging.Handler):
-    def emit(self, record):
-        # Get corresponding Loguru level if it exists
-        try:
-            level = logger.level(record.levelname).name
-        except ValueError:
-            level = record.levelno
-
-        # Find caller from where originated the logged message
-        frame, depth = logging.currentframe(), 2
-        while frame.f_code.co_filename == logging.__file__:
-            frame = frame.f_back
-            depth += 1
-
-        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
-
-
-logging.basicConfig(handlers=[InterceptHandler()])
+# import logging
+# import signal
+#
+#
+# class InterceptHandler(logging.Handler):
+#     def emit(self, record):
+#         # Get corresponding Loguru level if it exists
+#         try:
+#             level = logger.level(record.levelname).name
+#         except ValueError:
+#             level = record.levelno
+#
+#         # Find caller from where originated the logged message
+#         frame, depth = logging.currentframe(), 2
+#         while frame.f_code.co_filename == logging.__file__:
+#             frame = frame.f_back
+#             depth += 1
+#
+#         logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+#
+#
+# logging.basicConfig(handlers=[InterceptHandler()], level=0)
 
 from pycaw.pycaw import AudioUtilities
 import AudioController
 
 
-mgr = AudioUtilities.GetAudioSessionManager()
+# mgr = AudioUtilities.GetAudioSessionManager()
+#
+# audio_controller = AudioController.AudioController()
+#
+# signal.signal(signal.SIGTERM, audio_controller.shutdown_callback)
+# signal.signal(signal.SIGINT, audio_controller.shutdown_callback)
+#
+# callback = AudioController.SessionCreateCallback(audio_controller)
+#
+# mgr.RegisterSessionNotification(callback)
+# mgr.GetSessionEnumerator()
+#
+# audio_controller.start_blocking()
+#
+# mgr.UnregisterSessionNotification(callback)
+# audio_controller.pre_shutdown()
+def main():
+    ...
 
-audio_controller = AudioController.AudioController()
 
-signal.signal(signal.SIGTERM, audio_controller.shutdown_callback)
-signal.signal(signal.SIGINT, audio_controller.shutdown_callback)
-
-callback = AudioController.SessionCreateCallback(audio_controller)
-
-mgr.RegisterSessionNotification(callback)
-mgr.GetSessionEnumerator()
-
-audio_controller.start_blocking()
-
-mgr.UnregisterSessionNotification(callback)
-audio_controller.pre_shutdown()
-
-logger.trace(f'Shutdown completed')
+if __name__ == '__main__':
+    main()
+    logger.trace(f'Shutdown completed')

plugins/AIMPPlugin.py

@@ -0,0 +1,80 @@
+import typing
+from time import time
+from loguru import logger
+import pyaimp
+
+from PluginABC import PluginABC
+if typing.TYPE_CHECKING:
+    from PluginSystem import PluginSystem
+
+from AudioSessionState import AudioSessionState
+
+
+class AIMPPlugin(PluginABC):
+    def __init__(self, plugin_system: 'PluginSystem', plugin_id: int):
+        self.id = plugin_id
+        self.ps = plugin_system
+        self._prev_tick = time()
+        self.tick_interval_secs = 1
+        self._client: None | pyaimp.Client = None
+        self._latest_state: None | AudioSessionState = None
+
+    def _init_client(self) -> bool:
+        if self._is_client_alive():
+            return True
+
+        try:
+            self._client = pyaimp.Client()
+            return True
+
+        except RuntimeError:  # Unable to find AIMP window probably
+            logger.opt(exception=True).trace(f'Failed to init AIMP client')
+            return False
+
+    def _is_client_alive(self) -> bool:
+        if self._client is not None:
+            is_alive = self._client.get_version() is not None
+            if not is_alive:
+                self._client = None
+
+            return is_alive
+
+        return False
+
+    def _scheduled_tick(self):
+        if self._init_client():
+            state = AudioSessionState(
+                plugin_id=self.id, id=0,
+                is_active=self._client.get_playback_state() == pyaimp.PlayBackState.Playing, name='AIMP',
+                volume=self._client.get_volume(), is_muted=self._client.is_muted()
+            )
+            if state != self._latest_state:
+                self.ps.handle_session_change(state)
+                self._latest_state = state
+
+        else:  # If init failed we have to consider two situations
+            if self._latest_state is not None:  # It is not long ago became like that (i.e. AIMP shutdown)
+                self.ps.remove_session(self._latest_state.plugin_id, self._latest_state.id)
+                self._latest_state = None
+
+        # else: # If it is like that for a while then we have nothing to do
+
+    def handle_new_state(self, new_state: AudioSessionState) -> None:
+        if self._latest_state is None:
+            logger.warning(f'Got new_state while _latest_state is None {new_state=}, {self._client=}')
+            return
+
+        if new_state.volume != self._latest_state.volume:
+            self._client.set_volume(new_state.volume)
+
+        if new_state.is_muted != self._latest_state.is_muted:
+            self._client.set_muted(new_state.is_muted)
+
+        # Note: We do not set new state to _latest_state to let _scheduled_tick detect state change and send it
+
+    def tick(self) -> None:
+        time_now = time()
+        if (time_now - self._prev_tick) > self.tick_interval_secs:
+            self._prev_tick = time_now
+            self._scheduled_tick()
+

test.py

@@ -0,0 +1,67 @@
+import sys; sys.coinit_flags = 0  # noqa
+import signal
+from pycaw.pycaw import AudioUtilities
+from pycaw.callbacks import AudioSessionEvents, AudioSessionNotification
+from loguru import logger
+from time import sleep
+
+
+class Globals:
+    shutting_down = False
+
+
+def shutdown_callback(*args):
+    logger.info(f'Shutting down')
+    Globals.shutting_down = True
+
+
+class SessionCreateCallback(AudioSessionNotification):
+    references = list()
+
+    def on_session_created(self, new_session):
+        logger.info(f'Session created: {new_session}')
+        new_session.register_notification(SessionCallback(new_session.Process, 'by on_session_created'))
+        self.references.append(new_session)
+
+
+class SessionCallback(AudioSessionEvents):
+    def __init__(self, process, obtained_via):
+        self.process = getattr(process, '_name', None)
+        self.ov = obtained_via
+
+    def on_simple_volume_changed(self, new_volume, new_mute, event_context):
+        logger.info(f'{self.process} - {new_volume} - {self.ov}')
+
+    def on_state_changed(self, new_state, new_state_id):
+        logger.info(f'{self.process} - {new_state} - {self.ov}')
+
+    def on_session_disconnected(self, disconnect_reason, disconnect_reason_id):
+        logger.info(f'{self.process} - {disconnect_reason} - {self.ov}')
+
+
+def main():
+    signal.signal(signal.SIGTERM, shutdown_callback)
+    signal.signal(signal.SIGINT, shutdown_callback)
+
+    mgr = AudioUtilities.GetAudioSessionManager()
+    callback = SessionCreateCallback()
+    mgr.RegisterSessionNotification(callback)
+    mgr.GetSessionEnumerator()
+
+    sessions = AudioUtilities.GetAllSessions()
+    for session in sessions:
+        logger.info(f'Registering callback for {session}')
+        session.register_notification(SessionCallback(session.Process, 'by initial discover'))
+
+    while not Globals.shutting_down:
+        sleep(0.1)
+
+    mgr.UnregisterSessionNotification(callback)
+
+    for session in sessions:
+        logger.info(f'Unregistering callback for {session}')
+        session.unregister_notification()
+
+
+if __name__ == '__main__':
+    main()