Apple-2-Tools/applepy
1
0
Fork 0
mirror of https://github.com/jtauber/applepy.git synced 2024-12-28 09:29:51 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #13 from ghewgill/socket

Browse Source 
Change comms to use sockets instead of stdio
This commit is contained in:
James Tauber 2011-08-20 01:50:43 -07:00
commit fad0de8cc9
4 changed files with 109 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
applepy.py
View File

@ -5,6 +5,8 @@
import numpy import numpy
import pygame import pygame
import select
import socket
import struct import struct
import subprocess import subprocess
import sys import sys
@ -357,30 +359,38 @@ class Apple2:
self.speaker = speaker self.speaker = speaker
self.softswitches = SoftSwitches(display, speaker, cassette) self.softswitches = SoftSwitches(display, speaker, cassette)
listener = socket.socket()
listener.bind(("127.0.0.1", 0))
listener.listen(0)
args = [ args = [
sys.executable, sys.executable,
"cpu6502.py", "cpu6502.py",
"--bus", str(listener.getsockname()[1]),
"--rom", options.rom, "--rom", options.rom,
] ]
if options.ram: if options.ram:
args.extend([ args.extend([
"--ram", options.ram, "--ram", options.ram,
]) ])
self.core = subprocess.Popen( self.core = subprocess.Popen(args)
args=args,
stdin=subprocess.PIPE, rs, _, _ = select.select([listener], [], [], 2)
stdout=subprocess.PIPE, if not rs:
) print >>sys.stderr, "CPU module did not start"
sys.exit(1)
self.cpu, _ = listener.accept()
def run(self): def run(self):
update_cycle = 0 update_cycle = 0
quit = False quit = False
while not quit: while not quit:
op = self.core.stdout.read(8) op = self.cpu.recv(8)
if len(op) == 0:
break
cycle, rw, addr, val = struct.unpack("<IBHB", op) cycle, rw, addr, val = struct.unpack("<IBHB", op)
if rw == 0: if rw == 0:
self.core.stdin.write(chr(self.softswitches.read_byte(cycle, addr))) self.cpu.send(chr(self.softswitches.read_byte(cycle, addr)))
self.core.stdin.flush()
elif rw == 1: elif rw == 1:
self.display.update(addr, val) self.display.update(addr, val)
else: else:

58
applepy_curses.py
View File

@ -4,6 +4,7 @@
import curses import curses
import socket
import struct import struct
import subprocess import subprocess
import sys import sys
@ -55,20 +56,29 @@ def write(win, addr, val):
def run(win): def run(win):
global kbd global kbd
p = subprocess.Popen(
args=[sys.executable, "cpu6502.py"], listener = socket.socket()
stdin=subprocess.PIPE, listener.bind(("127.0.0.1", 0))
stdout=subprocess.PIPE, listener.listen(0)
)
args = [
sys.executable,
"cpu6502.py",
"--bus", str(listener.getsockname()[1]),
"--rom", options.rom,
]
p = subprocess.Popen(args)
cpu, _ = listener.accept()
win.clear() win.clear()
curses.noecho() curses.noecho()
win.nodelay(True) win.nodelay(True)
while True: while True:
op = p.stdout.read(8) op = cpu.recv(8)
cycle, rw, addr, val = struct.unpack("<IBHB", op) cycle, rw, addr, val = struct.unpack("<IBHB", op)
if rw == 0: if rw == 0:
p.stdin.write(chr(read(addr, val))) cpu.send(chr(read(addr, val)))
p.stdin.flush()
elif rw == 1: elif rw == 1:
write(win, addr, val) write(win, addr, val)
else: else:
@ -87,5 +97,37 @@ def run(win):
pass pass
def usage():
print >>sys.stderr, "ApplePy - an Apple ][ emulator in Python"
print >>sys.stderr, "James Tauber / http://jtauber.com/"
print >>sys.stderr
print >>sys.stderr, "Usage: applepy_curses.py [options]"
print >>sys.stderr
print >>sys.stderr, " -R, --rom ROM file to use (default A2ROM.BIN)"
sys.exit(1)
def get_options():
class Options:
def __init__(self):
self.rom = "A2ROM.BIN"
options = Options()
a = 1
while a < len(sys.argv):
if sys.argv[a].startswith("-"):
if sys.argv[a] in ("-R", "--rom"):
a += 1
options.rom = sys.argv[a]
else:
usage()
else:
usage()
a += 1
return options
if __name__ == "__main__": if __name__ == "__main__":
options = get_options()
curses.wrapper(run) curses.wrapper(run)

42
cpu6502.py
View File

@ -3,10 +3,14 @@
# originally written 2001, updated 2011 # originally written 2001, updated 2011
import socket
import struct import struct
import sys import sys
bus = None # socket for bus I/O
def signed(x): def signed(x):
if x > 0x7F: if x > 0x7F:
x = x - 0x100 x = x - 0x100
@ -42,8 +46,8 @@ class RAM(ROM):
class Memory: class Memory:
def __init__(self, options=None, use_stdio=True): def __init__(self, options=None, use_bus=True):
self.use_stdio = use_stdio self.use_bus = use_bus
self.rom = ROM(0xD000, 0x3000) self.rom = ROM(0xD000, 0x3000)
if options: if options:
@ -82,26 +86,24 @@ class Memory:
self.bus_write(cycle, address, value) self.bus_write(cycle, address, value)
def bus_read(self, cycle, address): def bus_read(self, cycle, address):
if not self.use_stdio: if not self.use_bus:
return 0 return 0
op = struct.pack("<IBHB", cycle, 0, address, 0) op = struct.pack("<IBHB", cycle, 0, address, 0)
try: try:
sys.stdout.write(op) bus.send(op)
sys.stdout.flush() b = bus.recv(1)
except IOError:
sys.exit(0)
b = sys.stdin.read(1)
if len(b) == 0: if len(b) == 0:
sys.exit(0) sys.exit(0)
return ord(b) return ord(b)
except socket.error:
sys.exit(0)
def bus_write(self, cycle, address, value): def bus_write(self, cycle, address, value):
if not self.use_stdio: if not self.use_bus:
return return
op = struct.pack("<IBHB", cycle, 1, address, value) op = struct.pack("<IBHB", cycle, 1, address, value)
try: try:
sys.stdout.write(op) bus.send(op)
sys.stdout.flush()
except IOError: except IOError:
sys.exit(0) sys.exit(0)
@ -509,7 +511,10 @@ class CPU:
def reset(self): def reset(self):
self.program_counter = self.read_word(self.RESET_VECTOR) self.program_counter = self.read_word(self.RESET_VECTOR)
def run(self): def run(self, bus_port):
global bus
bus = socket.socket()
bus.connect(("127.0.0.1", bus_port))
while True: while True:
self.cycles += 2 # all instructions take this as a minimum self.cycles += 2 # all instructions take this as a minimum
op = self.read_pc_byte() op = self.read_pc_byte()
@ -973,6 +978,7 @@ def usage():
print >>sys.stderr print >>sys.stderr
print >>sys.stderr, "Usage: cpu6502.py [options]" print >>sys.stderr, "Usage: cpu6502.py [options]"
print >>sys.stderr print >>sys.stderr
print >>sys.stderr, " -b, --bus Bus port number"
print >>sys.stderr, " -R, --rom ROM file to use (default A2ROM.BIN)" print >>sys.stderr, " -R, --rom ROM file to use (default A2ROM.BIN)"
print >>sys.stderr, " -r, --ram RAM file to load (default none)" print >>sys.stderr, " -r, --ram RAM file to load (default none)"
sys.exit(1) sys.exit(1)
@ -983,12 +989,16 @@ def get_options():
def __init__(self): def __init__(self):
self.rom = "A2ROM.BIN" self.rom = "A2ROM.BIN"
self.ram = None self.ram = None
self.bus = None
options = Options() options = Options()
a = 1 a = 1
while a < len(sys.argv): while a < len(sys.argv):
if sys.argv[a].startswith("-"): if sys.argv[a].startswith("-"):
if sys.argv[a] in ("-R", "--rom"): if sys.argv[a] in ("-b", "--bus"):
a += 1
options.bus = int(sys.argv[a])
elif sys.argv[a] in ("-R", "--rom"):
a += 1 a += 1
options.rom = sys.argv[a] options.rom = sys.argv[a]
elif sys.argv[a] in ("-r", "--ram"): elif sys.argv[a] in ("-r", "--ram"):
@ -1004,13 +1014,13 @@ def get_options():
if __name__ == "__main__": if __name__ == "__main__":
if sys.stdout.isatty(): options = get_options()
if options.bus is None:
print "ApplePy cpu core" print "ApplePy cpu core"
print "Run applepy.py instead" print "Run applepy.py instead"
sys.exit(0) sys.exit(0)
options = get_options()
mem = Memory(options) mem = Memory(options)
cpu = CPU(mem) cpu = CPU(mem)
cpu.run() cpu.run(options.bus)

26
tests.py
View File

@ -5,7 +5,7 @@ from cpu6502 import Memory, CPU
class TestMemory(unittest.TestCase): class TestMemory(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
def test_load(self): def test_load(self):
self.memory.load(0x1000, [0x01, 0x02, 0x03]) self.memory.load(0x1000, [0x01, 0x02, 0x03])
@ -25,7 +25,7 @@ class TestMemory(unittest.TestCase):
class TestLoadStoreOperations(unittest.TestCase): class TestLoadStoreOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
self.memory.load(0x1000, [0x00, 0x01, 0x7F, 0x80, 0xFF]) self.memory.load(0x1000, [0x00, 0x01, 0x7F, 0x80, 0xFF])
@ -114,7 +114,7 @@ class TestLoadStoreOperations(unittest.TestCase):
class TestRegisterTransferOperations(unittest.TestCase): class TestRegisterTransferOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_TAX(self): def test_TAX(self):
@ -189,7 +189,7 @@ class TestRegisterTransferOperations(unittest.TestCase):
class TestStackOperations(unittest.TestCase): class TestStackOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_TSX(self): def test_TSX(self):
@ -237,7 +237,7 @@ class TestStackOperations(unittest.TestCase):
class TestLogicalOperations(unittest.TestCase): class TestLogicalOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_AND(self): def test_AND(self):
@ -325,7 +325,7 @@ class TestLogicalOperations(unittest.TestCase):
class TestArithmeticOperations(unittest.TestCase): class TestArithmeticOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_ADC_without_BCD(self): def test_ADC_without_BCD(self):
@ -544,7 +544,7 @@ class TestArithmeticOperations(unittest.TestCase):
class TestIncrementDecrementOperations(unittest.TestCase): class TestIncrementDecrementOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_INC(self): def test_INC(self):
@ -653,7 +653,7 @@ class TestIncrementDecrementOperations(unittest.TestCase):
class TestShiftOperations(unittest.TestCase): class TestShiftOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_ASL(self): def test_ASL(self):
@ -760,7 +760,7 @@ class TestShiftOperations(unittest.TestCase):
class TestJumpCallOperations(unittest.TestCase): class TestJumpCallOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_JMP(self): def test_JMP(self):
@ -792,7 +792,7 @@ class TestJumpCallOperations(unittest.TestCase):
class TestBranchOperations(unittest.TestCase): class TestBranchOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_BCC(self): def test_BCC(self):
@ -879,7 +879,7 @@ class TestBranchOperations(unittest.TestCase):
class TestStatusFlagOperations(unittest.TestCase): class TestStatusFlagOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_CLC(self): def test_CLC(self):
@ -921,7 +921,7 @@ class TestStatusFlagOperations(unittest.TestCase):
class TestSystemFunctionOperations(unittest.TestCase): class TestSystemFunctionOperations(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_BRK(self): def test_BRK(self):
@ -951,7 +951,7 @@ class TestSystemFunctionOperations(unittest.TestCase):
class Test6502Bugs(unittest.TestCase): class Test6502Bugs(unittest.TestCase):
def setUp(self): def setUp(self):
self.memory = Memory(use_stdio=False) self.memory = Memory(use_bus=False)
self.cpu = CPU(self.memory) self.cpu = CPU(self.memory)
def test_zero_page_x(self): def test_zero_page_x(self):