oak umbra
I'm trying to implement a simple event handler class but for some reason I can't remove the function from the list.
class EventManager(PlayerStatus):
def __init__(self, login: Login, s_device_id: str | None = None) -> None:
super().__init__(login, s_device_id)
self._current_state = self.state # Need this to activate websocket
self.wlock = threading.Lock()
self._subscriptions: Dict[str, List[Callable[..., Any]]] = {}
self.listener = threading.Thread(target=self._listen, daemon=True)
self.listener.start()
def _subscribe_callable(self, event: str, func: Callable[..., Any]) -> None:
with self.wlock:
if event not in self._subscriptions:
self._subscriptions[event] = []
if func not in self._subscriptions[event]:
self._subscriptions[event].append(func)
else:
raise ValueError(f"Function {func.__name__} is already subscribed to event '{event}'")
def subscribe(self, event: str) -> Callable[..., Any]:
"""Decorator to subscribe a function to a Spotify websocket event."""
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
result = func(*args, **kwargs)
return result
self._subscribe_callable(event, func)
return wrapper
return decorator
def _emit(self, event: str, *args: Any, **kwargs: Any) -> None:
"""
Emit an event, triggering all subscribed functions.
Should only be called from the WebsocketStreamer thread.
"""
print(self._subscriptions)
if event in self._subscriptions:
for func in self._subscriptions[event]:
func(*args, **kwargs)
def unsubscribe(self, event: str, func: Callable[..., Any]) -> None:
"""Unsubscribe a function from an event."""
with self.wlock:
if event in self._subscriptions:
for i, f in enumerate(self._subscriptions[event]):
if f.__code__ is func.__code__:
del self._subscriptions[event][i]
break
def _listen(self) -> None:
while True:
event = self.get_packet()
if event is None or event.get("payloads") is None:
continue
for payload in event["payloads"]:
self._emit(payload["update_reason"], payload)
I've tried using __code__, I've tried id() [is], I've tried .remove. Nothing seems to work. Should I store the function id in the list or something?
jolly wraith
marble cliff
