Implementing serial support over IP socket

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

Author: Baekalfen

pyboy/__main__.py

@@ -88,6 +88,11 @@ def valid_file_path(path):
 parser.add_argument("--sound", action="store_true", help="Enable sound (beta)")
 parser.add_argument("--no-renderer", action="store_true", help="Disable rendering (internal use)")
 
+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)"
@@ -106,6 +111,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")
+
     print(
         """
 The Game Boy controls are as follows:

pyboy/core/mb.pxd

@@ -22,7 +22,6 @@ from pyboy.utils cimport IntIOInterface, WindowEvent
 cdef Logger logger
 
 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
 
@@ -36,6 +35,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 char[1024] serialbuffer
     cdef uint16_t serialbuffer_count

pyboy/core/mb.py

@@ -6,7 +6,7 @@
 from pyboy import utils
 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)]
 OPCODE_BRK = 0xDB
@@ -27,6 +27,8 @@ def __init__(
         sound_emulated,
         cgb,
         randomize=False,
+        serial_address=None,
+        serial_bind=None,
     ):
         if bootrom_file is not None:
             logger.info("Boot-ROM file provided")
@@ -42,6 +44,7 @@ def __init__(
         self.bootrom = bootrom.BootROM(bootrom_file, cgb)
         self.ram = ram.RAM(cgb, randomize=randomize)
         self.cpu = cpu.CPU(self)
+        self.serial = serial.Serial(serial_address, serial_bind)
 
         if cgb:
             self.lcd = lcd.CGBLCD(
@@ -204,6 +207,7 @@ def buttonevent(self, key):
 
     def stop(self, save):
         self.sound.stop()
+        self.serial.stop()
         if save:
             self.cartridge.stop()
 
@@ -298,10 +302,11 @@ def tick(self):
                     min(
                         self.lcd.cycles_to_interrupt(),
                         self.timer.cycles_to_interrupt(),
-                        # self.serial.cycles_to_interrupt(),
+                        self.serial.cycles_to_transmit(),
                         mode0_cycles
                     )
                 )
+                cycles = 4
 
             #TODO: Support General Purpose DMA
             # https://gbdev.io/pandocs/CGB_Registers.html#bit-7--0---general-purpose-dma
@@ -315,6 +320,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:
@@ -364,7 +371,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
@@ -478,10 +491,16 @@ 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[self.serialbuffer_count] = value
                 self.serialbuffer_count += 1
                 self.serialbuffer_count &= 0x3FF
                 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

@@ -137,6 +137,8 @@ def __init__(
             sound_emulated,
             cgb,
             randomize=randomize,
+            serial_address=kwargs["serial_address"],
+            serial_bind=kwargs["serial_bind"],
         )
 
         # Validate all kwargs