Belkadan Software - Source - sicpelago/blob - apworld/sicp/client.py

   1 from __future__ import annotations
   2 import asyncio
   3 import copy
   4 import logging
   5 import os
   6 import re
   7 import sys
   8 import time
   9 import typing
  10 from typing import Any, Optional
  11
  12 import requests
  13 import json
  14 import traceback
  15
  16
  17 import ModuleUpdate
  18 ModuleUpdate.update()
  19
  20 import Utils
  21
  22 if __name__ == "__main__":
  23     Utils.init_logging("SICPelago", exception_logger="Client")
  24
  25 from NetUtils import ClientStatus
  26 from CommonClient import CommonContext, logger, get_base_parser, ClientCommandProcessor, server_loop
  27 from MultiServer import mark_raw
  28
  29 from . import all_items
  30 from .game import game
  31 from .game.scheme import buffer_input, scheme_read, scheme_eval, scheme_format, SchemeError
  32
  33 items_by_id = {v.id(): v for v in all_items.values()}
  34 macguffins = [game.MacGuffin(x) for x in game.MacGuffins]
  35
  36 async def send_locations(ctx: Context, problem: game.Problem):
  37     def id(problem):
  38         (chapter, exercise) = str(problem).split(' ')[0].split('.')
  39         return int(chapter) * 100 + int(exercise)
  40     problem_id = id(problem)
  41     new_locs = await ctx.check_locations([problem_id * 10, problem_id * 10 + 1])
  42     if new_locs and not ctx.watcher_event.is_set():
  43         ctx.should_wait_for_events = True
  44
  45 async def game_loop(ctx: Context):
  46     game.setup()
  47     unsent_checks = []
  48     game.problem_solved_hook = lambda problem: unsent_checks.append(problem) or True
  49     env = game.ThePlayerFrame
  50
  51     game.show_help()
  52
  53     while True:
  54         try:
  55             for problem in unsent_checks:
  56                 await send_locations(ctx, problem)
  57             unsent_checks = []
  58
  59             if ctx.should_wait_for_events:
  60                 await ctx.watcher_event.wait()
  61             if ctx.watcher_event.is_set():
  62                 for msg in ctx.messages:
  63                     print(f"!! {msg}")
  64                 ctx.messages = []
  65                 for net_item in ctx.items_received:
  66                     item = items_by_id.get(net_item.item, None)
  67                     if not item:
  68                         continue
  69                     if item.id() <= len(macguffins):
  70                         next_mg = macguffins[item.id() - 1]
  71                         if not next_mg:
  72                             continue
  73                         macguffins[item.id() - 1] = None
  74                         game.receive_reward(next_mg)
  75                     elif not game.receive_reward(str(item)):
  76                         continue
  77                 ctx.watcher_event.clear()
  78                 ctx.should_wait_for_events = False
  79
  80             if not any(macguffins) and not ctx.finished_game:
  81                 await ctx.send_msgs([{"cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL}])
  82                 ctx.finished_game = True
  83
  84             src = await asyncio.to_thread(buffer_input)
  85             while src.more_on_line:
  86                 expression = scheme_read(src)
  87                 result = scheme_eval(expression, env)
  88                 if result is not None:
  89                     print(scheme_format(result))
  90         except (SchemeError, SyntaxError, ValueError, RuntimeError) as err:
  91             if (isinstance(err, RuntimeError) and
  92                 'maximum recursion depth exceeded' not in err.args[0]):
  93                 raise
  94             print("Error:", err)
  95         except KeyboardInterrupt:  # <Control>-C
  96             print("\nKeyboardInterrupt")
  97         except EOFError:  # <Control>-D, etc.
  98             return
  99
 100 class Context(CommonContext):
 101     game = "SICPelago"
 102     items_handling = 0b111  # full remote
 103     should_wait_for_events = True
 104     messages: list[str]
 105
 106     def __init__(self, *args, **kwargs):
 107         self.messages = []
 108         super().__init__(*args, **kwargs)
 109
 110     async def server_auth(self, password_requested: bool = False):
 111         await self.send_connect()
 112
 113     async def main(args):
 114         parser = get_base_parser(description="SICPelago")
 115         parser.add_argument('--name', default=None, help="Slot Name to connect as.")
 116         parser.add_argument('--debug', help="Show debug logging on stdout.")
 117         args = parser.parse_args(args)
 118
 119         if not args.debug:
 120             root_logger = logging.getLogger()
 121             for handler in root_logger.handlers[:]:
 122                 if isinstance(handler, logging.StreamHandler):
 123                     root_logger.removeHandler(handler)
 124                     handler.close()
 125
 126         ctx = Context(args.connect, args.password)
 127         ctx.auth = args.name
 128         ctx.server_task = asyncio.create_task(server_loop(ctx), name="server loop")
 129
 130         await game_loop(ctx)
 131         await ctx.shutdown()
 132
 133     def on_print(self, args: dict):
 134         super().on_print(args)
 135         self.messages.append(args['text'])
 136         self.watcher_event.set()
 137
 138     def on_print_json(self, args: dict):
 139         super().on_print_json(args)
 140         text = self.jsontotextparser(copy.deepcopy(args["data"]))
 141         if 'Now that you are connected' in text:
 142             return
 143         self.messages.append(text)
 144         self.watcher_event.set()
 145
 146 def run(*args) -> None:
 147     asyncio.run(Context.main(args))
 148
 149 if __name__ == '__main__':
 150     run(*sys.argv[1:])