Coverage for MC6809/core/cpu_control

1#!/usr/bin/env python

3"""

4 DragonPy - CPU control http server

5 ==================================

7 TODO: Use bottle!

9 :copyleft: 2013-2014 by the MC6809 team, see AUTHORS for more details.

10 :license: GNU GPL v3 or above, see LICENSE for more details.

12 Based on:

13 * ApplyPy by James Tauber (MIT license)

14 * XRoar emulator by Ciaran Anscomb (GPL license)

15 more info, see README

16"""

19import json

20import logging

21import re

22import select

23import threading

24import traceback

25from http.server import BaseHTTPRequestHandler, HTTPServer

28log = logging.getLogger("MC6809")

31class ControlHandler(BaseHTTPRequestHandler):

33 def __init__(self, request, client_address, server, cpu):

34 log.error("ControlHandler %s %s %s", request, client_address, server)

35 self.cpu = cpu

37 self.get_urls = {

38 r"/disassemble/(\s+)/$": self.get_disassemble,

39 r"/memory/(\s+)(-(\s+))?/$": self.get_memory,

40 r"/memory/(\s+)(-(\s+))?/raw/$": self.get_memory_raw,

41 r"/status/$": self.get_status,

42 r"/$": self.get_index,

43 }

45 self.post_urls = {

46 r"/memory/(\s+)(-(\s+))?/$": self.post_memory,

47 r"/memory/(\s+)(-(\s+))?/raw/$": self.post_memory_raw,

48 r"/quit/$": self.post_quit,

49 r"/reset/$": self.post_reset,

50 r"/debug/$": self.post_debug,

51 }

53 BaseHTTPRequestHandler.__init__(self, request, client_address, server)

55 def log_message(self, format, *args):

56 msg = f"{self.client_address[0]} - - [{self.log_date_time_string()}] {format % args}\n"

57 log.critical(msg)

59 def dispatch(self, urls):

60 for r, f in list(urls.items()):

61 m = re.match(r, self.path)

62 if m is not None:

63 log.critical("call %s", f.__name__)

64 try:

65 f(m)

66 except Exception as err:

67 txt = traceback.format_exc()

68 self.response_500(f"Error call {f.__name__!r}: {err}", txt)

69 return

70 else:

71 self.response_404(f"url {self.path!r} doesn't match any urls")

73 def response(self, s, status_code=200):

74 log.critical("send %s response", status_code)

75 self.send_response(status_code)

76 self.send_header("Content-Length", str(len(s)))

77 self.end_headers()

78 self.wfile.write(s)

80 def response_html(self, headline, text=""):

81 html = (

82 "<!DOCTYPE html><html><body>"

83 "<h1>%s</h1>"

84 "%s"

85 "</body></html>"

86 ) % (headline, text)

87 self.response(html)

89 def response_404(self, txt):

90 log.error(txt)

91 html = (

92 "<!DOCTYPE html><html><body>"

93 "<h1>DragonPy - 6809 CPU control server</h1>"

94 "<h2>404 - Error:</h2>"

95 "<p>%s</p>"

96 "</body></html>"

97 ) % txt

98 self.response(html, status_code=404)

100 def response_500(self, err, tb_txt):

101 log.error(err, tb_txt)

102 html = (

103 "<!DOCTYPE html><html><body>"

104 "<h1>DragonPy - 6809 CPU control server</h1>"

105 "<h2>500 - Error:</h2>"

106 "<p>%s</p>"

107 "<pre>%s</pre>"

108 "</body></html>"

109 ) % (err, tb_txt)

110 self.response(html, status_code=500)

111

112 def do_GET(self):

113 log.critical("do_GET(): %r", self.path)

114 self.dispatch(self.get_urls)

115

116 def do_POST(self):

117 log.critical("do_POST(): %r", self.path)

118 self.dispatch(self.post_urls)

119

120 def get_index(self, m):

121 self.response_html(

122 headline="DragonPy - 6809 CPU control server",

123 text=(

124 "<p>Example urls:"

125 "<ul>"

126 '<li>CPU status:<a href="/status/">/status/</a></li>'

127 '<li>6809 interrupt vectors memory dump:'

128 '<a href="/memory/fff0-ffff/">/memory/fff0-ffff/</a></li>'

129 '</ul>'

130 '<form action="/quit/" method="post">'

131 '<input type="submit" value="Quit CPU">'

132 '</form>'

133 ))

134

135 def get_disassemble(self, m):

136 addr = int(m.group(1))

137 r = []

138 n = 20

139 while n > 0:

140 dis, length = self.disassemble.disasm(addr)

141 r.append(dis)

142 addr += length

143 n -= 1

144 self.response(json.dumps(r))

145

146 def get_memory_raw(self, m):

147 addr = int(m.group(1))

148 e = m.group(3)

149 if e is not None:

150 end = int(e)

151 else:

152 end = addr

153 self.response("".join(chr(self.cpu.read_byte(x)) for x in range(addr, end + 1)))

154

155 def get_memory(self, m):

156 addr = int(m.group(1), 16)

157 e = m.group(3)

158 if e is not None:

159 end = int(e, 16)

160 else:

161 end = addr

162 self.response(json.dumps(list(map(self.cpu.read_byte, list(range(addr, end + 1))))))

163

164 def get_status(self, m):

165 data = {

166 "cpu": self.cpu.get_info,

167 "cc": self.cpu.get_cc_info(),

168 "pc": self.cpu.program_counter.get(),

169 "cycle_count": self.cpu.cycles,

170 }

171 log.critical("status dict: %s", repr(data))

172 json_string = json.dumps(data)

173 self.response(json_string)

174

175 def post_memory(self, m):

176 addr = int(m.group(1))

177 e = m.group(3)

178 if e is not None:

179 end = int(e)

180 else:

181 end = addr

182 data = json.loads(self.rfile.read(int(self.headers["Content-Length"])))

183 for i, a in enumerate(range(addr, end + 1)):

184 self.cpu.write_byte(a, data[i])

185 self.response("")

186

187 def post_memory_raw(self, m):

188 addr = int(m.group(1))

189 e = m.group(3)

190 if e is not None:

191 end = int(e)

192 else:

193 end = addr

194 data = self.rfile.read(int(self.headers["Content-Length"]))

195 for i, a in enumerate(range(addr, end + 1)):

196 self.cpu.write_byte(a, data[i])

197 self.response("")

198

199 def post_debug(self, m):

200 handler = logging.StreamHandler()

201 handler.level = 5

202 log.handlers = (handler,)

203 log.critical("Activate full debug logging in %s!", __file__)

204 self.response("")

205

206 def post_quit(self, m):

207 log.critical("Quit CPU from controller server.")

208 self.cpu.quit()

209 self.response_html(headline="CPU running")

210

211 def post_reset(self, m):

212 self.cpu.reset()

213 self.response_html(headline="CPU reset")

214

215

216class ControlHandlerFactory:

217 def __init__(self, cpu):

218 self.cpu = cpu

219

220 def __call__(self, request, client_address, server):

221 return ControlHandler(request, client_address, server, self.cpu)

222

223

224def control_server_thread(cpu, cfg, control_server):

225 # FIXME: Refactor this!

226 timeout = 1

227

228 sockets = [control_server]

229 rs, _, _ = select.select(sockets, [], [], timeout)

230 for s in rs:

231 if s is control_server:

232 control_server._handle_request_noblock()

233 else:

234 pass

235

236 if cpu.running:

237 threading.Timer(interval=0.5,

238 function=control_server_thread,

239 args=(cpu, cfg, control_server)

240 ).start()

241 else:

242 log.critical("Quit control server thread, because CPU doesn't run.")

243

244

245class CPUControlServerMixin:

246 def __init__(self, *args, **kwargs):

247 control_handler = ControlHandlerFactory(self)

248 server_address = (self.cfg.CPU_CONTROL_ADDR, self.cfg.CPU_CONTROL_PORT)

249 try:

250 control_server = HTTPServer(server_address, control_handler)

251 except BaseException:

252 self.running = False

253 raise

254 url = "http://%s:%s" % server_address

255 log.error("Start http control server on: %s", url)

256

257 control_server_thread(self, self.cfg, control_server)

Coverage for MC6809/core/cpu_control_server.py: 20%

146 statements