Implementing serial support over IP socket

Co-authored-by: thejomas <flachenjensen@gmail.com>
Baekalfen · Baekalfen · Sep 15, 2022 · Sep 15, 2022 · May 21, 2025 · May 20, 2025
commit b8a81dd7a3f74b1c66c053ca2e24ee6480bd2750
diff --git a/pyboy/__main__.py b/pyboy/__main__.py
@@ -107,6 +107,11 @@ def valid_sample_rate(freq):
     help="Add GameShark cheats on start-up. Add multiple by comma separation (i.e. '010138CD, 01033CD1')",
 )
 
+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)"
@@ -160,6 +165,9 @@ def valid_sample_rate(freq):
 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:

diff --git a/pyboy/core/mb.pxd b/pyboy/core/mb.pxd
@@ -39,6 +39,7 @@ cdef class Motherboard:
     cdef pyboy.core.serial.Serial serial
     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

diff --git a/pyboy/core/mb.py b/pyboy/core/mb.py
@@ -11,6 +11,7 @@
     INTR_TIMER,
     INTR_SERIAL,
     INTR_HIGHTOLOW,
+    INTR_SERIAL,
     OPCODE_BRK,
     MAX_CYCLES,
 )
@@ -32,6 +33,8 @@ def __init__(
         sound_sample_rate,
         cgb,
         randomize=False,
+        serial_address=None,
+        serial_bind=None,
     ):
         if bootrom_file is not None:
             logger.info("Boot-ROM file provided")
@@ -53,6 +56,7 @@ def __init__(
         self.interaction = interaction.Interaction()
         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(
@@ -228,6 +232,7 @@ def buttonevent(self, key):
 
     def stop(self, save):
         self.sound.stop()
+        self.serial.stop()
         if save:
             self.cartridge.stop()
 
@@ -346,6 +351,8 @@ def tick(self):
 
             if self.timer.tick(self.cpu.cycles):
                 self.cpu.set_interruptflag(INTR_TIMER)
+            if self.serial.tick(cycles):
+                self.cpu.set_interruptflag(INTR_SERIAL)
 
             if lcd_interrupt := self.lcd.tick(self.cpu.cycles):
                 self.cpu.set_interruptflag(lcd_interrupt)

diff --git a/pyboy/core/serial.py b/pyboy/core/serial.py
@@ -3,13 +3,17 @@
 # GitHub: https://github.com/Baekalfen/PyBoy
 #
 
+import socket
+import logging
 from pyboy.utils import MAX_CYCLES
 
+logger = logging.getLogger(__name__)
+
 CYCLES_8192HZ = 128
 
 
 class Serial:
-    def __init__(self):
+    def __init__(self, serial_address, serial_bind, serial_interrupt_based=True):
         self.SB = 0xFF  # Always 0xFF for a disconnected link cable
         self.SC = 0
         self.transfer_enabled = 0
@@ -19,6 +23,37 @@ def __init__(self):
         self.clock = 0
         self.clock_target = MAX_CYCLES
 
+        self.connection = None
+
+        self.trans_bits = 0
+        self.serial_interrupt_based = serial_interrupt_based
+
+        if not serial_address:
+            logger.info("No serial address supplied. Link Cable emulated as disconnected.")
+            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 set_SB(self, value):
         # Always 0xFF when cable is disconnected. Connecting is not implemented yet.
         self.SB = 0xFF
@@ -47,16 +82,64 @@ def tick(self, _cycles):
         self.clock += cycles
 
         interrupt = False
-        if self.transfer_enabled and self.clock >= self.clock_target:
-            self.SC &= 0x80
-            self.transfer_enabled = 0
-            # self._cycles_to_interrupt = MAX_CYCLES
-            self.clock_target = MAX_CYCLES
-            interrupt = True
+        if self.transfer_enabled:
+            if self.connection is None:
+                # Disconnected emulation
+                if self.clock >= self.clock_target:
+                    self.SC &= 0x80
+                    self.transfer_enabled = 0
+                    # self._cycles_to_interrupt = MAX_CYCLES
+                    self.clock_target = MAX_CYCLES
+                    interrupt = True
+            else:
+                # Connected emulation
+                if self.clock >= self.clock_target:
+                    # 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
+
+                    if self.trans_bits == 8:
+                        self.trans_bits = 0
+                        self.SC &= 0b0111_1111
+                        return True
+                    return False
 
         self._cycles_to_interrupt = self.clock_target - self.clock
         return interrupt
 
+    # 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
+
     def save_state(self, f):
         f.write(self.SB)
         f.write(self.SC)
@@ -76,3 +159,8 @@ def load_state(self, f, state_version):
         self._cycles_to_interrupt = f.read_64bit()
         self.clock = f.read_64bit()
         self.clock_target = f.read_64bit()
+
+
+    def stop(self):
+        if self.connection:
+            self.connection.close()
diff --git a/pyboy/pyboy.py b/pyboy/pyboy.py
@@ -209,6 +209,8 @@ def __init__(
             sound_sample_rate,
             cgb,
             randomize=randomize,
+            serial_address=kwargs["serial_address"],
+            serial_bind=kwargs["serial_bind"],
         )
 
         # Validate all kwargs