|
|
@@ -1,64 +1,53 @@
|
|
|
+import copy
|
|
|
import datetime
|
|
|
import json
|
|
|
import os
|
|
|
+import pathlib
|
|
|
import ssl
|
|
|
import subprocess
|
|
|
import sys
|
|
|
import traceback
|
|
|
+import typing
|
|
|
import urllib.parse
|
|
|
import urllib.request
|
|
|
-import tooncher.controls
|
|
|
-
|
|
|
-"""
|
|
|
-official api documentation:
|
|
|
-https://github.com/ToontownRewritten/api-doc/blob/master/login.md
|
|
|
-https://github.com/ToontownRewritten/api-doc/blob/master/invasions.md
|
|
|
-"""
|
|
|
-
|
|
|
-INVASIONS_API_URL = "https://www.toontownrewritten.com/api/invasions?format=json"
|
|
|
-LOGIN_API_URL = "https://www.toontownrewritten.com/api/login?format=json"
|
|
|
-
|
|
|
-if sys.platform == "darwin":
|
|
|
- TOONTOWN_LIBRARY_PATH = os.path.join(
|
|
|
- os.path.expanduser("~"), "Library", "Application Support", "Toontown Rewritten",
|
|
|
- )
|
|
|
- TOONTOWN_ENGINE_DEFAULT_PATH = os.path.join(
|
|
|
- TOONTOWN_LIBRARY_PATH, "Toontown Rewritten",
|
|
|
- )
|
|
|
-else:
|
|
|
- TOONTOWN_LIBRARY_PATH = None
|
|
|
- TOONTOWN_ENGINE_DEFAULT_PATH = None
|
|
|
|
|
|
+import tooncher.controls
|
|
|
|
|
|
-def start_engine(engine_path, gameserver, playcookie, **kwargs):
|
|
|
- env = {
|
|
|
- "TTR_GAMESERVER": gameserver,
|
|
|
- "TTR_PLAYCOOKIE": playcookie,
|
|
|
- }
|
|
|
+# official api documentation:
|
|
|
+# https://github.com/ToontownRewritten/api-doc/blob/master/login.md
|
|
|
+# https://github.com/ToontownRewritten/api-doc/blob/master/invasions.md
|
|
|
+
|
|
|
+_LOGIN_API_URL = "https://www.toontownrewritten.com/api/login?format=json"
|
|
|
+
|
|
|
+
|
|
|
+def start_engine(
|
|
|
+ engine_path: pathlib.Path, gameserver: str, playcookie: str, **popen_kwargs
|
|
|
+) -> subprocess.Popen:
|
|
|
+ # without XAUTHORITY:
|
|
|
+ # > :display:x11display(error): Could not open display ":0.0".
|
|
|
+ # > :ToonBase: Default graphics pipe is glxGraphicsPipe (OpenGL).
|
|
|
+ # > :ToonBase(warning): Unable to open 'onscreen' window.
|
|
|
+ # > Traceback (most recent call last):
|
|
|
+ # > File "<compiled '__voltorbmain__'>", line 0, in <module>
|
|
|
+ # > [...]
|
|
|
+ # > File "<compiled 'direct.vlt8f63e471.ShowBase'>", line 0, in vltf05fd21b
|
|
|
+ # > Exception: Could not open window.
|
|
|
+ # optirun sets plenty of env vars
|
|
|
+ env = copy.copy(os.environ)
|
|
|
+ env["TTR_GAMESERVER"] = gameserver
|
|
|
+ env["TTR_PLAYCOOKIE"] = playcookie
|
|
|
+ engine_path = engine_path.resolve()
|
|
|
if sys.platform == "darwin":
|
|
|
- env["DYLD_LIBRARY_PATH"] = os.path.join(
|
|
|
- TOONTOWN_LIBRARY_PATH, "Libraries.bundle",
|
|
|
- )
|
|
|
- env["DYLD_FRAMEWORK_PATH"] = os.path.join(TOONTOWN_LIBRARY_PATH, "Frameworks",)
|
|
|
- elif sys.platform == "linux" and "XAUTHORITY" in os.environ:
|
|
|
- """
|
|
|
- Fix for TTREngine reporting:
|
|
|
- > :display:x11display(error): Could not open display ":0.0".
|
|
|
- > :ToonBase: Default graphics pipe is glxGraphicsPipe (OpenGL).
|
|
|
- > :ToonBase(warning): Unable to open 'onscreen' window.
|
|
|
- > Traceback (most recent call last):
|
|
|
- > File "<compiled '__voltorbmain__'>", line 0, in <module>
|
|
|
- > [...]
|
|
|
- > File "<compiled 'direct.vlt8f63e471.ShowBase'>", line 0, in vltf05fd21b
|
|
|
- > Exception: Could not open window.
|
|
|
- """
|
|
|
- env["XAUTHORITY"] = os.environ["XAUTHORITY"]
|
|
|
+ env["DYLD_LIBRARY_PATH"] = str(engine_path.parent.joinpath("Libraries.bundle"))
|
|
|
+ env["DYLD_FRAMEWORK_PATH"] = str(engine_path.parent.joinpath("Frameworks"))
|
|
|
return subprocess.Popen(
|
|
|
- args=[engine_path], cwd=os.path.dirname(engine_path), env=env, **kwargs,
|
|
|
+ args=[str(engine_path)], cwd=engine_path.parent, env=env, **popen_kwargs,
|
|
|
)
|
|
|
|
|
|
|
|
|
-def api_request(url, params=None, validate_ssl_cert=True):
|
|
|
+def _api_request(
|
|
|
+ url: str, params: typing.Optional[dict] = None, validate_ssl_cert: bool = True
|
|
|
+):
|
|
|
resp = urllib.request.urlopen(
|
|
|
url=url,
|
|
|
data=urllib.parse.urlencode(params).encode("ascii") if params else None,
|
|
|
@@ -67,18 +56,23 @@ def api_request(url, params=None, validate_ssl_cert=True):
|
|
|
return json.loads(resp.read().decode("ascii"))
|
|
|
|
|
|
|
|
|
-class LoginSuccessful:
|
|
|
- def __init__(self, playcookie, gameserver):
|
|
|
+class _LoginSuccessful:
|
|
|
+ def __init__(self, playcookie: str, gameserver: str):
|
|
|
self.playcookie = playcookie
|
|
|
self.gameserver = gameserver
|
|
|
|
|
|
|
|
|
-class LoginDelayed:
|
|
|
- def __init__(self, queue_token):
|
|
|
+class _LoginDelayed:
|
|
|
+ def __init__(self, queue_token: str):
|
|
|
self.queue_token = queue_token
|
|
|
|
|
|
|
|
|
-def login(username=None, password=None, queue_token=None, validate_ssl_cert=True):
|
|
|
+def _login(
|
|
|
+ username: typing.Optional[str] = None,
|
|
|
+ password: typing.Optional[str] = None,
|
|
|
+ queue_token: typing.Optional[str] = None,
|
|
|
+ validate_ssl_cert: bool = True,
|
|
|
+) -> typing.Union[_LoginSuccessful, _LoginDelayed]:
|
|
|
if username is not None and queue_token is None:
|
|
|
assert password is not None
|
|
|
req_params = {
|
|
|
@@ -91,99 +85,66 @@ def login(username=None, password=None, queue_token=None, validate_ssl_cert=True
|
|
|
}
|
|
|
else:
|
|
|
raise Exception("either specify username or queue token")
|
|
|
- resp_data = api_request(
|
|
|
- url=LOGIN_API_URL, params=req_params, validate_ssl_cert=validate_ssl_cert,
|
|
|
+ resp_data = _api_request(
|
|
|
+ url=_LOGIN_API_URL, params=req_params, validate_ssl_cert=validate_ssl_cert,
|
|
|
)
|
|
|
if resp_data["success"] == "true":
|
|
|
- return LoginSuccessful(
|
|
|
+ return _LoginSuccessful(
|
|
|
playcookie=resp_data["cookie"], gameserver=resp_data["gameserver"],
|
|
|
)
|
|
|
- elif resp_data["success"] == "delayed":
|
|
|
- return LoginDelayed(queue_token=resp_data["queueToken"],)
|
|
|
- else:
|
|
|
- raise Exception(repr(resp_data))
|
|
|
+ if resp_data["success"] == "delayed":
|
|
|
+ return _LoginDelayed(queue_token=resp_data["queueToken"],)
|
|
|
+ raise Exception(repr(resp_data))
|
|
|
|
|
|
|
|
|
def launch(
|
|
|
- engine_path,
|
|
|
- username,
|
|
|
- password,
|
|
|
- validate_ssl_certs=True,
|
|
|
- cpu_limit_percent=None,
|
|
|
+ engine_path: pathlib.Path,
|
|
|
+ username: str,
|
|
|
+ password: str,
|
|
|
+ validate_ssl_certs: bool = True,
|
|
|
+ cpu_limit_percent: typing.Optional[int] = None,
|
|
|
enable_extended_keyboard_controls=False,
|
|
|
extended_keyboard_control_toggle_keysym_name=None,
|
|
|
engine_window_name=None,
|
|
|
-):
|
|
|
+) -> None:
|
|
|
if engine_window_name and not enable_extended_keyboard_controls:
|
|
|
raise Exception("Enable Extended Controls to change engine's window name",)
|
|
|
- result = login(
|
|
|
+ result = _login(
|
|
|
username=username, password=password, validate_ssl_cert=validate_ssl_certs,
|
|
|
)
|
|
|
- if isinstance(result, LoginDelayed):
|
|
|
- result = login(
|
|
|
+ if isinstance(result, _LoginDelayed):
|
|
|
+ result = _login(
|
|
|
queue_token=result.queue_token, validate_ssl_cert=validate_ssl_certs,
|
|
|
)
|
|
|
- if isinstance(result, LoginSuccessful):
|
|
|
- p = start_engine(
|
|
|
- engine_path=engine_path,
|
|
|
- gameserver=result.gameserver,
|
|
|
- playcookie=result.playcookie,
|
|
|
+ if not isinstance(result, _LoginSuccessful):
|
|
|
+ raise Exception("unexpected response: {!r}".format(result))
|
|
|
+ process = start_engine(
|
|
|
+ engine_path=engine_path,
|
|
|
+ gameserver=result.gameserver,
|
|
|
+ playcookie=result.playcookie,
|
|
|
+ )
|
|
|
+ if cpu_limit_percent is not None:
|
|
|
+ subprocess.Popen(
|
|
|
+ args=[
|
|
|
+ "cpulimit",
|
|
|
+ "--pid",
|
|
|
+ str(process.pid),
|
|
|
+ "--limit",
|
|
|
+ str(cpu_limit_percent),
|
|
|
+ # '--verbose',
|
|
|
+ ]
|
|
|
)
|
|
|
- if cpu_limit_percent is not None:
|
|
|
- subprocess.Popen(
|
|
|
- args=[
|
|
|
- "cpulimit",
|
|
|
- "--pid",
|
|
|
- str(p.pid),
|
|
|
- "--limit",
|
|
|
- str(cpu_limit_percent),
|
|
|
- # '--verbose',
|
|
|
- ]
|
|
|
- )
|
|
|
- if enable_extended_keyboard_controls:
|
|
|
- try:
|
|
|
- tooncher.controls.ExtendedControls(
|
|
|
- primary_engine_pid=p.pid,
|
|
|
- primary_engine_window_name=engine_window_name,
|
|
|
- toggle_keysym_name=extended_keyboard_control_toggle_keysym_name,
|
|
|
- ).run()
|
|
|
- except Exception as e:
|
|
|
- if isinstance(e, KeyboardInterrupt):
|
|
|
- raise e
|
|
|
- else:
|
|
|
- traceback.print_exc()
|
|
|
- if p.poll() is None:
|
|
|
- p.wait()
|
|
|
- else:
|
|
|
- raise Exception(repr(result))
|
|
|
-
|
|
|
-
|
|
|
-class InvasionProgress:
|
|
|
- def __init__(self, district, date, cog_type, despawned_number, total_number):
|
|
|
- self.district = district
|
|
|
- self.date = date
|
|
|
- self.cog_type = cog_type
|
|
|
- self.despawned_number = despawned_number
|
|
|
- self.total_number = total_number
|
|
|
-
|
|
|
- @property
|
|
|
- def remaining_number(self):
|
|
|
- return self.total_number - self.despawned_number
|
|
|
-
|
|
|
-
|
|
|
-def request_active_invasions(validate_ssl_certs=True):
|
|
|
- resp_data = api_request(INVASIONS_API_URL)
|
|
|
- if resp_data["error"] is not None:
|
|
|
- raise Exception(resp_data["error"])
|
|
|
- else:
|
|
|
- invs = {}
|
|
|
- for district, inv_data in resp_data["invasions"].items():
|
|
|
- despawned_number, total_number = inv_data["progress"].split("/")
|
|
|
- invs[district] = InvasionProgress(
|
|
|
- district=district,
|
|
|
- date=datetime.datetime.utcfromtimestamp(inv_data["asOf"]),
|
|
|
- cog_type=inv_data["type"],
|
|
|
- despawned_number=int(despawned_number),
|
|
|
- total_number=int(total_number),
|
|
|
- )
|
|
|
- return invs
|
|
|
+ if enable_extended_keyboard_controls:
|
|
|
+ try:
|
|
|
+ tooncher.controls.ExtendedControls(
|
|
|
+ primary_engine_pid=process.pid,
|
|
|
+ primary_engine_window_name=engine_window_name,
|
|
|
+ toggle_keysym_name=extended_keyboard_control_toggle_keysym_name,
|
|
|
+ ).run()
|
|
|
+ except Exception as e:
|
|
|
+ if isinstance(e, KeyboardInterrupt):
|
|
|
+ raise e
|
|
|
+ else:
|
|
|
+ traceback.print_exc()
|
|
|
+ if process.poll() is None:
|
|
|
+ process.wait()
|
