Удалённое управление эмулятором Fceux с помощью Python

В статье я опишу, как сделать эмулятор NES управляемым удалённо, и сервер для удалённой отправки команд на него.

Зачем это нужно?

Некоторые эмуляторы различных игровых консолей, в том числе и Fceux, позволяют писать и запускать пользовательские скрипты на Lua. Но Lua – плохой язык для написания серьёзных программ. Это скорее язык для вызова функций, написанных на Си. Авторы эмуляторов используют его только из-за легковесности и простоты встраивания. Точная эмуляция требует много ресурсов процессора, и ранее скорость эмуляции была одной из главных целей авторов, а о возможности скриптования действий если и вспоминали, то далеко не в первую очередь.

Сейчас мощности среднего процессора с головой хватает для эмуляции NES, почему бы тогда не использовать в эмуляторах мощные скриптовые языки вроде Python или JavaScript?

К сожалению, ни в одном из популярных эмуляторов NES нет возможности использовать эти или другие языки. Я обнаружил только малоизвестный проект Nintaco, который также основан на ядре Fceux, зачем-то переписанном на Java. Тогда я решил добавить возможность написания скриптов на Python для управления эмулятором сам.

Мой результат – это Proof-of-Concept возможности управления эмулятором, он не претендует на скорость или надёжность, но он работает. Я делал его для себя, но так как вопрос о том, как управлять эмулятором с помощью скриптов, встречается достаточно часто, то я выложил исходники на гитхаб.

Как это устроено

На стороне эмулятора

Эмулятор Fceux уже включает в себя несколько Lua-библиотек, включённых в него в виде скомпилированного кода. Одна из них – LuaSocket. Она плохо документирована, однако мне удалось найти пример работающего кода среди коллекции скриптов Xkeeper0. Он использовал сокеты для управления эмулятором через Mirc. Собственно, код, который открывает tcp-сокет:

function connect(address, port, laddress, lport)
    local sock, err = socket.tcp()
    if not sock then return nil, err end
    if laddress then
        local res, err = sock:bind(laddress, lport, -1)
        if not res then return nil, err end
    end
    local res, err = sock:connect(address, port)
    if not res then return nil, err end
    return sock
end

sock2, err2 = connect("127.0.0.1", 81)
sock2:settimeout(0) --it's our socket object
print("Connected", sock2, err2)

Это низкоуровневый сокет, который получает и отправляет данные по 1 байту.

В эмуляторе Fceux основной цикл Lua-скрипта выглядит так:

function main()
    while true do                 --вечный цикл
        passiveUpdate()        --проверка, не пришли ли новые команды через сокет
        emu.frameadvance() --передача управления эмулятору для отрисовки следующего кадра
    end
end

А проверка данных из сокета:

function passiveUpdate()
    local message, err, part = sock2:receive("*all")
    if not message then
        message = part
    end
    if message and string.len(message)>0 then
        --print(message)
        local recCommand = json.decode(message)
        table.insert(commandsQueue, recCommand)
        coroutine.resume(parseCommandCoroutine)
    end
end

Код достаточно прост – делается чтение данных из сокета, и если была обнаружена следующая команда, то осуществляется её парсинг и исполнение. Парсинг и выполнение организованы с помощью корутин (сопрограмм) – это мощная концепция языка Lua для приостановки и продолжения выполнения кода.

И ещё важная одна вещь про Lua-скриптинг в Fceux – эмуляция может быть временно остановлена из скрипта. Каким образом организовать продолжения выполнения Lua-кода и вновь запустить её командой, полученной из сокета? Это было бы невозможно, но существует плохо документированная возможность вызвать Lua-код даже при остановленной эмуляции (спасибо feos за то, что навёл на неё):

gui.register(passiveUpdate) --undocumented. this function will call even if emulator paused

С помощью неё можно останавливать и продолжать эмуляцию внутри passiveUpdate – так можно организовать установку брейкпоинтов эмулятора через сокет.

На стороне сервера команд

Я использую очень простой текстовый RPC-протокол, основанный на JSON. Сервер сериализует название функции и аргументы в JSON-строку и отправляет её через сокет. Дальше выполнение кода останавливается, пока эмулятор не ответит строкой завершения выполнения команды. Ответ будет содержать поля "FUNCTIONNAME_finished" и результат выполнения функции.

Идея реализована в классе syncCall:

class syncCall:
    @classmethod
    def waitUntil(cls, messageName):
        """cycle for reading data from socket until needed message was read from it. All other messages will added in message queue"""
        while True:
            cmd = messages.parseMessages(asyncCall.waitAnswer(), [messageName])
            #print(cmd)
            if cmd != None:
                if len(cmd)>1:
                    return cmd[1]
                return
               
    @classmethod
    def call(cls, *params):
        """wrapper for sending [functionName, [param1, param2, ...]] to socket and wait until client return [functionName_finished, [result1,...]] answer"""
        sender.send(*params)
        funcName = params[0]
        return syncCall.waitUntil(funcName + "_finished")

С помощью этого класса Lua-методы эмулятора Fceux могут быть обёрнуты в Python-классы:

class emu:
    @classmethod
    def poweron(cls):
        return syncCall.call("emu.poweron")
        
    @classmethod
    def pause(cls):
        return syncCall.call("emu.pause")
        
    @classmethod
    def unpause(cls):
        return syncCall.call("emu.unpause")
        
    @classmethod
    def message(cls, str):
        return syncCall.call("emu.message", str)
        
    @classmethod
    def softreset(cls):
        return syncCall.call("emu.softreset")
        
    @classmethod
    def speedmode(cls, str):
        return syncCall.call("emu.speedmode", str)

И затем вызваны дословно так же, как и из Lua:

#Перезапуск игры:
emu.poweron()

Методы обратного вызова

В Lua можно зарегистрировать колбеки – функции, которые будут вызваны при выполнении определённого условия. Мы можем перенести это поведение в сервер на Python с помощью следующего приёма. Сначала мы сохраняем идентификатор функции-колбека, написанной на Python, и передаём его в Lua-код:

class callbacks:
    functions = {}
    
    callbackList = [
        "emu.registerbefore_callback",
        "emu.registerafter_callback",
        "memory.registerexecute_callback",
        "memory.registerwrite_callback",
    ]
    
    @classmethod
    def registerfunction(cls, func):
        if func == None:
            return 0
        hfunc = hash(func)
        callbacks.functions[hfunc] = func
        return hfunc
        
    @classmethod 
    def error(cls, e):
        emu.message("Python error: " + str(e))
    
    @classmethod
    def checkAllCallbacks(cls, cmd):
        #print("check:", cmd)
        for callbackName in callbacks.callbackList:
            if cmd[0] == callbackName:
                hfunc = cmd[1]
                #print("hfunc:", hfunc)
                func = callbacks.functions.get(hfunc)
                #print("func:", func)
                if func:
                    try:
                        func(*cmd[2:]) #skip function name and function hash and save others arguments
                    except Exception as e:
                        callbacks.error(e)
                        pass
                    #TODO: thread locking
                    sender.send(callbackName + "_finished")

Lua-код также сохраняет этот идентификатор и регистрирует обычный Lua-колбек, который будет передавать управление в Python-код. Далее, в Python-коде создаётся отдельный поток, который занимается только тем, что проверяет, не была ли принята команда вызова колбека из Lua:

def callbacksThread():
    cycle = 0
    while True:
        cycle += 1
        try:
            cmd = messages.parseMessages(asyncCall.waitAnswer(), callbacks.callbackList)
            if cmd:
                #print("Callback received:", cmd)
                callbacks.checkAllCallbacks(cmd)
            pass
        except socket.timeout:
            pass
        time.sleep(0.001)

Последний шаг – после выполнения Python-колбека управление возвращается в Lua с помощью команды "CALLBACKNAME_finished", чтобы информировать эмулятор, что колбек закончен.

Как запустить пример

• У вас должны быть работающие Python 3 и Jupyter Notebook в системе. Необходимо запустить Jupyter командой

  jupyter notebook
  

• Откройте ноутбук FceuxPythonServer.py.ipynb и запустите первую строку
  
• Теперь вы должны запустить эмулятор Fceux, открыть в нём ROM-файл (я использую игру Castlevania (U) (PRG0) [!].nes в своём примере) и запустить Lua-скрипт с именем fceux_listener.lua. Он должен соединиться с сервером, запущенном в ноутбуке Jupyter.

  Эти действия можно выполнить с помощью командной строки:

  fceux.exe -lua fceux_listener.lua "Castlevania (U) (PRG0) [!].nes"
  

• Теперь снова переключитесь на Jupyter Notebook. Вы должны увидеть сообщение об успешном соединении с эмулятором:

  

Всё, вы можете посылать команды из ноутбука Jupyter в браузере прямо в эмулятор Fceux.

Можно выполнять все строки ноутбука-примера последовательно и наблюдать за результатом выполнения в эмуляторе.

Полный пример:
https://github.com/spiiin/fceux_luaserver/blob/master/FceuxPythonServer.py.ipynb

Он содержит простые функции вроде чтения памяти:

Более сложные примеры с созданием колбеков:

И скрипт для конкретной игры, позволяющий перемещать врагов из Super Mario Bros. с помощью мыши:

Видео выполнения ноутбука:

Ограничения и применения

Скрипт не имеет защиты от дурака и не оптимизирован по скорости выполнения – лучше было бы использовать бинарный RPC-протокол вместо текстового и группировать сообщения вместе, но моя реализация не требуется компиляции. Скрипт может переключать контексты выполнения из Lua в Python и обратно 500-1000 раз в секунду на моём ноутбуке. Этого достаточно почти для любых применений, кроме специфических случаев попиксельной или построчной отладки видеопроцессора, но Fceux всё равно не позволяет проводить такие операции из Lua, так что это неважно.

Возможные идеи применения:

• Как пример реализации подобного управления для других эмуляторов и языков
• Исследование игр
• Добавление читов или фич для организации TAS-прохождений
• Вставка или извлечение данных и кода в игры
• Расширение возможностей эмуляторов — написание отладчиков, скриптов записи и просмотра прохождений, скриптовых библиотек, редакторов игр
• Сетевая игра, контроль игры с помощью мобильных устройств, удалённых сервисов, джойпадов или других устройств управления, сохранение и патчи в облачных сервисах
• Кросс-эмуляторные фичи
• Использование библиотек языков Python или других для анализа данных и управления игрой (создание ботов)

Стек технологий

Я использовал:

Fceux — www.fceux.com/web/home.html
Это классический эмулятор NES, и большинство людей используют его. Он не обновлялся уже долгое время, и не лучший по возможностям, но он остаётся эмулятором по умолчанию для множества ромхакеров. Также, я выбрал его из-за того, что в него интегрирована поддержка Lua-сокетов, и нет необходимости подключать её самому.

Json.lua — github.com/spiiin/json.lua
Это реализация JSON на чистом Lua. Я выбрал её, потому что хотел сделать пример, который не требует компиляции кода. Но мне всё равно пришлось сделать форк библиотеки, потому что какая-то из встроенных во Fceux библиотек перегружала библиотечную функцию tostring и ломала сериализацию (мой отклонённый пул-реквест автору оригинальной библиотеки).

Python 3 — www.python.org
Fceux Lua сервер открыват tcp-сокет и слушает команды, полученные от него. Сервер, который посылает команды эмулятору, может быть реализован на любом языке. Я выбрал Python за его философию «Battery included» – большинство модулей включены в стандартную библиотеку (работа с сокетами и JSON в том числе). Также Python известен библиотека работы с нейронными сетями, и мне хочется попробовать использовать их для создания ботов в NES-играх.

Jupyter Notebook — jupyter.org
Jupyter Notebook – очень крутая среда для интерактивного выполнения Python-кода. С помощью неё вы можете писать и выполнять команды в табличном редакторе внутри браузера. Он также хорош для создания презентабельных примеров.

Dexpot — www.dexpot.de
Я использовал этот менеджер виртуальных рабочих столов, для того, чтобы закреплять окно эмулятора поверх других. Это очень удобно при разворачивании сервера на полный экран для мгновенного отслеживание изменений в окне эмулятора. Штатные средства Windows не позволяют организовать закрепление окна поверх других.

Ссылки

Собственно, репозиторий проекта.

Nintaco — эмулятор NES на Java с возможностью удалённого управления
Xkeeper0 emu-lua collection — коллекция различных Lua-скриптов
Mesen — современный эмулятор NES на C# с мощными возможностями написания скриптов на Lua. Пока без поддержки сокетов и удалённого управления.
CadEditor — мой проект универсального редактора уровней для NES и других платформ, а также мощные инструменты для исследования игр. Я использую скрипт и сервер, описанные в посте, для того, чтобы исследовать игры и добавлять их в редактор.

Буду признателен за отзывы, тестирование и попытки использования скрипта.

Автор: spiiin

Источник