from threading import RLock from typing import Optional, Iterator, Dict, List import sys from itertools import cycle import chube_youtube from channel import Channel, Subscriber from chube_enums import * from chube_ws import Resolver, Message, start_server, make_message class Chueue: _lock: RLock _queue: List[int] _codes: Dict[int, str] _id_iter: Iterator[int] _played_queue: Optional[List[int]] _repeat_enabled: bool = False def __init__(self): self._lock = RLock() self._queue = [] self._codes = dict() self._id_iter = cycle(range(sys.maxsize)) self._played_queue = None def add(self, code): with self: song_id = next(self._id_iter) self._queue.append(song_id) self._codes[song_id] = code return song_id def remove(self, song_id): with self: self._queue.remove(song_id) self._codes.pop(song_id) def move(self, song_id, displacement): with self: i = self._queue.index(song_id) new_i = min(len(self._queue) - 1, max(0, i + displacement)) self._queue.pop(i) self._queue.insert(new_i, song_id) return new_i - i def pop(self): with self: if len(self._queue) <= 0: if self._repeat_enabled and len(self._played_queue) > 0: self._queue = self._played_queue self._played_queue = [] else: return None song_id = self._queue.pop(0) if self._repeat_enabled: code = self._codes[song_id] self._played_queue.append(song_id) else: code = self._codes.pop(song_id) return self.as_song(song_id, code) def set_repeat_enabled(self, enable, playback_song): self._repeat_enabled = enable if enable: if playback_song is not None: self._played_queue = [playback_song["id"]] self._codes[playback_song["id"]] = playback_song["code"] else: self._played_queue = [] else: self._played_queue = None def is_repeat_enabled(self): return self._repeat_enabled def as_song(self, song_id, code=None): if code is None: code = self._codes[song_id] return {"id": song_id, "code": code} def as_lists(self): with self: queue_as_list = list(map(self.as_song, self._queue)) played_as_list = list(map(self.as_song, self._played_queue)) if self.is_repeat_enabled() else None return {"next": queue_as_list, "previous": played_as_list} def lock(self): self._lock.acquire() def unlock(self): self._lock.release() def __enter__(self): self.lock() def __exit__(self, exc_type, exc_val, exc_tb): self.unlock() def __len__(self): return len(self._queue) class Playback: _song: Optional[Dict] = None _state: PlayerState = PlayerState.LIST_END lock: RLock() def __init__(self): self.lock = RLock() def set_song(self, song): with self.lock: self._song = song def get_song(self): with self.lock: return self._song def get_song_id(self): with self.lock: if self._song is not None: return self._song["id"] else: return None def get_state(self): return self._state def set_state(self, state): self._state = state class Room: chueue: Chueue channel: Channel controller: Optional[Subscriber] controller_lock: RLock playback: Playback def __init__(self): self.chueue = Chueue() self.channel = Channel() self.controller_lock = RLock() self.playback = Playback() self.controller = None rooms: Dict[str, Room] = dict() async def request_state_processor(ws, data, path): room = rooms[path] await ws.send(make_message(Message.STATE, { "lists": room.chueue.as_lists(), "playing": room.playback.get_song(), "state": room.playback.get_state().value })) async def request_list_operation_processor(ws, data, path): room = rooms[path] chueue = room.chueue op = data["op"] message = None if op == QueueOp.ADD.value: kind = data["kind"] if kind == YoutubeResourceType.VIDEO.value: code = data["code"] song_id = chueue.add(code) message = make_message(Message.LIST_OPERATION, {"op": QueueOp.ADD.value, "items": [{"code": code, "id": song_id}]}) elif kind == YoutubeResourceType.PLAYLIST.value: code = data["code"] playlist_items = await chube_youtube.get_all_playlist_items(code) response_items = [] with room.chueue: for item in playlist_items["items"]: code = item["snippet"]["resourceId"]["videoId"] song_id = chueue.add(code) response_items.append({"code": code, "id": song_id, "snippet": item["snippet"]}) message = make_message(Message.LIST_OPERATION, {"op": QueueOp.ADD.value, "items": response_items}) with room.playback.lock: if room.playback.get_state() == PlayerState.LIST_END: playing = chueue.pop() if playing is not None: room.playback.set_state(PlayerState.PLAYING) room.playback.set_song(playing) elif op == QueueOp.DEL.value: song_id = data["id"] chueue.remove(song_id) message = make_message(Message.LIST_OPERATION, {"op": QueueOp.DEL.value, "items": [{"id": song_id}]}) elif op == QueueOp.MOVE.value: song_id = data["id"] displacement = data["displacement"] actual_displacement = chueue.move(song_id, displacement) if actual_displacement != 0: message = make_message(Message.LIST_OPERATION, {"op": QueueOp.MOVE.value, "items": [{"id": song_id, "displacement": actual_displacement}]}) if message is not None: await room.channel.send(message) async def media_action_processor(ws, data, path): room = rooms[path] action = data["action"] send_next = False if action == MediaAction.NEXT.value: current_id = data["current_id"] with room.playback.lock, room.chueue: old_song_id = room.playback.get_song_id() if old_song_id == current_id: send_next = True new_song = play_next_song(room) if new_song is None: new_song_id = None else: new_song_id = new_song["id"] if send_next: await room.channel.send(make_message( Message.MEDIA_ACTION, {"action": MediaAction.NEXT.value, "ended_id": old_song_id, "current_id": new_song_id})) if action == MediaAction.PLAY.value or send_next: send_play = False with room.playback.lock: if room.playback.get_state() == PlayerState.PAUSED: send_play = True room.playback.set_state(PlayerState.PLAYING) if send_play: await room.channel.send(make_message(Message.MEDIA_ACTION, {"action": MediaAction.PLAY.value})) if action == MediaAction.PAUSE.value: send_pause = False with room.playback.lock: if room.playback.get_state() == PlayerState.PLAYING: send_pause = True room.playback.set_state(PlayerState.PAUSED) if send_pause: await room.channel.send(make_message(Message.MEDIA_ACTION, {"action": MediaAction.PAUSE.value})) if action == MediaAction.REPEAT.value: enable = data["enable"] if room.chueue.is_repeat_enabled() != enable: with room.chueue: room.chueue.set_repeat_enabled(enable, room.playback.get_song()) await room.channel.send(make_message(Message.MEDIA_ACTION, {"action": MediaAction.REPEAT.value, "enable": enable})) async def obtain_control_processor(ws, data, path): room = rooms[path] await obtain_control(ws, room) async def release_control_processor(ws, data, path): room = rooms[path] if len(room.channel.subscribers) > 1: await release_control(ws, room) else: pass # TODO error here def play_next_song(room): new_song = room.chueue.pop() room.playback.set_song(new_song) if new_song is None: room.playback.set_state(PlayerState.LIST_END) return new_song async def song_end_processor(ws, data, path): room = rooms[path] old_song_id = data["id"] with room.controller_lock, room.playback.lock: if room.controller is not None and ws is room.controller.ws and old_song_id == room.playback.get_song_id(): new_song = play_next_song(room) if new_song is None: new_song_id = None else: new_song_id = new_song["id"] await room.channel.send( make_message(Message.SONG_END, {"ended_id": old_song_id, "current_id": new_song_id})) async def player_enabled_processor(ws, data, path): room = rooms[path] room.channel.subscribers[ws].player_enabled = data["enabled"] if data["enabled"]: with room.controller_lock: if room.controller is None: await obtain_control(ws, room) else: await release_control(ws, room) # TODO There is some potential concurrent bug here, when the controller loses/releases control right before a song end. async def obtain_control(ws, room: Room): with room.controller_lock: if room.controller is None or room.controller.ws is not ws: prev_controller = room.controller room.controller = room.channel.subscribers[ws] await ws.send(make_message(Message.OBTAIN_CONTROL)) if prev_controller is not None: await prev_controller.ws.send(make_message(Message.RELEASE_CONTROL)) async def release_control(ws, room: Room): with room.controller_lock: if room.controller is not None and room.controller.ws is ws: room.controller = next(room.channel.get_player_enabled_subscribers(), None) if room.controller is not None: await room.controller.ws.send(make_message(Message.OBTAIN_CONTROL)) await ws.send(make_message(Message.RELEASE_CONTROL)) async def on_connect(ws, path): if path not in rooms: rooms[path] = Room() room = rooms[path] room.channel.subscribe(ws) async def on_disconnect(ws, path): room = rooms[path] room.channel.unsubscribe(ws) await release_control(ws, room) # with room.controller_lock: # if room.controller is None: # room.playback. def make_resolver(): resolver = Resolver() resolver.register(Message.STATE, request_state_processor) resolver.register(Message.LIST_OPERATION, request_list_operation_processor) resolver.register(Message.MEDIA_ACTION, media_action_processor) resolver.register(Message.PLAYER_ENABLED, player_enabled_processor) resolver.register(Message.OBTAIN_CONTROL, obtain_control_processor) resolver.register(Message.RELEASE_CONTROL, release_control_processor) resolver.register(Message.SONG_END, song_end_processor) search_resolver = chube_youtube.make_resolver() resolver.add_all(search_resolver) return resolver def init_rooms(): # rooms["main"] = Room() pass if __name__ == "__main__": player_resolver = make_resolver() init_rooms() start_server(player_resolver, on_connect, on_disconnect)