citra-shitamoto-network/dist/scripting/citra.py
EverOddish 04dd91be82 Initial support for scripting (#4016)
* Add ZeroMQ external submodule

* ZeroMQ libzmq building on macOS

* Added RPC namespace, settings and logging

* Added request queue handling and new classes

* Add C++ interface to ZeroMQ

* Added start of ZeroMQ RPC Server implementation.

* Request construction and callback request handling

* Read and write memory implementation

* Add ID to request format and send reply

* Add RPC setting to macOS UI

* Fixed initialization order bug and added exception handling

* Working read-write through Python

* Update CMakeLists for libzmq to resolve target name conflict on Windows

* Platform-specific CMake definitions for Windows/non-Windows

* Add comments

* Revert "Add RPC setting to macOS UI"

* Always run RPC server instead of configurable

* Add Python scripting example. Updated .gitignore

* Rename member variables to remove trailing underscore

* Finally got libzmq external project building on macOS

* Add missing dependency during libzmq build

* Adding more missing dependencies [skip ci]

* Only build what is required from libzmq

* Extra length checks on client input

* Call InvalidateCacheRange after memory write

* Revert MinGW change. Fix clang-format. Improve error handling in request/reply. Allow any length of data read/write in Python.

* Re-organized RPC static global state into a proper class. [skip ci]

* Make sure libzmq always builds in Release mode

* Renamed Request to Packet since Request and Reply are the same thing

* Moved request fulfillment out of Packet and into RPCServer

* Change request thread from sleep to condition variable

* Remove non-blocking polling from ZMQ server code. Receive now blocks and terminates properly without sleeping. This change significantly improves script speed.

* Move scripting files to dist/ instead of src/

* C++ code review changes for jroweboy [skip ci]

* Python code review changes for jroweboy [skip ci]

* Add docstrings and tests to citra.py [skip ci]

* Add host OS check for libzmq build

* Revert "Add host OS check for libzmq build"

* Fixed a hang when emulation is stopped and restarted due to improper destruction order of ZMQ objects [skip ci]

* Add scripting directory to archive packaging [skip ci]

* Specify C/CXX compiler variables on MinGW build

* Only specify compiler on Linux mingw

* Use gcc and g++ on Windows mingw

* Specify generator for mingw

* Don't specify toolchain on windows mingw

* Changed citra.py to support Python 3 instead of Python 2

* Fix bug where RPC wouldn't restart after Stop/Start emulation

* Added copyright to headers and reorganized includes and forward declarations
2018-09-11 22:00:12 +02:00

95 lines
3.3 KiB
Python

import zmq
import struct
import random
import binascii
CURRENT_REQUEST_VERSION = 1
MAX_REQUEST_DATA_SIZE = 32
REQUEST_TYPE_READ_MEMORY = 1
REQUEST_TYPE_WRITE_MEMORY = 2
CITRA_PORT = "45987"
class Citra:
def __init__(self, address="127.0.0.1", port=CITRA_PORT):
self.context = zmq.Context()
self.socket = self.context.socket(zmq.REQ)
self.socket.connect("tcp://" + address + ":" + port)
def is_connected(self):
return self.socket is not None
def _generate_header(self, request_type, data_size):
request_id = random.getrandbits(32)
return (struct.pack("IIII", CURRENT_REQUEST_VERSION, request_id, request_type, data_size), request_id)
def _read_and_validate_header(self, raw_reply, expected_id, expected_type):
reply_version, reply_id, reply_type, reply_data_size = struct.unpack("IIII", raw_reply[:4*4])
if (CURRENT_REQUEST_VERSION == reply_version and
expected_id == reply_id and
expected_type == reply_type and
reply_data_size == len(raw_reply[4*4:])):
return raw_reply[4*4:]
return None
def read_memory(self, read_address, read_size):
"""
>>> c.read_memory(0x100000, 4)
b'\\x07\\x00\\x00\\xeb'
"""
result = bytes()
while read_size > 0:
temp_read_size = min(read_size, MAX_REQUEST_DATA_SIZE)
request_data = struct.pack("II", read_address, temp_read_size)
request, request_id = self._generate_header(REQUEST_TYPE_READ_MEMORY, len(request_data))
request += request_data
self.socket.send(request)
raw_reply = self.socket.recv()
reply_data = self._read_and_validate_header(raw_reply, request_id, REQUEST_TYPE_READ_MEMORY)
if reply_data:
result += reply_data
read_size -= len(reply_data)
read_address += len(reply_data)
else:
return None
return result
def write_memory(self, write_address, write_contents):
"""
>>> c.write_memory(0x100000, b"\\xff\\xff\\xff\\xff")
True
>>> c.read_memory(0x100000, 4)
b'\\xff\\xff\\xff\\xff'
>>> c.write_memory(0x100000, b"\\x07\\x00\\x00\\xeb")
True
>>> c.read_memory(0x100000, 4)
b'\\x07\\x00\\x00\\xeb'
"""
write_size = len(write_contents)
while write_size > 0:
temp_write_size = min(write_size, MAX_REQUEST_DATA_SIZE - 8)
request_data = struct.pack("II", write_address, temp_write_size)
request_data += write_contents[:temp_write_size]
request, request_id = self._generate_header(REQUEST_TYPE_WRITE_MEMORY, len(request_data))
request += request_data
self.socket.send(request)
raw_reply = self.socket.recv()
reply_data = self._read_and_validate_header(raw_reply, request_id, REQUEST_TYPE_WRITE_MEMORY)
if None != reply_data:
write_address += temp_write_size
write_size -= temp_write_size
write_contents = write_contents[temp_write_size:]
else:
return False
return True
if "__main__" == __name__:
import doctest
doctest.testmod(extraglobs={'c': Citra()})