Implementing serial support over IP socket

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

Author: Baekalfen

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:

pyboy/core/mb.pxd

@@ -23,7 +23,7 @@ cdef Logger logger
 
 cdef int64_t MAX_CYCLES
 cdef uint16_t STAT, LY, LYC
-cdef int INTR_TIMER, INTR_HIGHTOLOW
+cdef int INTR_TIMER, INTR_HIGHTOLOW, INTR_SERIAL
 cdef uint16_t OPCODE_BRK
 cdef int STATE_VERSION
 
@@ -37,6 +37,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

@@ -10,10 +10,11 @@
     PyBoyOutOfBoundsException,
     INTR_TIMER,
     INTR_HIGHTOLOW,
+    INTR_SERIAL,
     OPCODE_BRK,
 )
 
-from . import bootrom, cartridge, cpu, interaction, lcd, ram, sound, timer
+from . import bootrom, cartridge, cpu, interaction, lcd, ram, serial, sound, timer
 
 logger = pyboy.logging.get_logger(__name__)
 MAX_CYCLES = 1 << 31
@@ -31,6 +32,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")
@@ -51,6 +54,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(
@@ -224,6 +228,7 @@ def buttonevent(self, key):
 
     def stop(self, save):
         self.sound.stop()
+        self.serial.stop()
         if save:
             self.cartridge.stop()
 
@@ -321,7 +326,7 @@ def tick(self):
                         self.lcd._cycles_to_interrupt,  # TODO: Be more agreesive. Only if actual interrupt enabled.
                         self.lcd._cycles_to_frame,
                         self.sound._cycles_to_interrupt,
-                        # self.serial.cycles_to_interrupt(),
+                        self.serial.cycles_to_transmit(),
                         mode0_cycles,
                     ),
                 )
@@ -336,6 +341,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)
@@ -380,7 +387,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 0xFF04 <= i <= 0xFF07:
+            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 0xFF04 <= i <= 0xFF07:
                 if self.timer.tick(self.cpu.cycles):
                     self.cpu.set_interruptflag(INTR_TIMER)
 
@@ -511,10 +524,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 0xFF04 <= i <= 0xFF07:
                 if self.timer.tick(self.cpu.cycles):
                     self.cpu.set_interruptflag(INTR_TIMER)

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

@@ -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