must

1
import os
2
import ctypes
3
import backend_base
4
import log
5

6

7
class WinMMWrapper(backend_base.BaseWrapper):
8
    def __init__(self, mm_lib: ctypes.CDLL) -> None:
9
        super().__init__()
10
        self.lib = mm_lib
11
        if not self.lib:
12
            raise FileNotFoundError('Failed to load Windows MultiMedia library')
13
        self.mciSendStringW = self.wrap('mciSendStringW', args=(
14
            ctypes.c_wchar_p, ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p
15
        ), res=ctypes.c_ulong)
16
        self.mciGetErrorStringW = self.wrap('mciGetErrorStringW', args=(
17
            ctypes.c_ulong, ctypes.c_void_p, ctypes.c_uint
18
        ), res=ctypes.c_int)
19

20

21
class WinMMMusic(backend_base.BaseMusic):
22
    def __init__(self, bk: any, fp: str, alias: str) -> None:
23
        super().__init__(fp)
24
        self.bk = bk
25
        self.al = alias
26
        self.type = os.path.splitext(fp)[-1][1:]
27
        try:
28
            self.length = float(bk.send_warn(f'status {self.al} length', 'Failed to get music length')) / 1000
29
        except ValueError:
30
            self.length = 0.0
31

32
    def play(self) -> None:
33
        self.bk.send_warn(f'play {self.al}', 'Failed to play music')
34

35
    def stop(self) -> None:
36
        self.bk.send_warn(f'stop {self.al}', 'Failed to stop music')
37

38
    def is_playing(self) -> bool:
39
        return self.bk.send_warn(f'status {self.al} mode', 'Failed to get music state') in ('playing', 'paused')
40

41
    def set_volume(self, volume: float = 1.0) -> None:
42
        self.bk.send_warn(f'setaudio {self.al} volume to {round(volume * 1000)}', 'Failed to set music volume')
43

44
    def set_speed(self, volume: float = 1.0) -> None:
45
        self.bk.send_warn(f'set {self.al} speed {round(volume * 1000)}', 'Failed to set music speed')
46

47
    def set_paused(self, paused: bool) -> None:
48
        if paused == self.paused:
49
            return
50
        self.paused = paused
51
        self.bk.send_warn(('pause ' if paused else 'resume ') + self.al, 'Failed to set music paused')
52

53
    def set_pos(self, pos: float) -> None:
54
        self.bk.send_warn(f'seek {self.al} to {round(pos * 1000)}', 'Failed to set music position')
55
        self.bk.send_warn(f'play {self.al}', 'Failed to play music')
56

57
    def get_pos(self) -> float:
58
        res = self.bk.send_warn(f'status {self.al} position', 'Failed to get music position')
59
        try:
60
            return float(res) / 1000
61
        except ValueError:
62
            return 0.0
63

64
    def rewind(self) -> None:
65
        self.set_pos(0.0)
66

67
    def destroy(self) -> None:
68
        self.bk.send_warn(f'close {self.al}', 'Failed to close music')
69
        self.bk = None
70

71

72
class WinMMBackend(backend_base.BaseBackend):
73
    def __init__(self, app: any, mm_lib: ctypes.CDLL) -> None:
74
        super().__init__()
75
        self.title = 'WindowsMultimedia'
76
        self.app = app
77
        self.buffer_size = 1024
78
        self.mm = WinMMWrapper(mm_lib)
79

80
    def open_music(self, fp: str) -> WinMMMusic:
81
        alias = os.path.basename(fp).replace(' ', '')
82
        self.send_err(f'open "{fp}" alias {alias}', 'Failed to open music')
83
        self.send_warn(f'set {alias} time format milliseconds', 'Failed to set time format for music')
84
        return WinMMMusic(self, fp, alias)
85

86
    def destroy(self) -> None:
87
        self.mm = None
88
        self.app = None
89

90
    def get_audio_devices_names(self) -> list:
91
        return ['Default Device']
92

93
    def get_audio_drivers(self) -> list:
94
        return ['winmm']
95

96
    def get_current_audio_driver(self) -> str:
97
        return 'winmm'
98

99
    def get_current_audio_device_name(self) -> str:
100
        return 'Default Device'
101

102
    def send(self, command: str) -> tuple:
103
        buffer = ctypes.create_unicode_buffer(self.buffer_size + 1)
104
        code = self.mm.mciSendStringW(command, buffer, self.buffer_size, None)
105
        if code:
106
            new_buf = ctypes.create_unicode_buffer(self.buffer_size + 1)
107
            if self.mm.mciGetErrorStringW(code, buffer, self.buffer_size):
108
                return new_buf.value, True
109
            return 'Unknown Error', True
110
        return buffer.value, False
111

112
    def send_err(self, command: str, error_text: str) -> str:
113
        resp, is_err = self.send(command)
114
        if is_err:
115
            raise RuntimeError(error_text + ' (' + resp + ')')
116
        return resp
117

118
    def send_warn(self, command: str, error_text: str) -> str:
119
        resp, is_err = self.send(command)
120
        if is_err:
121
            log.warn(error_text + ' (' + resp + ')')
122
            return ''
123
        return resp
124