yanwah/tennoui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace requests with aiohttp and make warpy asynchronous

Browse Source 
This is a major change in how this overlay will be defined as it will now be asynchronous. After doing
a bit of research this is likely to be the best option since there will be various functions polling
the API for information. If one requests takes to long the GUI will get hung up, this resolves that
issue.

Note: This is a first dive into asynchronous so future commits may aim to resolve sloppy asynchronous
implementations.
This commit is contained in:
雲華
2021-05-02 00:36:42 -04:00
parent 5a92803bbe
commit c3a5f10b82
3 changed files with 246 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
.gitignore vendored Normal file
View File

@@ -0,0 +1,141 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
.idea/

18
src/overlay.py
View File

@@ -1,9 +1,10 @@
import asyncio
import aiohttp
import tkinter as tk
import warframe_api as warpy
import win32gui
import win32api
import win32con
import requests
from ctypes import windll
GWL_EXSTYLE = -20
@@ -86,7 +87,9 @@ else:
timers = tk.Frame(master=root, bg="#FFFFFF", width=340, height=160, cursor="none")
warframe = warpy.WarframeAPI("pc", requests.session())
loop = asyncio.get_event_loop()
warframe = warpy.WarframeAPI("pc", loop=loop)
timers.grid(row=0, column=0)
@@ -102,10 +105,11 @@ cambion_timer.grid(row=0, column=3)
vallis_timer.grid(row=0, column=5)
def current_cycles():
cetus_status = warframe.cetus_status()
vallis_status = warframe.vallis_status()
cambion_status = warframe.cambion_status()
async def current_cycles():
print("Running...")
cetus_status = await warframe.cetus_status()
vallis_status = await warframe.vallis_status()
cambion_status = await warframe.cambion_status()
if cetus_status["isDay"]:
cetus_timer.config(text="Day")
else:
@@ -118,6 +122,6 @@ def current_cycles():
root.after(300000, current_cycles)
current_cycles()
loop.run_until_complete(current_cycles())
root.mainloop()

211
src/warframe_api.py
View File

@@ -1,160 +1,137 @@
from requests import request
import asyncio
import aiohttp
WARFRAME_API = 'https://api.warframestat.us'
class NonPlatformError(Exception):
pass
def catch_status_code(f):
def func(*args, **kwargs):
response = f(*args, **kwargs)
if response.status_code != 200:
response.raise_for_status()
return response.json()
return func
class WarframeAPI:
_platforms = ['pc', 'ps4', 'xb1', 'swi']
def __init__(self, platform: str, session: request, language: str = 'en'):
def __init__(self, platform: str, language: str = 'en', loop=None):
if platform not in self._platforms:
raise NonPlatformError(platform)
self.platform = platform
self.language = language
self.api = 'https://api.warframestat.us/{platform}'.format(platform=self.platform)
self.session = session
self._loop = asyncio.get_event_loop() if loop is None else loop
self.session = aiohttp.ClientSession(loop=self._loop)
@catch_status_code
def worldstate(self):
response = self.session.get(self.api)
return response
async def _fetch(self, url):
async with self.session.get(url) as response:
resp = await response.json()
return resp
@catch_status_code
def alerts(self):
response = self.session.get(self.api + "/alerts")
return response
@catch_status_code
def arbitration(self):
response = self.session.get(self.api + "/arbitration")
return response
async def worldstate(self):
url = WARFRAME_API + "/{platform}".format(platform=self.platform)
return await self._fetch(url)
# response = self.session.get(self.api)
# return response
@catch_status_code
def cambion_status(self):
response = self.session.get(self.api + "/cambionCycle")
return response
async def alerts(self):
url = WARFRAME_API + "/{platform}/alerts".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def cetus_status(self):
response = self.session.get(self.api + "/cetusCycle")
return response
async def arbitration(self):
url = WARFRAME_API + "/{platform}/arbitration".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def conclave_challenges(self):
response = self.session.get(self.api + "/cetusCycle")
return response
async def cambion_status(self):
url = WARFRAME_API + "/{platform}/cambionCycle".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def construction_progress(self):
response = self.session.get(self.api + "/constructionProgress")
return response
async def cetus_status(self):
url = WARFRAME_API + "/{platform}/cetusCycle".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def darvo_deal(self):
response = self.session.get(self.api + "/dailyDeals")
return response
async def conclave_challenges(self, query: str = None):
url = WARFRAME_API + "/{platform}/conclaveChallenges".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def earth_cycle(self):
response = self.session.get(self.api + "/earthCycle")
return response
async def construction_progress(self):
url = WARFRAME_API + "/{platform}/constructionProgress".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def ongoing_events(self):
response = self.session.get(self.api + "/events")
return response
async def darvo_deal(self):
url = WARFRAME_API + "/{platform}/dailyDeals".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def fissures(self):
response = self.session.get(self.api + "/fissures")
return response
async def earth_cycle(self):
url = WARFRAME_API + "/{platform}/earthCycle".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def darvo_flash_sale(self):
response = self.session.get(self.api + "/flashSales")
return response
async def ongoing_events(self):
url = WARFRAME_API + "/{platform}/events".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def global_upgrades(self):
response = self.session.get(self.api + "/globalUpgrades")
return response
async def fissures(self):
url = WARFRAME_API + "/{platform}/fissures".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def invasions(self):
response = self.session.get(self.api + "/invasions")
return response
@catch_status_code
def kuva_nodes(self):
response = self.session.get(self.api + "/kuva")
return response
async def darvo_flash_sale(self):
url = WARFRAME_API + "/{platform}/flashSales".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def news(self):
response = self.session.get(self.api + "/news")
return response
async def global_upgrades(self):
url = WARFRAME_API + "/{platform}/globalUpgrades".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def nightwave(self):
response = self.session.get(self.api + "/nightwave")
return response
async def invasions(self):
url = WARFRAME_API + "/{platform}/invasions".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def persistent_enemy_data(self):
response = self.session.get(self.api + "/persistentEnemies")
return response
async def kuva_nodes(self):
url = WARFRAME_API + "/{platform}/kuva".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def riven_stats(self, query: str = None):
async def news(self):
url = WARFRAME_API + "/{platform}/news".format(platform=self.platform)
return await self._fetch(url)
async def nightwave(self):
url = WARFRAME_API + "/{platform}/nightwave".format(platform=self.platform)
return await self._fetch(url)
async def persistent_enemy_data(self):
url = WARFRAME_API + "/{platform}/persistentEnemies".format(platform=self.platform)
return await self._fetch(url)
async def riven_stats(self, query: str = None):
if query:
response = self.session.get(self.api + "/rivens/search/{query}".format(query=query))
url = WARFRAME_API + "/{platform}/rivens/search/{query}".format(platform=self.platform, query=query)
else:
response = self.session.get(self.api + "/rivens")
return response
url = WARFRAME_API + "/{platform}/rivens".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def sentient_outpost(self):
response = self.session.get(self.api + "/sentientOutposts")
return response
async def sentient_outpost(self):
url = WARFRAME_API + "/{platform}/sentientOutposts".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def sanctuary_status(self):
response = self.session.get(self.api + "/simaris")
return response
async def sanctuary_status(self):
url = WARFRAME_API + "/{platform}/simaris".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def sortie(self):
response = self.session.get(self.api + "/sortie")
return response
async def sortie(self):
url = WARFRAME_API + "/{platform}/sortie".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def syndicate_nodes(self):
response = self.session.get(self.api + "/syndicateMissions")
return response
async def syndicate_nodes(self):
url = WARFRAME_API + "/{platform}/syndicateMissions".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def worldstate_timestamp(self):
response = self.session.get(self.api + "/timestamp")
return response
async def worldstate_timestamp(self):
url = WARFRAME_API + "/{platform}/timestamp".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def vallis_status(self):
response = self.session.get(self.api + "/vallisCycle")
return response
async def vallis_status(self):
url = WARFRAME_API + "/{platform}/vallisCycle".format(platform=self.platform)
return await self._fetch(url)
@catch_status_code
def void_trader(self):
response = self.session.get(self.api + "/voidTrader")
return response
async def void_trader(self):
url = WARFRAME_API + "/{platform}/voidTrader".format(platform=self.platform)
return await self._fetch(url)