Implementing serial support over IP socket

Co-authored-by: thejomas <[email protected]>

Author: Baekalfen

pyboy/__main__.py

@@ -74,6 +74,11 @@ def valid_file_path(path):
 parser.add_argument("--disable-renderer", action="store_true", help="Disables screen rendering for higher performance")
 parser.add_argument("--sound", action="store_true", help="Enable sound (beta)")
 
+parser.add_argument("--serial-bind", action="store_true", help="Bind to this TCP addres for using Link Cable")
+parser.add_argument(
+    "--serial-address", default=None, type=str, help="Connect (or bind) to this TCP addres for using Link Cable"
+)
+
 gameboy_type_parser = parser.add_mutually_exclusive_group()
 gameboy_type_parser.add_argument(
     "--dmg", action="store_const", const=False, dest="cgb", help="Force emulator to run as original Game Boy (DMG)"
@@ -91,6 +96,9 @@ def valid_file_path(path):
 
 def main():
     argv = parser.parse_args()
+    if argv.serial_bind and not argv.serial_address:
+        parser.error("--serial-bind requires --serial-address")
+
     log_level(argv.log_level)
 
     logger.info(

pyboy/core/mb.pxd

@@ -19,7 +19,6 @@ from pyboy.utils cimport WindowEvent
 
 
 cdef uint16_t STAT, LY, LYC
-cdef short VBLANK, LCDC, TIMER, SERIAL, HIGHTOLOW
 cdef int INTR_VBLANK, INTR_LCDC, INTR_TIMER, INTR_SERIAL, INTR_HIGHTOLOW
 cdef int STATE_VERSION
 
@@ -33,6 +32,7 @@ cdef class Motherboard:
     cdef pyboy.core.timer.Timer timer
     cdef pyboy.core.sound.Sound sound
     cdef pyboy.core.cartridge.base_mbc.BaseMBC cartridge
+    cdef object serial
     cdef bint bootrom_enabled
     cdef str serialbuffer
 

pyboy/core/mb.py

@@ -8,7 +8,7 @@
 from pyboy.core.opcodes import CPU_COMMANDS
 from pyboy.utils import STATE_VERSION
 
-from . import bootrom, cartridge, cpu, interaction, lcd, ram, sound, timer
+from . import bootrom, cartridge, cpu, interaction, lcd, ram, serial, sound, timer
 
 INTR_VBLANK, INTR_LCDC, INTR_TIMER, INTR_SERIAL, INTR_HIGHTOLOW = [1 << x for x in range(5)]
 
@@ -26,6 +26,8 @@ def __init__(
         cgb,
         randomize=False,
         profiling=False,
+        serial_address=None,
+        serial_bind=None,
     ):
         if bootrom_file is not None:
             logger.info("Boot-ROM file provided")
@@ -43,6 +45,7 @@ def __init__(
         self.bootrom = bootrom.BootROM(bootrom_file, cgb)
         self.ram = ram.RAM(cgb, randomize=randomize)
         self.cpu = cpu.CPU(self, profiling)
+        self.serial = serial.Serial(serial_address, serial_bind)
 
         if cgb:
             self.lcd = lcd.CGBLCD(
@@ -105,6 +108,7 @@ def buttonevent(self, key):
 
     def stop(self, save):
         self.sound.stop()
+        self.serial.stop()
         if save:
             self.cartridge.stop()
 
@@ -210,9 +214,10 @@ def tick(self):
                 cycles = min(
                     self.lcd.cycles_to_interrupt(),
                     self.timer.cycles_to_interrupt(),
-                    # self.serial.cycles_to_interrupt(),
-                    mode0_cycles
+                    self.serial.cycles_to_transmit(),
+                    mode0_cycles,
                 )
+                cycles = 4
 
                 # Profiling
                 self.cpu.add_opcode_hit(0x76, cycles // 4)
@@ -228,6 +233,8 @@ def tick(self):
 
             if self.timer.tick(cycles):
                 self.cpu.set_interruptflag(INTR_TIMER)
+            if self.serial.tick(cycles):
+                self.cpu.set_interruptflag(INTR_SERIAL)
 
             lcd_interrupt = self.lcd.tick(cycles)
             if lcd_interrupt:
@@ -280,7 +287,13 @@ def getitem(self, i):
         elif 0xFEA0 <= i < 0xFF00: # Empty but unusable for I/O
             return self.ram.non_io_internal_ram0[i - 0xFEA0]
         elif 0xFF00 <= i < 0xFF4C: # I/O ports
-            if i == 0xFF04:
+            if i == 0xFF01:
+                logger.info(f"get SB {self.serial.SB}")
+                return self.serial.SB
+            elif i == 0xFF02:
+                logger.info(f"get SC {self.serial.SC}")
+                return self.serial.SC
+            elif i == 0xFF04:
                 return self.timer.DIV
             elif i == 0xFF05:
                 return self.timer.TIMA
@@ -398,8 +411,14 @@ def setitem(self, i, value):
             if i == 0xFF00:
                 self.ram.io_ports[i - 0xFF00] = self.interaction.pull(value)
             elif i == 0xFF01:
+                self.serial.SB = value
+                logger.info(f"SB: {value:02x}")
+
                 self.serialbuffer += chr(value)
                 self.ram.io_ports[i - 0xFF00] = value
+            elif i == 0xFF02:
+                self.serial.SC = value
+                logger.info(f"SC: {value:02x}")
             elif i == 0xFF04:
                 self.timer.reset()
             elif i == 0xFF05:

pyboy/core/serial.py

@@ -0,0 +1,112 @@
+import logging
+import os
+import socket
+import sys
+
+logger = logging.getLogger(__name__)
+
+SERIAL_FREQ = 8192 # Hz
+CPU_FREQ = 4213440 # Hz
+
+
+class Serial:
+    def __init__(self, serial_address, serial_bind, serial_interrupt_based=True):
+        self.SB = 0
+        self.SC = 0
+        self.connection = None
+
+        self.trans_bits = 0
+        self.cycles_count = 0
+        self.cycles_target = CPU_FREQ // SERIAL_FREQ
+        self.serial_interrupt_based = serial_interrupt_based
+
+        if not serial_address:
+            logger.info("No serial address supplied. Link Cable emulation disabled.")
+            return
+
+        if not serial_address.count(".") == 3 and serial_address.count(":") == 1:
+            logger.info("Only IP-addresses of the format x.y.z.w:abcd is supported")
+            return
+
+        address_ip, address_port = serial_address.split(":")
+        address_tuple = (address_ip, int(address_port))
+
+        if serial_bind:
+            self.binding_connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self.binding_connection.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            logger.info(f"Binding to {serial_address}")
+            self.binding_connection.bind(address_tuple)
+            self.binding_connection.listen(1)
+            self.connection, _ = self.binding_connection.accept()
+            logger.info(f"Client has connected!")
+        else:
+            self.connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            logger.info(f"Connecting to {serial_address}")
+            self.connection.connect(address_tuple)
+            logger.info(f"Connection successful!")
+        # self.connection.setblocking(False)
+
+    def tick(self, cycles):
+        # if self.serial_interrupt_based:
+        #     if self.SC & 1: # Master
+        #         if self.SC & 0x80:
+        #             logger.info(f'Master sending!')
+        #             self.connection.send(bytes([self.SB]))
+        #             # self.connection.setblocking(True)
+        #             data = self.connection.recv(1)
+        #             self.SB = data[0]
+        #             self.SC &= 0b0111_1111
+        #             return True
+        #     else:
+        #         try:
+        #             if self.SC & 0x80:
+        #                 # self.connection.setblocking(False)
+        #                 logger.info(f'Slave recv!')
+        #                 self.connection.send(bytes([self.SB]))
+        #                 data = self.connection.recv(1)
+        #                 self.SB = data[0]
+        #                 self.SC &= 0b0111_1111
+        #                 return True
+        #         except BlockingIOError:
+        #             pass
+        #     return False
+        # return False
+        # else:
+        # Check if serial is in progress
+
+        if self.connection is None:
+            return
+
+        if self.SC & 0x80 == 0:
+            return False
+
+        self.cycles_count += 1
+
+        if (self.cycles_to_transmit() == 0):
+            # if self.SC & 1: # Master
+            send_bit = bytes([(self.SB >> 7) & 1])
+            self.connection.send(send_bit)
+
+            data = self.connection.recv(1)
+            self.SB = ((self.SB << 1) & 0xFF) | data[0] & 1
+
+            logger.info(f"recv sb: {self.SB:08b}")
+            self.trans_bits += 1
+
+            self.cycles_count = 0
+
+            if self.trans_bits == 8:
+                self.trans_bits = 0
+                self.SC &= 0b0111_1111
+                return True
+            return False
+
+    def cycles_to_transmit(self):
+        if self.SC & 0x80:
+            return max(self.cycles_target - self.cycles_count, 0)
+        else:
+            return 1 << 16
+
+    def stop(self):
+        if self.connection:
+            self.connection.close()

pyboy/pyboy.py

@@ -88,6 +88,8 @@ def __init__(
             cgb,
             randomize=randomize,
             profiling=profiling,
+            serial_address=kwargs["serial_address"],
+            serial_bind=kwargs["serial_bind"],
         )
 
         # Performance measures