__init__.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199
  1. import datetime
  2. import json
  3. import os
  4. import ssl
  5. import subprocess
  6. import sys
  7. import traceback
  8. import urllib.parse
  9. import urllib.request
  10. import tooncher.controls
  11. """
  12. official api documentation:
  13. https://github.com/ToontownRewritten/api-doc/blob/master/login.md
  14. https://github.com/ToontownRewritten/api-doc/blob/master/invasions.md
  15. """
  16. INVASIONS_API_URL = 'https://www.toontownrewritten.com/api/invasions?format=json'
  17. LOGIN_API_URL = 'https://www.toontownrewritten.com/api/login?format=json'
  18. if sys.platform == 'darwin':
  19. TOONTOWN_LIBRARY_PATH = os.path.join(
  20. os.path.expanduser('~'), 'Library',
  21. 'Application Support', 'Toontown Rewritten',
  22. )
  23. TOONTOWN_ENGINE_DEFAULT_PATH = os.path.join(
  24. TOONTOWN_LIBRARY_PATH,
  25. 'Toontown Rewritten',
  26. )
  27. else:
  28. TOONTOWN_LIBRARY_PATH = None
  29. TOONTOWN_ENGINE_DEFAULT_PATH = None
  30. def start_engine(engine_path, gameserver, playcookie, **kwargs):
  31. env = {
  32. 'TTR_GAMESERVER': gameserver,
  33. 'TTR_PLAYCOOKIE': playcookie,
  34. }
  35. if sys.platform == 'darwin':
  36. env['DYLD_LIBRARY_PATH'] = os.path.join(
  37. TOONTOWN_LIBRARY_PATH,
  38. 'Libraries.bundle',
  39. )
  40. env['DYLD_FRAMEWORK_PATH'] = os.path.join(
  41. TOONTOWN_LIBRARY_PATH,
  42. 'Frameworks',
  43. )
  44. elif sys.platform == 'linux' and 'XAUTHORITY' in os.environ:
  45. """
  46. Fix for TTREngine reporting:
  47. > :display:x11display(error): Could not open display ":0.0".
  48. > :ToonBase: Default graphics pipe is glxGraphicsPipe (OpenGL).
  49. > :ToonBase(warning): Unable to open 'onscreen' window.
  50. > Traceback (most recent call last):
  51. > File "<compiled '__voltorbmain__'>", line 0, in <module>
  52. > [...]
  53. > File "<compiled 'direct.vlt8f63e471.ShowBase'>", line 0, in vltf05fd21b
  54. > Exception: Could not open window.
  55. """
  56. env['XAUTHORITY'] = os.environ['XAUTHORITY']
  57. return subprocess.Popen(
  58. args=[engine_path],
  59. cwd=os.path.dirname(engine_path),
  60. env=env,
  61. **kwargs,
  62. )
  63. def api_request(url, params=None, validate_ssl_cert=True):
  64. resp = urllib.request.urlopen(
  65. url=url,
  66. data=urllib.parse.urlencode(params).encode('ascii')
  67. if params else None,
  68. context=None if validate_ssl_cert
  69. else ssl._create_unverified_context(),
  70. )
  71. return json.loads(resp.read().decode('ascii'))
  72. class LoginSuccessful:
  73. def __init__(self, playcookie, gameserver):
  74. self.playcookie = playcookie
  75. self.gameserver = gameserver
  76. class LoginDelayed:
  77. def __init__(self, queue_token):
  78. self.queue_token = queue_token
  79. def login(username=None, password=None,
  80. queue_token=None, validate_ssl_cert=True):
  81. if username is not None and queue_token is None:
  82. assert password is not None
  83. req_params = {
  84. 'username': username,
  85. 'password': password,
  86. }
  87. elif username is None and queue_token is not None:
  88. req_params = {
  89. 'queueToken': queue_token,
  90. }
  91. else:
  92. raise Exception('either specify username or queue token')
  93. resp_data = api_request(
  94. url=LOGIN_API_URL,
  95. params=req_params,
  96. validate_ssl_cert=validate_ssl_cert,
  97. )
  98. if resp_data['success'] == 'true':
  99. return LoginSuccessful(
  100. playcookie=resp_data['cookie'],
  101. gameserver=resp_data['gameserver'],
  102. )
  103. elif resp_data['success'] == 'delayed':
  104. return LoginDelayed(
  105. queue_token=resp_data['queueToken'],
  106. )
  107. else:
  108. raise Exception(repr(resp_data))
  109. def launch(engine_path, username, password, validate_ssl_certs=True,
  110. cpu_limit_percent=None, enable_extended_keyboard_controls=False,
  111. extended_keyboard_control_toggle_keysym_name=None):
  112. result = login(
  113. username=username,
  114. password=password,
  115. validate_ssl_cert=validate_ssl_certs,
  116. )
  117. if isinstance(result, LoginDelayed):
  118. result = login(
  119. queue_token=result.queue_token,
  120. validate_ssl_cert=validate_ssl_certs,
  121. )
  122. if isinstance(result, LoginSuccessful):
  123. p = start_engine(
  124. engine_path=engine_path,
  125. gameserver=result.gameserver,
  126. playcookie=result.playcookie,
  127. )
  128. if cpu_limit_percent is not None:
  129. subprocess.Popen(args=[
  130. 'cpulimit',
  131. '--pid', str(p.pid),
  132. '--limit', str(cpu_limit_percent),
  133. # '--verbose',
  134. ])
  135. if enable_extended_keyboard_controls:
  136. try:
  137. tooncher.controls.ExtendedControls(
  138. primary_engine_pid=p.pid,
  139. toggle_keysym_name=extended_keyboard_control_toggle_keysym_name,
  140. ).run()
  141. except Exception as e:
  142. if isinstance(e, KeyboardInterrupt):
  143. raise e
  144. else:
  145. traceback.print_exc()
  146. if p.poll() is None:
  147. p.wait()
  148. else:
  149. raise Exception(repr(result))
  150. class InvasionProgress:
  151. def __init__(self, district, date, cog_type,
  152. despawned_number, total_number):
  153. self.district = district
  154. self.date = date
  155. self.cog_type = cog_type
  156. self.despawned_number = despawned_number
  157. self.total_number = total_number
  158. @property
  159. def remaining_number(self):
  160. return self.total_number - self.despawned_number
  161. def request_active_invasions(validate_ssl_certs=True):
  162. resp_data = api_request(INVASIONS_API_URL)
  163. if resp_data['error'] is not None:
  164. raise Exception(resp_data['error'])
  165. else:
  166. invs = {}
  167. for district, inv_data in resp_data['invasions'].items():
  168. despawned_number, total_number = inv_data['progress'].split('/')
  169. invs[district] = InvasionProgress(
  170. district=district,
  171. date=datetime.datetime.utcfromtimestamp(inv_data['asOf']),
  172. cog_type=inv_data['type'],
  173. despawned_number=int(despawned_number),
  174. total_number=int(total_number),
  175. )
  176. return invs