From 978d9585780a2e7b2cc9ca59f8b450926727cced Mon Sep 17 00:00:00 2001 From: DevilXD Date: Thu, 6 Jan 2022 14:33:00 +0100 Subject: [PATCH] Improve the drop progress tracking system --- .gitignore | 1 + channel.py | 21 +----- constants.py | 2 +- gui.py | 170 ++++++++++++++++++++++++++------------------ inventory.py | 18 ++--- twitch.py | 196 ++++++++++++++++++++++++++++++++++++++------------- 6 files changed, 261 insertions(+), 147 deletions(-) diff --git a/.gitignore b/.gitignore index 91f64dc..3af29b6 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ __pycache__ /cookies.jar /settings.json /*.spec +/log*.txt diff --git a/channel.py b/channel.py index 239ff65..4a73329 100644 --- a/channel.py +++ b/channel.py @@ -11,9 +11,7 @@ from typing import Any, Optional, TYPE_CHECKING from inventory import Game from exceptions import MinerException -from constants import ( - JsonType, BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, WATCH_INTERVAL, DROPS_ENABLED_TAG -) +from constants import JsonType, BASE_URL, GQL_OPERATIONS, ONLINE_DELAY, DROPS_ENABLED_TAG if TYPE_CHECKING: from twitch import Twitch @@ -252,7 +250,7 @@ class Channel: json_event = json.dumps(payload, separators=(",", ":")) return {"data": (b64encode(json_event.encode("utf8"))).decode("utf8")} - async def _send_watch(self) -> bool: + async def send_watch(self) -> bool: """ This uses the encoded payload on spade url to simulate watching the stream. Optimally, send every 60 seconds to advance drops. @@ -267,18 +265,3 @@ class Channel: self._spade_url, data=self._encode_payload() ) as response: return response.status == 204 - - async def watch_loop(self): - # last_watch is a timestamp of the last time we've sent a watch payload - # We need this because watch_loop can be cancelled and rescheduled multiple times - # in quick succession, and apparently Twitch doesn't like that very much - interval = WATCH_INTERVAL.total_seconds() - await asyncio.sleep(self._twitch._last_watch + interval - time()) - i = 0 - while True: - await self._send_watch() - if i == 0: - # ensure every 30 minutes that we don't have unclaimed points bonus - await self.claim_bonus() - i = (i + 1) % 30 - await asyncio.sleep(interval) diff --git a/constants.py b/constants.py index dc0a8f4..8b8c89b 100644 --- a/constants.py +++ b/constants.py @@ -29,7 +29,7 @@ COOKIES_PATH = "cookies.jar" PING_INTERVAL = timedelta(minutes=3) PING_TIMEOUT = timedelta(seconds=10) ONLINE_DELAY = timedelta(seconds=60) -WATCH_INTERVAL = timedelta(seconds=59) +WATCH_INTERVAL = timedelta(seconds=58.7) # Tags DROPS_ENABLED_TAG = "c2542d6d-cd10-4532-919b-3d19f30a768b" diff --git a/gui.py b/gui.py index ec46c73..668ec05 100644 --- a/gui.py +++ b/gui.py @@ -9,7 +9,9 @@ from math import log10, ceil from tkinter.font import Font from collections import namedtuple, OrderedDict from tkinter import Tk, ttk, StringVar, DoubleVar -from typing import Any, Optional, List, Dict, Set, TypedDict, Iterable, NoReturn, TYPE_CHECKING +from typing import ( + Any, Optional, List, Dict, Set, Tuple, TypedDict, Iterable, NoReturn, TYPE_CHECKING +) from version import __version__ from constants import WS_TOPICS_LIMIT, MAX_WEBSOCKETS, State @@ -324,7 +326,6 @@ class _DropVars(_BaseVars): class _ProgressVars(TypedDict): campaign: _CampaignVars drop: _DropVars - seconds: int class CampaignProgress: @@ -346,7 +347,6 @@ class CampaignProgress: "remaining": StringVar(), # as above "minutes": 0, # as above }, - "seconds": 1, # remaining seconds (common for both campaign and drop) } self._frame = frame = ttk.LabelFrame( master, text="Campaign Progress", padding=(4, 0, 4, 4) @@ -386,65 +386,48 @@ class CampaignProgress: variable=self._vars["drop"]["progress"], ).grid(column=0, row=10, columnspan=2) self._timer_task: Optional[asyncio.Task[None]] = None - self._update_time() + self._update_time(0) - def _update_time(self) -> bool: - # read vars - minutes_changed: bool = False - seconds: int = self._vars["seconds"] + @staticmethod + def _divmod(minutes: int, subone: bool = False) -> Tuple[int, int]: + hours, minutes = divmod(minutes, 60) + if subone and minutes > 0: + minutes -= 1 + return (hours, minutes) + + def _update_time(self, seconds: int): drop_vars: _DropVars = self._vars["drop"] campaign_vars: _CampaignVars = self._vars["campaign"] - drop_minutes: int = drop_vars["minutes"] - campaign_minutes: int = campaign_vars["minutes"] - # handle seconds - if seconds <= 0: - if drop_minutes > 0: - drop_minutes -= 1 - minutes_changed = True - if campaign_minutes > 0: - campaign_minutes -= 1 - minutes_changed = True - if minutes_changed: - seconds = 60 - if seconds > 0: - seconds -= 1 - # display time - hours, minutes = divmod(drop_minutes, 60) - drop_vars["remaining"].set(f"{hours:>2}:{minutes:02}:{seconds:02} remaining") - hours, minutes = divmod(campaign_minutes, 60) - campaign_vars["remaining"].set(f"{hours:>2}:{minutes:02}:{seconds:02} remaining") - # store back - self._vars["seconds"] = seconds - if minutes_changed: - drop_vars["minutes"] = drop_minutes - campaign_vars["minutes"] = campaign_minutes - # if there's no time left, stop the loop - if campaign_minutes + drop_minutes + seconds > 0: - return True - return False + dseconds = seconds % 60 + hours, minutes = self._divmod(drop_vars["minutes"], seconds < 60) + drop_vars["remaining"].set(f"{hours:>2}:{minutes:02}:{dseconds:02} remaining") + hours, minutes = self._divmod(campaign_vars["minutes"], seconds < 60) + campaign_vars["remaining"].set(f"{hours:>2}:{minutes:02}:{dseconds:02} remaining") async def _timer_loop(self): - run = self._update_time() - while run: + seconds = 60 + self._update_time(seconds) + while seconds > 0: await asyncio.sleep(1) - run = self._update_time() + seconds -= 1 + self._update_time(seconds) self._timer_task = None def start_timer(self): if self._timer_task is None: - self._vars["seconds"] = 1 - self._timer_task = asyncio.create_task(self._timer_loop()) + if self._vars["drop"]["minutes"] <= 0: + # if we're starting the timer at 0 drop minutes, all we need + # is a single instant time update setting seconds to 0 + self._update_time(0) + else: + self._timer_task = asyncio.create_task(self._timer_loop()) def stop_timer(self): if self._timer_task is not None: self._timer_task.cancel() self._timer_task = None - def restart_timer(self): - self.stop_timer() - self.start_timer() - - def update(self, drop: TimedDrop): + def display(self, drop: TimedDrop, *, countdown: bool = True): # campaign update campaign = drop.campaign vars_campaign = self._vars["campaign"] @@ -460,8 +443,15 @@ class CampaignProgress: vars_drop["progress"].set(drop.progress) vars_drop["percentage"].set(f"{drop.progress:6.1%}") vars_drop["minutes"] = drop.remaining_minutes - # reschedule our seconds update timer - self.restart_timer() + if countdown: + # reschedule our seconds update timer + self.stop_timer() + self.start_timer() + else: + # display the current remaining time at 0 seconds (after substracting the minute) + # this is because the watch loop will substract this minute + # right after the first watch payload is sent + self._update_time(0) class ConsoleOutput: @@ -745,8 +735,13 @@ class GUIManager: """ update = self._root.update while True: - update() + try: + update() + except tk.TclError: + # root has been destroyed + break await asyncio.sleep(0.05) + self._poll_task = None def unfocus(self, event): self._root.focus_set() @@ -823,22 +818,61 @@ if __name__ == "__main__": game=game_obj, viewers=viewers, ) - # Login form - gui.login.update("Login required", None) - # Game selector - # gui.games.set_games([ - # create_game(491115, "Paladins"), - # create_game(460630, "Tom Clancy's Rainbow Six Siege"), - # ]) - # game = gui.games.get_next_selection() - # game = gui.games.get_next_selection() - # game = gui.games.get_next_selection() - # Channel list - gui.channels.display(create_channel("PaladinsGame", 0, None, 0, 0)) - channel = create_channel("Traitus", 1, None, 0, 0) - gui.channels.display(channel) - gui.channels.display(create_channel("Testus", 2, "Paladins", 42, 1234567)) - gui.channels.set_watching(channel) - gui._root.update() - gui.channels.get_selection() - gui._root.mainloop() + + def create_drop( + campaign_name: str, + rewards: str, + claimed_drops: int, + total_drops: int, + current_minutes: int, + total_minutes: int, + ): + cd = claimed_drops + td = total_drops + cm = current_minutes + tm = total_minutes + mock = SimpleNamespace( + id="0", + campaign=SimpleNamespace( + name=campaign_name, + timed_drops={}, + claimed_drops=cd, + total_drops=td, + remaining_drops=td - cd, + progress=(cd * tm + cm) / (td * tm), + remaining_minutes=(td - cd) * tm - cm, + ), + rewards_text=lambda: rewards, + progress=cm/tm, + current_minutes=cm, + required_minutes=tm, + remaining_minutes=tm-cm, + ) + mock.campaign.timed_drops["0"] = mock + return mock + + async def main(): + # Drop progress + gui.progress.display(create_drop("Wardrobe Cleaning", "Fancy Pants", 2, 7, 134, 240)) + # Login form + gui.login.update("Login required", None) + # Game selector + gui.games.set_games([ + create_game(491115, "Paladins"), + # create_game(460630, "Tom Clancy's Rainbow Six Siege"), + ]) + gui.games.get_next_selection() + gui.games.get_next_selection() + gui.games.get_next_selection() + # Channel list + gui.channels.display(create_channel("PaladinsGame", 0, None, 0, 0)) + channel = create_channel("Traitus", 1, None, 0, 0) + gui.channels.display(channel) + gui.channels.display(create_channel("Testus", 2, "Paladins", 42, 1234567)) + gui.channels.set_watching(channel) + gui._root.update() + gui.channels.get_selection() + # asyncio stuff + loop = asyncio.get_event_loop() + loop.create_task(main()) + loop.run_until_complete(gui._poll()) diff --git a/inventory.py b/inventory.py index dc78ee8..7e57f06 100644 --- a/inventory.py +++ b/inventory.py @@ -96,20 +96,20 @@ class TimedDrop(BaseDrop): @property def remaining_minutes(self) -> int: - return self.required_minutes - self.current_minutes + 1 + return self.required_minutes - self.current_minutes @property def progress(self) -> float: return self.current_minutes / self.required_minutes - def update(self, message: JsonType): - # See Twitch.process_drop for message examples - msg_type = message["type"] - if msg_type == "drop-progress": - self.current_minutes = message["data"]["current_progress_min"] - self.required_minutes = message["data"]["required_progress_min"] - elif msg_type == "drop-claim": - self.claim_id = message["data"]["drop_instance_id"] + def update_claim(self, claim_id: str): + self.claim_id = claim_id + + def update_minutes(self, minutes: int): + self.current_minutes = minutes + + def display(self, *, countdown: bool = True): + self.campaign._twitch.gui.progress.display(self, countdown=countdown) def bump_minutes(self): if self.current_minutes < self.required_minutes: diff --git a/twitch.py b/twitch.py index 63b9bb1..c189150 100644 --- a/twitch.py +++ b/twitch.py @@ -29,6 +29,7 @@ from constants import ( COOKIES_PATH, AUTH_URL, GQL_URL, + WATCH_INTERVAL, GQL_OPERATIONS, DROPS_ENABLED_TAG, GQLOperation, @@ -66,6 +67,7 @@ class Twitch: self._watching_channel: Optional[Channel] = None self._watching_task: Optional[asyncio.Task[Any]] = None self._last_watch = time() - 60 + self._drop_update: Optional[asyncio.Future[bool]] = None # Websocket self.websocket = WebsocketPool(self) # Runner task @@ -188,6 +190,13 @@ class Twitch: await self.websocket.start() self.gui.games.set_games(games) selected_game = self.gui.games.get_selection() + # pre-display the active drop without a countdown + for campaign in self.inventory: + if campaign.active and campaign.game == selected_game: + active_drop = campaign.get_active_drop() + if active_drop is not None: + active_drop.display(countdown=False) + break self.change_state(State.CHANNEL_CLEANUP) elif self._state is State.CHANNEL_FETCH: if selected_game is None: @@ -262,6 +271,80 @@ class Twitch: self.change_state(State.CHANNEL_CLEANUP) await self._state_change.wait() + async def _watch_loop(self, channel: Channel): + # last_watch is a timestamp of the last time we've sent a watch payload + # We need this because watch_loop can be cancelled and rescheduled multiple times + # in quick succession, and apparently Twitch doesn't like that very much + interval = WATCH_INTERVAL.total_seconds() + await asyncio.sleep(self._last_watch + interval - time()) + i = 0 + while True: + await channel.send_watch() + self._last_watch = time() + self._drop_update = asyncio.Future() + try: + updated = await asyncio.wait_for(self._drop_update, timeout=10) + except asyncio.TimeoutError: + # there was no websocket update within 10s + self._drop_update = None + selected_game = self.gui.games.get_selection() + drop = None + for campaign in self.inventory: + if campaign.active and campaign.game == selected_game: + drop = campaign.get_active_drop() + break + if drop is not None and drop.campaign.active: + drop.bump_minutes() + drop.display() + else: + logger.error("Active drop search failed") + else: + self._drop_update = None + if not updated: + # there was no websocket update, or the update was for an unrelated drop + # we need to use GQL to get the current progress + context = await self.gql_request(GQL_OPERATIONS["CurrentDrop"]) + drop_data: JsonType = context["data"]["currentUser"]["dropCurrentSession"] + drop_id = drop_data["dropID"] + drop = self.get_drop(drop_id) + if drop is None: + logger.error(f"Missing drop: {drop_id}") + elif not drop.campaign.active: + # Sometimes, even GQL fails to give us the correct drop. + # In that case, we can use the locally cached inventory to try + # and put together the drop that we're actually mining right now. + selected_game = self.gui.games.get_selection() + drop = None + for campaign in self.inventory: + if campaign.active and campaign.game == selected_game: + drop = campaign.get_active_drop() + break + if drop is not None and drop.campaign.active: + drop.bump_minutes() + drop.display() + else: + logger.error("Active drop search failed") + else: + # drop is not None and campaign.active + drop.update_minutes(drop_data["currentMinutesWatched"]) + drop.display() + with open("log.txt", 'a') as file: + print( + time(), + drop_id, + "GQL", + drop_data["currentMinutesWatched"], + drop.current_minutes, + drop.is_claimed, + sep='\t', + file=file, + ) + if i == 0: + # ensure every 30 minutes that we don't have unclaimed points bonus + await channel.claim_bonus() + i = (i + 1) % 30 + await asyncio.sleep(self._last_watch + interval - time()) + def watch(self, channel: Channel): if self.is_watching(channel): # we're already watching the same channel, so there's no point switching @@ -270,8 +353,7 @@ class Twitch: self._watching_task.cancel() self.gui.channels.set_watching(channel) self._watching_channel = channel - self._watching_task = asyncio.create_task(channel.watch_loop()) - self.gui.progress.start_timer() + self._watching_task = asyncio.create_task(self._watch_loop(channel)) def stop_watching(self): self.gui.progress.stop_timer() @@ -316,58 +398,63 @@ class Twitch: return drop_id: str = message["data"]["drop_id"] drop: Optional[TimedDrop] = self.get_drop(drop_id) - if msg_type == "drop-claim" and drop is None: - logger.error( - f"Received a drop claim ID for a non-existing drop: {drop_id}\n" - f"Drop claim ID: {message['data']['drop_instance_id']}" - ) - return - # Sometimes, the drop update we receive doesn't actually match what we're mining. - # This is a Twitch bug workaround: use GQL to get the current drop progress. - if msg_type == "drop-progress" and (drop is None or not drop.campaign.active): - logger.debug( - "Received a drop update for an inactive campaign, using drop context instead" - ) - context = await self.gql_request(GQL_OPERATIONS["CurrentDrop"]) - drop_data = context["data"]["currentUser"]["dropCurrentSession"] - drop_id = drop_data["dropID"] - drop = self.get_drop(drop_id) - if drop is None: - logger.warning(f"Received an update for a non-existing drop: {drop_id}") - return - if not drop.campaign.active: - # Sometimes, even GQL fails to give us the correct drop. - # In that case, we can use the locally cached inventory to try and put together - # the drop that we're actually mining right now. - # TODO: Find a better way of figuring out the correct game - game = drop.campaign.game - drop = None - for campaign in self.inventory: - if campaign.game == game: - drop = campaign.get_active_drop() - break - if drop is None or not drop.campaign.active: - logger.error("Active drop search failed") - return - drop.bump_minutes() - self.gui.progress.update(drop) - return - else: - # TODO: Use a cleaner solution than modifying the raw payload - message["data"]["current_progress_min"] = drop_data["currentMinutesWatched"] - message["data"]["required_progress_min"] = drop_data["requiredMinutesWatched"] - assert drop is not None - drop.update(message) if msg_type == "drop-claim": + if drop is None: + logger.error( + f"Received a drop claim ID for a non-existing drop: {drop_id}\n" + f"Drop claim ID: {message['data']['drop_instance_id']}" + ) + return + drop.update_claim(message["data"]["drop_instance_id"]) campaign = drop.campaign - await drop.claim() - self.gui.print( - f"Claimed drop: {drop.rewards_text()} " - f"({campaign.claimed_drops}/{campaign.total_drops})" - ) + mined = await drop.claim() + if mined: + self.gui.print( + f"Claimed drop: {drop.rewards_text()} " + f"({campaign.claimed_drops}/{campaign.total_drops})" + ) + else: + logger.error(f"Drop claim failed! Drop ID: {drop_id}") if campaign.remaining_drops == 0: self.change_state(State.INVENTORY_FETCH) - self.gui.progress.update(drop) + return + # About 6s after claiming the drop, next drop can be started + # by resending the watch payload + await asyncio.sleep(6) + # Force-restart the watch task to send the watch payload right away + channel = self._watching_channel + if channel is not None: + self.stop_watching() + self._last_watch = time() - 60 + self.watch(channel) + return + assert msg_type == "drop-progress" + if self._drop_update is None: + # we aren't actually waiting for a progress update right now, so we can just + # ignore the event this time + return + elif drop is not None and drop.campaign.active: + drop.update_minutes(message["data"]["current_progress_min"]) + drop.display() + self._drop_update.set_result(True) + self._drop_update = None # TODO: remove this together with debug code below + with open("log.txt", 'a') as file: + print( + time(), + drop_id, + "WS", + message["data"]["current_progress_min"], + drop.current_minutes, + drop.is_claimed, + sep='\t', + file=file, + ) + else: + # Sometimes, the drop update we receive doesn't actually match what we're mining. + # This is a Twitch bug workaround: signal the watch loop to use GQL + # to get the current drop progress. + self._drop_update.set_result(False) + self._drop_update = None @task_wrapper async def process_points(self, user_id: int, message: JsonType): @@ -589,6 +676,15 @@ class Twitch: self.inventory = [ DropsCampaign(self, data) for data in inventory["dropCampaignsInProgress"] ] + context = await self.gql_request(GQL_OPERATIONS["CurrentDrop"]) + drop_data = context["data"]["currentUser"]["dropCurrentSession"] + with open("log.txt", 'a') as file: + if drop_data is None: + print(time(), None, sep='\t', file=file, flush=True) + else: + print( + time(), drop_data.get("currentMinutesWatched"), sep='\t', file=file, flush=True + ) def get_drop(self, drop_id: str) -> Optional[TimedDrop]: for campaign in self.inventory: