-
Notifications
You must be signed in to change notification settings - Fork 1
/
connector.py
157 lines (124 loc) · 6.3 KB
/
connector.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any
from lcu_driver import Connector
from gui.confirmation import ConfirmationBox
from .errors import ConfirmationDenied, CustomException
if TYPE_CHECKING:
from lcu_driver.connection import Connection
log = logging.getLogger(__name__)
# I'm doing it here so all if __main__ == '__main__': also get logging
logging.basicConfig(
format="{asctime} | {levelname:<7} | {funcName:<30} | {message}",
datefmt="%H:%M:%S %d/%m",
style="{",
level=logging.INFO,
)
class AluConnector(Connector):
"""Aluerie's Connector -
Subclass to lcu_driver's Connector aimed to reduce size of code and spam of the same codeblocks.
Subclasses of AluConnector are supposed to implement the following functions:
* coroutine `async def callback(self) -> str:`
which should do the whole work of the supposed script,
i.e. `be_mass_disenchant.py` it should do the mass disenchanting.
Also doc-strings of AluConnector subclasses should be written
in user-friendly way since they are going to be shown to them.
"""
if TYPE_CHECKING:
connection: Connection
def __init__(self, need_confirmation: bool = False):
# Probable shit-code ahead warning
# lcu-driver library has a bug/bad design/oversight where
# on connector.stop() it doesn't properly close the connection/loop
# meaning with just `super().__init__()` it won't work on multiple GUI button presses
# because the infinite loop from the first button connector is left behind waiting for a web sockets to proc
# so one possible solution to this is to register a new loop on each GUI button press
# it leaves previous infinitely sleeping event loop untouched though...
# if anybody knows better - hit me up, I beg you.
# links:
# https://github.com/sousa-andre/lcu-driver/issues/18
# https://github.com/sousa-andre/lcu-driver/pull/34
# todo: open issue in HextechButEfficient repository
new_loop = asyncio.new_event_loop()
super().__init__(loop=new_loop)
self.console_text: str = "No result yet"
self.need_confirmation: bool = need_confirmation
# let's continue.
self._set_event("ready", self.connect)
self._set_event("disconnect", self.disconnect)
async def connect(self, _: Connection):
log.info("LCU API is connected")
summoner = await self.get("/lol-summoner/v1/current-summoner")
if summoner.status != 200:
# silly check if league is not down
log.warning("Please login into your account and restart the script...")
else:
try:
self.console_text = await self.callback()
log.info(self.console_text)
except Exception as exc:
if isinstance(exc, CustomException):
self.console_text = str(exc)
log.info(exc)
else:
self.console_text = (
f"Failed with exception. Contact developers about it:" f"\n{exc.__class__.__name__}: {str(exc)}"
)
log.error("%s: %s", exc.__class__.__name__, exc, exc_info=True)
async def disconnect(self, _: Connection):
log.info("Finished task. The LCU API client have been closed!")
async def callback(self: AluConnector) -> str:
"""This function will be called on @ready event
It is supposed to be implemented by subclasses and do the script job.
* For streamlined UI experience callbacks should at some point
call self.confirm or self.output to showcase their results.
* And it also should return a small descriptive string to print in the GUI console.
"""
raise NotImplementedError("You need to implement `async def callback(self):` in AluConnector subclasses.")
def confirm(self, script_message: str) -> bool:
"""Bring Confirmation dialog that describes what the script gonna do
This function is made to used in callback body to request confirming loot-sensitive scripts like disenchant.
It can be defaulted to skip confirmation if self.need_confirmation is False
"""
log.info(script_message)
if not self.need_confirmation:
# no confirmation is needed
return True
confirm = ConfirmationBox(script_message).get()
if not confirm:
# user pressed No, closed the window or etc.
raise ConfirmationDenied("Confirmation was not received. Not executing.")
else:
# user pressed Yes
return True
def output(self, script_message: str):
"""Bring OK-confirmation dialog that shows the script output to user.
If the script is not loot-sensitive then we can just show the result of the script.
"""
log.info(script_message)
if not self.need_confirmation:
# no confirmation is needed
return
ConfirmationBox(script_message, option_no=False)
# SHORTCUTS
async def delete(self, endpoint: str, **kwargs: Any):
"""Shortcut: perform DELETE request against LCU API."""
log.debug("DELETE %s %s", endpoint, kwargs or "")
return await self.connection.request("delete", endpoint, **kwargs)
async def get(self, endpoint: str, **kwargs: Any):
"""Shortcut: perform GET request against LCU API."""
log.debug("GET %s %s", endpoint, kwargs or "")
return await self.connection.request("get", endpoint, **kwargs)
async def patch(self, endpoint: str, **kwargs: Any):
"""Shortcut: perform PATCH request against LCU API."""
log.debug("PATCH %s %s", endpoint, kwargs or "")
return await self.connection.request("patch", endpoint, **kwargs)
async def post(self, endpoint: str, **kwargs: Any):
"""Shortcut: perform POST request against LCU API."""
log.debug("POST %s %s", endpoint, kwargs or "")
return await self.connection.request("post", endpoint, **kwargs)
async def put(self, endpoint: str, **kwargs: Any):
"""Shortcut: perform PUT request against LCU API."""
log.debug("PUT %s %s", endpoint, kwargs or "")
return await self.connection.request("put", endpoint, **kwargs)