GitHub/TwitchDropsMiner
1
0
Fork 0
mirror of https://github.com/rangermix/TwitchDropsMiner.git synced 2026-06-08 21:34:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a prioritization strategy for games outside of the priority list

Browse Source
This commit is contained in:
DevilXD
2024-09-11 22:58:48 +02:00
parent ab37b0a143
commit a30e8d3ce4
7 changed files with 123 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
constants.py
View File

@@ -105,6 +105,7 @@ JsonType = Dict[str, Any]
URLType = NewType("URLType", str)
TopicProcess: TypeAlias = "abc.Callable[[int, JsonType], Any]"
# Values
MAX_INT = sys.maxsize
BASE_TOPICS = 3
MAX_WEBSOCKETS = 8
WS_TOPICS_LIMIT = 50
@@ -223,6 +224,12 @@ class State(Enum):
EXIT = auto()
class PriorityMode(Enum):
PRIORITY_ONLY = 0
ENDING_SOONEST = 1
LOW_AVBL_FIRST = 2
class GQLOperation(JsonType):
def __init__(self, name: str, sha256: str, *, variables: JsonType | None = None):
super().__init__(

57
gui.py
View File

@@ -40,6 +40,7 @@ from constants import (
WS_TOPICS_LIMIT,
OUTPUT_FORMATTER,
State,
PriorityMode,
)
if sys.platform == "win32":
from registry import RegistryKey, ValueType, ValueNotFound
@@ -388,18 +389,24 @@ class SelectCombobox(ttk.Combobox):
self,
master: tk.Misc,
*args,
width_offset: int = 0,
width: int | None = None,
textvariable: tk.StringVar,
values: list[str] | tuple[str, ...],
command: abc.Callable[[tk.Event[SelectCombobox]], None] | None = None,
**kwargs,
) -> None:
if width is None:
width = max(len(v) for v in values)
width += width_offset
super().__init__(
master,
*args,
width=width,
values=values,
state="readonly",
exportselection=False,
textvariable=textvariable,
state="readonly",
**kwargs,
)
if command is not None:
@@ -1216,7 +1223,9 @@ class InventoryOverview:
self._cache: ImageCache = manager._cache
self._settings: Settings = manager._twitch.settings
self._filters = {
"not_linked": IntVar(master, 1),
"not_linked": IntVar(
master, self._settings.priority_mode is PriorityMode.PRIORITY_ONLY
),
"upcoming": IntVar(master, 1),
"expired": IntVar(master, 0),
"excluded": IntVar(master, 0),
@@ -1304,7 +1313,7 @@ class InventoryOverview:
excluded = bool(self._filters["excluded"].get())
upcoming = bool(self._filters["upcoming"].get())
finished = bool(self._filters["finished"].get())
priority_only = self._settings.priority_only
priority_only = self._settings.priority_mode is PriorityMode.PRIORITY_ONLY
if (
campaign.remaining_minutes > 0 # don't show sub-only campaigns
and (not_linked or campaign.linked)
@@ -1534,23 +1543,32 @@ class _SettingsVars(TypedDict):
proxy: StringVar
autostart: IntVar
language: StringVar
priority_only: IntVar
priority_mode: StringVar
tray_notifications: IntVar
class SettingsPanel:
AUTOSTART_NAME: str = "TwitchDropsMiner"
AUTOSTART_KEY: str = "HKCU/Software/Microsoft/Windows/CurrentVersion/Run"
PRIORITY_MODES: dict[PriorityMode, str] = {
PriorityMode.PRIORITY_ONLY: _("gui", "settings", "priority_modes", "priority_only"),
PriorityMode.ENDING_SOONEST: _("gui", "settings", "priority_modes", "ending_soonest"),
PriorityMode.LOW_AVBL_FIRST: _("gui", "settings", "priority_modes", "low_availability"),
}
def __init__(self, manager: GUIManager, master: ttk.Widget):
self._twitch = manager._twitch
self._settings: Settings = manager._twitch.settings
priority_mode = self._settings.priority_mode
if priority_mode not in self.PRIORITY_MODES:
priority_mode = PriorityMode.PRIORITY_ONLY
self._settings.priority_mode = priority_mode
self._vars: _SettingsVars = {
"autostart": IntVar(master, 0),
"language": StringVar(master, _.current),
"proxy": StringVar(master, str(self._settings.proxy)),
"tray": IntVar(master, self._settings.autostart_tray),
"autostart": IntVar(master, 0),
"priority_only": IntVar(master, self._settings.priority_only),
"priority_mode": StringVar(master, self.PRIORITY_MODES[priority_mode]),
"tray_notifications": IntVar(master, self._settings.tray_notifications),
}
self._game_names: set[str] = set()
@@ -1577,7 +1595,6 @@ class SettingsPanel:
ttk.Label(language_frame, text="Language 🌐 (requires restart): ").grid(column=0, row=0)
SelectCombobox(
language_frame,
width=12,
values=list(_.languages),
textvariable=self._vars["language"],
command=lambda e: setattr(self._settings, "language", self._vars["language"].get()),
@@ -1607,10 +1624,13 @@ class SettingsPanel:
command=self.update_notifications,
).grid(column=1, row=irow, sticky="w")
ttk.Label(
checkboxes_frame, text=_("gui", "settings", "general", "priority_only")
checkboxes_frame, text=_("gui", "settings", "general", "priority_mode")
).grid(column=0, row=(irow := irow + 1), sticky="e")
ttk.Checkbutton(
checkboxes_frame, variable=self._vars["priority_only"], command=self.priority_only
SelectCombobox(
checkboxes_frame,
command=self.priority_mode,
textvariable=self._vars["priority_mode"],
values=list(self.PRIORITY_MODES.values()),
).grid(column=1, row=irow, sticky="w")
# proxy frame
@@ -1808,13 +1828,6 @@ class SettingsPanel:
self.update_excluded_choices()
self.update_priority_choices()
def priorities(self) -> dict[str, int]:
# NOTE: we shift the indexes so that 0 can be used as the default one
size = self._priority_list.size()
return {
game_name: size - i for i, game_name in enumerate(self._priority_list.get(0, "end"))
}
def priority_add(self) -> None:
game_name: str = self._priority_entry.get()
if not game_name:
@@ -1868,8 +1881,12 @@ class SettingsPanel:
self._settings.alter()
self.update_priority_choices()
def priority_only(self) -> None:
self._settings.priority_only = bool(self._vars["priority_only"].get())
def priority_mode(self, event: tk.Event[ttk.Combobox]) -> None:
mode_name: str = self._vars["priority_mode"].get()
for value, name in self.PRIORITY_MODES.items():
if mode_name == name:
self._settings.priority_mode = value
break
def exclude_add(self) -> None:
game_name: str = self._exclude_entry.get()
@@ -2390,11 +2407,11 @@ if __name__ == "__main__":
proxy=URL(),
autostart=False,
language="English",
priority_only=False,
autostart_tray=False,
exclude={"Lit Game"},
tray_notifications=True,
alter=lambda: None,
priority_mode=PriorityMode.PRIORITY_ONLY,
)
)
mock.change_state = lambda state: mock.gui.print(f"State change: {state.value}")

13
inventory.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import re
import math
from itertools import chain
from typing import TYPE_CHECKING
from functools import cached_property
@@ -232,6 +233,14 @@ class TimedDrop(BaseDrop):
return 1.0
return self.current_minutes / self.required_minutes
@property
def availability(self) -> float:
if not self._base_can_earn():
# this verifies "self.total_remaining_minutes > 0" and "now < self.ends_at"
return math.inf
now = datetime.now(timezone.utc)
return ((self.ends_at - now).total_seconds() / 60) / self.total_remaining_minutes
def _base_earn_conditions(self) -> bool:
return super()._base_earn_conditions() and self.required_minutes > 0
@@ -346,6 +355,10 @@ class DropsCampaign:
def progress(self) -> float:
return sum(d.progress for d in self.drops) / self.total_drops
@property
def availability(self) -> float:
return min(d.availability for d in self.drops)
def _on_claim(self) -> None:
invalidate_cache(self, "finished", "claimed_drops", "remaining_drops")
for drop in self.drops:

8
settings.py
View File

@@ -5,7 +5,7 @@ from typing import Any, TypedDict, TYPE_CHECKING
from yarl import URL
from utils import json_load, json_save
from constants import SETTINGS_PATH, DEFAULT_LANG
from constants import SETTINGS_PATH, DEFAULT_LANG, PriorityMode
if TYPE_CHECKING:
from main import ParsedArgs
@@ -16,21 +16,21 @@ class SettingsFile(TypedDict):
language: str
exclude: set[str]
priority: list[str]
priority_only: bool
autostart_tray: bool
connection_quality: int
tray_notifications: bool
priority_mode: PriorityMode
default_settings: SettingsFile = {
"proxy": URL(),
"priority": [],
"exclude": set(),
"priority_only": True,
"autostart_tray": False,
"connection_quality": 1,
"language": DEFAULT_LANG,
"tray_notifications": True,
"priority_mode": PriorityMode.PRIORITY_ONLY,
}
@@ -48,10 +48,10 @@ class Settings:
language: str
exclude: set[str]
priority: list[str]
priority_only: bool
autostart_tray: bool
connection_quality: int
tray_notifications: bool
priority_mode: PriorityMode
PASSTHROUGH = ("_settings", "_args", "_altered")

16
translate.py
View File

@@ -165,12 +165,19 @@ class GUISettingsGeneral(TypedDict):
autostart: str
tray: str
tray_notifications: str
priority_only: str
priority_mode: str
proxy: str
class GUIPriorityModes(TypedDict):
priority_only: str
ending_soonest: str
low_availability: str
class GUISettings(TypedDict):
general: GUISettingsGeneral
priority_modes: GUIPriorityModes
game_name: str
priority: str
exclude: str
@@ -362,9 +369,14 @@ default_translation: Translation = {
"autostart": "Autostart: ",
"tray": "Autostart into tray: ",
"tray_notifications": "Tray notifications: ",
"priority_only": "Priority Only: ",
"priority_mode": "Priority mode: ",
"proxy": "Proxy (requires restart):",
},
"priority_modes": {
"priority_only": "Priority list only",
"ending_soonest": "Ending soonest",
"low_availability": "Low availability first",
},
"game_name": "Game name",
"priority": "Priority",
"exclude": "Exclude",

58
twitch.py
View File

@@ -42,6 +42,7 @@ from utils import (
)
from constants import (
CALL,
MAX_INT,
DUMP_PATH,
COOKIES_PATH,
MAX_CHANNELS,
@@ -49,6 +50,7 @@ from constants import (
WATCH_INTERVAL,
State,
ClientType,
PriorityMode,
WebsocketTopic,
)
@@ -419,7 +421,7 @@ class Twitch:
# State management
self._state: State = State.IDLE
self._state_change = asyncio.Event()
self.wanted_games: dict[Game, int] = {}
self.wanted_games: list[Game] = []
self.inventory: list[DropsCampaign] = []
self._drops: dict[str, TimedDrop] = {}
self._mnt_triggers: deque[datetime] = deque()
@@ -545,16 +547,16 @@ class Twitch:
"""
Return a priority number for a given channel.
Higher number, higher priority.
Priority 0 is given to channels streaming a game not on the priority list.
Priority -1 is given to OFFLINE channels, or channels streaming no particular games.
0 has the highest priority.
Higher numbers -> lower priority.
MAX_INT (a really big number) signifies the lowest possible priority.
"""
if (game := channel.game) is None:
# None when OFFLINE or no game set
return -1
elif game not in self.wanted_games:
return 0
return self.wanted_games[game]
if (
(game := channel.game) is None # None when OFFLINE or no game set
or game not in self.wanted_games # we don't care about the played game
):
return MAX_INT
return self.wanted_games.index(game)
@staticmethod
def _viewers_key(channel: Channel) -> int:
@@ -633,23 +635,35 @@ class Twitch:
await drop.claim()
# figure out which games we want
self.wanted_games.clear()
priorities = self.gui.settings.priorities()
exclude = self.settings.exclude
priority = self.settings.priority
priority_only = self.settings.priority_only
priority_mode = self.settings.priority_mode
priority_only = priority_mode is PriorityMode.PRIORITY_ONLY
next_hour = datetime.now(timezone.utc) + timedelta(hours=1)
for campaign in self.inventory:
game = campaign.game
# sorted_campaigns: list[DropsCampaign] = list(self.inventory)
sorted_campaigns: list[DropsCampaign] = self.inventory
if not priority_only:
if priority_mode is PriorityMode.ENDING_SOONEST:
sorted_campaigns.sort(key=lambda c: c.ends_at)
elif priority_mode is PriorityMode.LOW_AVBL_FIRST:
sorted_campaigns.sort(key=lambda c: c.availability)
sorted_campaigns.sort(
key=lambda c: (
priority.index(c.game.name) if c.game.name in priority else MAX_INT
)
)
for campaign in sorted_campaigns:
game: Game = campaign.game
if (
game not in self.wanted_games # isn't already there
and game.name not in exclude # and isn't excluded
# and isn't excluded by priority_only
# and isn't excluded by list or priority mode
and game.name not in exclude
and (not priority_only or game.name in priority)
# and can be progressed within the next hour
and campaign.can_earn_within(next_hour)
):
# non-excluded games with no priority are placed last, below priority ones
self.wanted_games[game] = priorities.get(game.name, 0)
self.wanted_games.append(game)
full_cleanup = True
self.restart_watching()
self.change_state(State.CHANNELS_CLEANUP)
@@ -734,7 +748,7 @@ class Twitch:
new_channels, key=self._viewers_key, reverse=True
)
ordered_channels.sort(key=lambda ch: ch.acl_based, reverse=True)
ordered_channels.sort(key=self.get_priority, reverse=True)
ordered_channels.sort(key=self.get_priority)
# ensure that we won't end up with more channels than we can handle
# NOTE: we trim from the end because that's where the non-priority,
# offline (or online but low viewers) channels end up
@@ -817,7 +831,7 @@ class Twitch:
# for a switch (including the watching one)
# NOTE: we need to sort the channels every time because one channel
# can end up streaming any game - channels aren't game-tied
for channel in sorted(channels.values(), key=self.get_priority, reverse=True):
for channel in sorted(channels.values(), key=self.get_priority):
if self.can_watch(channel) and self.should_switch(channel):
new_watching = channel
break
@@ -857,6 +871,10 @@ class Twitch:
interval: float = WATCH_INTERVAL.total_seconds()
while True:
channel: Channel = await self.watching_channel.get()
if not channel.online:
# if the channel isn't online anymore, we stop watching it
self.stop_watching()
continue
succeeded: bool = await channel.send_watch()
if not succeeded:
logger.log(CALL, f"Watch requested failed for channel: {channel.name}")
@@ -985,7 +1003,7 @@ class Twitch:
watching_order = self.get_priority(watching_channel)
return (
# this channel's game is higher order than the watching one's
channel_order > watching_order
channel_order < watching_order
or channel_order == watching_order # or the order is the same
# and this channel is ACL-based and the watching channel isn't
and channel.acl_based > watching_channel.acl_based

17
utils.py
View File

@@ -25,8 +25,8 @@ import yarl
from PIL.ImageTk import PhotoImage
from PIL import Image as Image_module
from constants import JsonType, IS_PACKAGED
from exceptions import ExitRequest, ReloadRequest
from constants import IS_PACKAGED, JsonType, PriorityMode
from constants import _resource_path as resource_path # noqa
@@ -165,15 +165,17 @@ def invalidate_cache(instance, *attrnames):
def _serialize(obj: Any) -> Any:
# convert data
d: int | str | float | list[Any] | JsonType
if isinstance(obj, set):
d = list(obj)
elif isinstance(obj, Enum):
d = obj.value
elif isinstance(obj, datetime):
if isinstance(obj, datetime):
if obj.tzinfo is None:
# assume naive objects are UTC
obj = obj.replace(tzinfo=timezone.utc)
d = obj.timestamp()
elif isinstance(obj, set):
d = list(obj)
elif isinstance(obj, Enum):
# NOTE: IntEnum cannot be used, as it will get serialized as a plain integer,
# then loaded back as an integer as well.
d = obj.value
elif isinstance(obj, yarl.URL):
d = str(obj)
else:
@@ -188,8 +190,9 @@ def _serialize(obj: Any) -> Any:
_MISSING = object()
SERIALIZE_ENV: dict[str, Callable[[Any], object]] = {
"set": set,
"datetime": lambda d: datetime.fromtimestamp(d, timezone.utc),
"URL": yarl.URL,
"PriorityMode": PriorityMode,
"datetime": lambda d: datetime.fromtimestamp(d, timezone.utc),
}