Sprinter-SDCC/toolchain/sdbg_launcher.py

#!/usr/bin/env python3
"""Изолированный DSS→EXE→main launcher для DAP/session server."""
from __future__ import annotations

import argparse
from contextlib import redirect_stdout
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import uuid

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT/'toolchain'))
from mame_interactive import basename_83, build_events
from make_disk import create_floppy_image
from mame_profile import (MameProfile, add_arguments, from_arguments,
                          write_keyboard_config)
from sdbg.image import read_ihx
from sdbg.model import DebugMap


def disk_entries(exe: Path, data) -> list[tuple[str, str]]:
    paths = [Path(exe), *(Path(name).resolve() for name in data)]
    if any(not path.is_file() for path in paths):
        missing = next(path for path in paths if not path.is_file())
        raise ValueError('Не найден файл для debug-дискеты: ' + str(missing))
    entries = [(basename_83(path.name), str(path)) for path in paths]
    names = [name.replace(' ', '') for name, _ in entries]
    if len(set(names)) != len(names):
        raise ValueError('Коллизия имён 8.3 на debug-дискете')
    clusters = sum((path.stat().st_size + 511) // 512 for path in paths)
    if len(paths) > 224 or clusters > 2847:
        raise ValueError('Файлы не помещаются на FAT12 debug-дискету')
    return entries


def lua_script(path, ready, main_address, signature, events, launch_at,
               dss_timeout=30):
    rows = ',\n'.join('{'+f'{t},"{tag}",{mask},{value}'+'}'
                      for t,tag,mask,value in events)
    text = '''if _G.sdbg_launcher_loaded then return end
_G.sdbg_launcher_loaded=true
local machine=manager.machine
local cpu=machine.devices[":maincpu"]
local debug=machine.debugger
local events={EVENTS}
local signature={SIGNATURE}
local index=1
local phase="booting"
local active=true
local bp=nil
local false_hits=0
local launch_started=nil
local prompt_since=nil
local prompt_row=nil
local function now() local t=machine.time;return t.seconds+t.attoseconds/1e18 end
local vram=nil
for tag,share in pairs(machine.memory.shares) do
  if tag:find("vram",1,true) then vram=share;break end
end
local function dss_prompt_row()
  if not vram then return nil end
  local rgmod=cpu.state.RGMOD.value & 1
  local function char_at(row,col)
    return vram:read_u8((1+col+0x80*rgmod)*1024+0x300+row*4+1)
  end
  for row=0,31 do
    local drive=char_at(row,0)
    if ((drive>=65 and drive<=90) or (drive>=97 and drive<=122)) and
       char_at(row,1)==58 then
      for col=2,38 do
        local value=char_at(row,col)
        if value==62 then
          local clean=true
          for tail=col+1,math.min(col+8,79) do
            local after=char_at(row,tail)
            if after~=0 and after~=32 and after~=95 then clean=false;break end
          end
          if clean then return row end
        elseif value<32 then
          break
        end
      end
    end
  end
  return nil
end
_G.sdbg_launcher_reset_subscription=emu.add_machine_reset_notifier(function()
  if phase~="booting" then active=false end
end)
debug.execution_state="run"
emu.register_periodic(function()
  if not active then return end
  local current=now()
  if phase=="booting" then
    local row=nil
    if current>=LAUNCH_NOT_BEFORE then row=dss_prompt_row() end
    if row then
      if prompt_row~=row then prompt_row=row;prompt_since=current end
      if current-prompt_since>=0.25 then
        machine.video:snapshot()
        bp=cpu.debug:bpset(MAIN,"","")
        launch_started=current
        phase="loading"
      end
    else
      prompt_row=nil;prompt_since=nil
    end
  end
  while phase=="loading" and index<=#events and current>=launch_started+events[index][1] do
    local e=events[index]
    for _,field in pairs(machine.ioport.ports[e[2]].fields) do
      if field.mask==e[3] then field:set_value(e[4]);break end
    end
    index=index+1
  end
  if phase=="loading" and debug.execution_state=="stop" then
    local pc=cpu.state.PC.value
    local match=pc==MAIN
    for i,b in ipairs(signature) do
      if cpu.spaces.program:read_u8(0x10000+MAIN+i-1)~=b then match=false end
    end
    if not match then false_hits=false_hits+1;debug.execution_state="run";return end
    cpu.debug:bpclear(bp)
    local file=assert(io.open(READY..".tmp","wb"))
    file:write('{"pc":'..pc..',"false_hits":'..false_hits..
               ',"dss_ready_time":'..launch_started..',"prompt_row":'..prompt_row..'}')
    file:close();assert(os.rename(READY..".tmp",READY))
    phase="attached"
  end
  if phase=="booting" and current>DSS_TIMEOUT then
    local file=io.open(READY..".error","wb")
    if file then file:write("DSS prompt timeout");file:close() end
    machine:exit()
  elseif phase=="loading" and current>launch_started+30 then
    local file=io.open(READY..".error","wb")
    if file then file:write("main timeout");file:close() end
    machine:exit()
  end
end)
'''
    text = (text.replace('EVENTS',rows).replace('SIGNATURE',','.join(map(str,signature)))
            .replace('LAUNCH_NOT_BEFORE',str(launch_at)).replace('MAIN',str(main_address))
            .replace('READY',json.dumps(str(ready))).replace('DSS_TIMEOUT',str(dss_timeout)))
    path.write_text(text)


def main():
    parser=argparse.ArgumentParser(description=__doc__)
    add_arguments(parser)
    parser.add_argument('--build',required=True)
    parser.add_argument('--socket',required=True)
    parser.add_argument('--data',action='append',default=[])
    parser.add_argument('--launch-at',type=float,default=0,
                        help='не начинать ввод раньше этой секунды эмуляции')
    parser.add_argument('--dss-timeout',type=float,default=30,
                        help='таймаут появления стабильного prompt DSS')
    parser.add_argument('--mame',default=None,
                        help='совместимый alias для --mame-bin')
    parser.add_argument('--debugger',default='sdbg',
                        help='OSD debugger provider (по умолчанию sdbg из project patch)')
    args=parser.parse_args()
    model=DebugMap(args.build)
    model.verify_executable()
    exe=Path(model.manifest['executable_path'])
    functions=[item for item in model.functions if item['name']=='main']
    function=functions[0] if len(functions)==1 else None
    if not function or function['bank'] is not None:
        raise ValueError('Нужна единственная resident-функция main')
    image=read_ihx(model.directory/(exe.stem+'.ihx'))
    signature=bytes(image[address] for address in range(
        function['start'],min(function['end'],function['start']+24)))
    if not signature:
        raise ValueError('У main нет проверяемой сигнатуры')
    if args.mame and not args.mame_bin:
        args.mame_bin=args.mame
    profile=from_arguments(args)
    if profile.binary is None and (ROOT/'mame/v306/mame.arm').is_file():
        profile=MameProfile.resolve({'MAME_HOME':str(ROOT/'mame/v306')})
    profile.validate(dss=True)
    mame=profile.binary
    stopping=False
    def stop(*_):
        nonlocal stopping
        stopping=True
    signal.signal(signal.SIGTERM,stop)
    signal.signal(signal.SIGINT,stop)
    with tempfile.TemporaryDirectory(prefix='sprinter-sdbg-launch-') as temporary:
        state=Path(temporary)
        ipc=state/'ipc';ipc.mkdir()
        for name in ('nvram','cfg','diff','snapshot'): (state/name).mkdir()
        write_keyboard_config(state/'cfg')
        system=state/'system.chd'
        shutil.copyfile(profile.system_hdd_image,system)
        disk_path=state/'debug.img'
        files=disk_entries(exe,args.data)
        with redirect_stdout(sys.stderr):
            if not create_floppy_image(str(disk_path),files):
                raise RuntimeError('Не удалось создать debug-дискету')
        command_text='a:\\'+basename_83(exe.name).replace(' ','')+'\n'
        events=build_events([(0,command_text)])
        ready=state/'main.json'
        lua=state/'launch.lua'
        lua_script(lua,ready,function['start'],signature,events,args.launch_at,
                   args.dss_timeout)
        session_id=uuid.uuid4().hex
        environment=dict(os.environ,SDBG_IPC_DIR=str(ipc),SDBG_SESSION_ID=session_id)
        plugin_paths=[str(ROOT/'toolchain/mcp')]
        installed_plugins=profile.home.parent/'plugins' if profile.home else None
        if installed_plugins and installed_plugins.is_dir():
            plugin_paths.append(str(installed_plugins))
        legacy_plugins=ROOT/'mame/sources/MAME/plugins'
        if legacy_plugins.is_dir():
            plugin_paths.append(str(legacy_plugins))
        command=[str(mame),'sprinter','-noreadconfig','-rompath',str(profile.rompath),
                 '-bios',profile.bios,'-kbd','ms_naturl,bios=sp2k','-video','soft','-window',
                 '-sound','none','-skip_gameinfo','-beta:wd179x:0','35hd',
                 '-beta:wd179x:1','35hd','-flop1',str(disk_path),
                 '-flop2',str(profile.dss_image),'-hard1',str(system),
                 '-debug','-debugger',args.debugger,
                 '-plugin','sdbgbridge','-pluginspath',';'.join(plugin_paths),
                 '-autoboot_delay','0','-autoboot_script',str(lua)]
        for name in ('nvram','cfg','diff','snapshot'):
            command.extend(['-'+name+'_directory',str(state/name)])
        log=(state/'mame.log').open('w')
        mame_process=subprocess.Popen(command,cwd=state,env=environment,
                                      stdout=log,stderr=subprocess.STDOUT)
        server=None
        def mame_diagnostics():
            log.flush()
            output=(state/'mame.log').read_text(errors='replace')[-4000:].strip()
            return ('\nMAME log:\n'+output) if output else ''
        try:
            deadline=time.monotonic()+args.dss_timeout+45
            while not ready.exists():
                if mame_process.poll() is not None:
                    raise RuntimeError('MAME завершился до main; лог: '+
                                       str(state/'mame.log')+mame_diagnostics())
                if (state/'main.json.error').exists() or time.monotonic()>deadline:
                    raise RuntimeError('Таймаут ожидания main; лог: '+
                                       str(state/'mame.log')+mame_diagnostics())
                time.sleep(.05)
            entry=json.loads(ready.read_text())
            server=subprocess.Popen([sys.executable,str(ROOT/'toolchain/sdbg_server.py'),
                '--build',str(model.directory),'--ipc',str(ipc),'--session',session_id,
                '--socket',args.socket],stdout=subprocess.PIPE,stderr=sys.stderr,text=True)
            line=server.stdout.readline()
            if not line:
                raise RuntimeError('Session server не запустился'+mame_diagnostics())
            server_ready=json.loads(line)
            print(json.dumps({'ready':True,'socket':args.socket,'pid':mame_process.pid,
                              'entry':entry,'build_id':server_ready['build_id']},
                             ensure_ascii=False),flush=True)
            while not stopping and mame_process.poll() is None and server.poll() is None:
                if os.getppid()==1: break
                time.sleep(.1)
        finally:
            if server is not None and server.poll() is None:
                server.terminate()
                try: server.wait(timeout=3)
                except subprocess.TimeoutExpired: server.kill();server.wait()
            if mame_process.poll() is None:
                mame_process.terminate()
                try: mame_process.wait(timeout=5)
                except subprocess.TimeoutExpired: mame_process.kill();mame_process.wait()
            log.close()
    return 0


if __name__=='__main__':
    try:
        sys.exit(main())
    except (ValueError,OSError,RuntimeError,KeyError,subprocess.SubprocessError) as error:
        print('sdbg-launcher: '+str(error),file=sys.stderr)
        sys.exit(1)