repl.it
@pyelias/

moargpy client

Python

No description

fork
loading
Files
  • main.py
  • pather.py
  • poetry.lock
  • pyproject.toml
  • requirements.txt
main.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
import json
import contextlib
import asyncio
import re

import websockets

from pather import Pather

URL = "wss://moargpy.eaz.repl.co"
STAT_RE = re.compile(r"\((-?\d+), (-?\d+)\)>"
                      "\((-?\d+), (-?\d+)\)$")

DIR_NAMES = {
  (0, -1): "u",
  (0, 1): "d",
  (-1, 0): "l",
  (1, 0): "r",
}

DIRS = {v: k for k, v in DIR_NAMES.items()}

def get_dir_name(dir_):
  return DIR_NAMES.get(dir_, dir_)

def get_dir(dir_):
  return DIRS.get(dir_, dir_)

class Client():
  @classmethod
  @contextlib.asynccontextmanager
  async def use(cls, auth):
    async with websockets.connect(URL) as ws:
      yield await cls.start(ws, auth)
  
  @classmethod
  async def create(cls, auth):
    ws = websockets.connect(URL)
    return await cls.start(ws, auth)
  
  @classmethod
  async def start(cls, ws, auth):
    inst = cls(ws)
    await inst.send("auth", auth)
    auth, reason = await inst.recv()
    if auth == "authfail":
      raise RuntimeError(f"auth fail: {reason['reason']}")

    asyncio.ensure_future(inst.listen())
    return inst
  
  def __init__(self, ws):
    self.ws = ws
    self.world = None
    self.coords = None
    self.view_evt = asyncio.Event()
  
  async def send(self, evt, data):
    data = {"evt": evt, **data}
    await self.ws.send(json.dumps(data))
  
  async def recv(self):
    res = json.loads(await self.ws.recv())
    return res["evt"], res
  
  async def listen(self):
    while True:
      try:
        evt, data = await self.recv()
      except websockets.exceptions.ConnectionClosedOK:
        return
      
      if evt == "view":
        self.world = [
          row.split(" ")
          for row in data["view"].split("\n")
        ]
        stat = STAT_RE.search(data["stat"])
        if stat is None:
          raise RuntimeError("invalid stat")
        room_x, room_y, x, y = stat.groups()
        self.coords = (
          int(room_x) * 11 + int(x),
          int(room_y) * 11 + int(y),
        )

        old_evt = self.view_evt
        self.view_evt = asyncio.Event()
        old_evt.set()
  
  async def move(self, dir_):
    dir_ = get_dir_name(dir_)
    await self.send("move", {"dir": dir_})
  
  async def path(self, path):
    for dir_ in path:
      await self.move(dir_)
  
  async def req_view(self, w, h):
    await self.send("viewreq", {"dims": [w, h]})
    await self.view_evt.wait()
    return self.world
  
  async def update_pather(self, p, w=3, h=3):
    world = await self.req_view(w, h)
    p.update_map(*self.coords, world)

async def main():
  async with Client.use({"name": "bot-test", "pw": "bot-test"}) as c:
    await c.req_view(0, 0)
    print("starting", c.coords)
    pather = Pather(c.coords)
    await c.update_pather(pather)

    r = pather.run()
    for path in pather.run():
      await c.path(path)
      await c.update_pather(pather)

asyncio.run(main())