-
Notifications
You must be signed in to change notification settings - Fork 0
/
roguelike-client
executable file
·187 lines (169 loc) · 6.57 KB
/
roguelike-client
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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
#!/usr/bin/python3
# This file is part of PlomRogue. PlomRogue is licensed under the GPL version 3
# or any later version. For details on its copyright, license, and warranties,
# see the file NOTICE in the root directory of the PlomRogue source package.
import curses
import os
import signal
import time
from client.config.world_data import world_data
from client.config.io import io
from client.config.commands import commands
from client.window_management import redraw_windows, set_windows, draw_screen, \
stdscr
from client.query_mapcell import query_mapcell
message_queue = {
"open_end": False,
"messages": []
}
def read_worldstate():
global redraw_windows
if not os.access(io["path_worldstate"], os.F_OK):
msg = "No world state file found at " + io["path_worldstate"] + "."
raise SystemExit(msg)
read_anew = False
worldstate_file = open(io["path_worldstate"], "r")
turn_string = worldstate_file.readline()
if int(turn_string) != world_data["turn"]:
read_anew = True
if not read_anew: # In rare cases, world may change, but not turn number.
mtime = os.stat(io["path_worldstate"])
if mtime != read_worldstate.last_checked_mtime:
read_worldstate.last_checked_mtime = mtime
read_anew = True
if read_anew:
# TODO: Hardcode order of necessary fields, ensure order dependencies.
redraw_windows = True
old_inventory_size = len(world_data["inventory"])
world_data["turn"] = int(turn_string)
for entry in io["worldstate_read_order"]:
if entry[1] == "int":
if 2 == len(entry):
world_data[entry[0]] = int(worldstate_file.readline())
elif 3 == len(entry):
world_data[entry[0]][entry[2]] = \
int(worldstate_file.readline())
elif entry[1] == "lines":
world_data[entry[0]] = []
while True:
line = worldstate_file.readline().replace("\n", "")
if line == '%':
break
world_data[entry[0]] += [line]
elif entry[1] == "map":
world_data[entry[0]] = ""
for i in range(world_data["map_size"]):
line = worldstate_file.readline().replace("\n", "")
world_data[entry[0]] += line
if not world_data["look_mode"]:
world_data["map_center"] = world_data["avatar_position"][:]
if world_data["inventory_selection"] > 0 and \
len(world_data["inventory"]) < old_inventory_size:
world_data["inventory_selection"] -= 1
worldstate_file.close()
read_worldstate.last_checked_mtime = -1
def read_message_queue():
global redraw_windows
while (len(message_queue["messages"]) > 1
or (len(message_queue["messages"]) == 1
and not message_queue["open_end"])):
message = message_queue["messages"].pop(0)
if message == "THINGS_HERE START":
read_message_queue.parse_thingshere = True
world_data["look"] = []
elif message == "THINGS_HERE END":
read_message_queue.parse_thingshere = False
if world_data["look"] == []:
world_data["look"] = ["(none known)"]
redraw_windows = True
elif read_message_queue.parse_thingshere:
world_data["look"] += [message]
elif message[0:4] == "LOG ":
world_data["log"] += [message[4:]]
redraw_windows = True
elif message == "WORLD_UPDATED":
query_mapcell()
elif message[:6] == "PLUGIN":
str_plugin = message[7:]
if (str_plugin.replace("_", "").isalnum()
and os.access("plugins/client/" + str_plugin + ".py",
os.F_OK)):
exec(open("plugins/client/" + str_plugin + ".py").read())
return
raise SystemExit("Invalid plugin load path in message: " + message)
read_message_queue.parse_thingshere = False
def cursed_main(stdscr):
global redraw_windows
def ping_test():
half_wait_time = 5
if len(new_data_from_server) > 0:
ping_test.sent = False
elif ping_test.wait_start + half_wait_time < time.time():
if not ping_test.sent:
io["file_out"].write("PING\n")
io["file_out"].flush()
ping_test.sent = True
ping_test.wait_start = time.time()
elif ping_test.sent:
raise SystemExit("Server not answering anymore.")
ping_test.wait_start = 0
def read_into_message_queue():
if new_data_from_server == "":
return
new_open_end = False
if new_data_from_server[-1] is not "\n":
new_open_end = True
new_messages = new_data_from_server.splitlines()
if message_queue["open_end"]:
message_queue["messages"][-1] += new_messages[0]
del new_messages[0]
message_queue["messages"] += new_messages
if new_open_end:
message_queue["open_end"] = True
def set_and_redraw_windows(*ignore):
set_windows()
draw_screen()
curses.noecho()
curses.curs_set(False)
signal.signal(signal.SIGWINCH, set_and_redraw_windows)
set_windows()
delay = 1
while True:
stdscr.timeout(int(delay))
if delay < 1000:
delay = delay * 1.1
if redraw_windows:
delay = 1
draw_screen()
redraw_windows = False
char = stdscr.getch()
if char >= 0:
char = chr(char)
if char in commands:
if len(commands[char]) == 1 or not world_data["look_mode"]:
commands[char][0]()
else:
commands[char][1]()
redraw_windows = True
new_data_from_server = io["file_in"].read()
ping_test()
read_into_message_queue()
read_worldstate()
read_message_queue()
try:
if (not os.access(io["path_out"], os.F_OK)):
msg = "No server input file found at " + io["path_out"] + "."
raise SystemExit(msg)
io["file_out"] = open(io["path_out"], "a")
io["file_in"] = open(io["path_in"], "r")
curses.wrapper(cursed_main)
except SystemExit as exit:
print("ABORTING: " + exit.args[0])
except:
print("SOMETHING WENT WRONG IN UNEXPECTED WAYS")
raise
finally:
if "file_out" in io:
io["file_out"].close()
if "file_in" in io:
io["file_in"].close()