Leahnaya/PokemonGB_Online_Trades
1
0
Fork 0
mirror of https://github.com/Lorenzooone/PokemonGB_Online_Trades.git synced 2026-03-21 18:04:50 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fix winusbcdc issues

Browse Source
This commit is contained in:
Lorenzo Carletti 2024-01-27 19:56:48 +01:00
parent 582117f7c4
commit 0267d9a161
2 changed files with 80 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
multiboot.py
View File

@ -1,7 +1,10 @@
import os
import time
import traceback
import sys
# From: https://github.com/Squaresweets/TileWorldGBA
max_packet_size_mb = 0x40
def get_configure_list(us_between_transfer, bytes_for_transfer):
config_base = [0xCA, 0xFE, 0xCA, 0xFE, 0xCA, 0xFE, 0xCA, 0xFE, 0xCA, 0xFE, 0xCA, 0xFE, 0xCA, 0xFE, 0xCA, 0xFE, 0xDE, 0xAD, 0xBE, 0xEF, 0xDE, 0xAD, 0xBE, 0xEF, 0xDE, 0xAD, 0xBE, 0xEF, 0xDE, 0xAD, 0xBE, 0xEF]
@ -9,16 +12,21 @@ def get_configure_list(us_between_transfer, bytes_for_transfer):
return config_base
def read_all(receiver, debug=False):
time.sleep(0.01)
output = 0
prev_len = 0
while True:
try:
data = receiver(0x40)
data = receiver()
if len(data) == 0:
break
output <<= (8 * len(data))
output |= int.from_bytes(data, byteorder='big')
if len(data) < max_packet_size_mb:
break
except:
#traceback.print_exc()
#print("Unexpected exception: ", sys.exc_info()[0])
break
if debug:
print("0x%02x " % output)
@ -90,7 +98,7 @@ def multiboot(receiver, sender, list_sender, path):
sender(0x63D1, 4)
token = read_all(receiver)
if (token >> 24) != 0x73:
if ((token >> 24) & 0xFF) != 0x73:
print("Failed handshake!")
return
else:
@ -117,7 +125,7 @@ def multiboot(receiver, sender, list_sender, path):
complete_sending_data[(i*4)+3] = ((sending_data[i] ^ seed)>>0) & 0xFF
time_transfer = time.time()
list_sender(complete_sending_data, chunk_size = 0x40)
list_sender(complete_sending_data, chunk_size = max_packet_size_mb)
time_transfer = time.time()-time_transfer
print(time_transfer)
@ -133,11 +141,12 @@ def multiboot(receiver, sender, list_sender, path):
crcC = (crcC >> 1) ^ 0xc37b
tmp >>= 1
read_all(receiver)
sender(0x0065, 4)
while True:
sender(0x0065, 4)
recv = read_all(receiver)
if (recv >> 16) == 0x0075:
if ((recv >> 16) & 0xFFFF) == 0x0075:
break
sender(0x0066, 4)

76
usb_trading.py
View File

@ -23,6 +23,7 @@ epOut = None
p = None
max_usb_timeout_w = 5
max_usb_timeout_r = 0.1
max_packet_size = 0x40
VID = 0xcafe
PID = 0x4011
@ -52,7 +53,9 @@ def transfer_func(sender, receiver, list_sender, raw_receiver, is_serial):
else:
config_base = multiboot.get_configure_list(1000, 1)
multiboot.read_all(raw_receiver)
result = 1
while result != 0:
result = multiboot.read_all(raw_receiver)
list_sender(config_base, chunk_size=len(config_base))
ret = multiboot.read_all(raw_receiver)
@ -103,11 +106,15 @@ def sendList(data, chunk_size=8):
if (num_iters*chunk_size) != len(data):
epOut.write(data[num_iters*chunk_size:], timeout=int(max_usb_timeout_w * 1000))
def receiveByte(num_bytes):
def receiveByte(num_bytes=None):
if num_bytes is None:
num_bytes = max_packet_size
recv = int.from_bytes(epIn.read(num_bytes, timeout=int(max_usb_timeout_r * 1000)), byteorder='big')
return recv
def receiveByte_raw(num_bytes):
def receiveByte_raw(num_bytes=None):
if num_bytes is None:
num_bytes = max_packet_size
return epIn.read(num_bytes, timeout=int(max_usb_timeout_r * 1000))
# Code dependant on this connection method
@ -125,11 +132,15 @@ def sendList_serial(data, chunk_size=8):
if (num_iters*chunk_size) != len(data):
serial_port.write(bytes(data[num_iters*chunk_size:]))
def receiveByte_serial(num_bytes):
def receiveByte_serial(num_bytes=None):
if num_bytes is None:
num_bytes = max_packet_size
recv = int.from_bytes(serial_port.read(num_bytes), byteorder='big')
return recv
def receiveByte_raw_serial(num_bytes):
def receiveByte_raw_serial(num_bytes=None):
if num_bytes is None:
num_bytes = max_packet_size
return serial_port.read(num_bytes)
# Code dependant on this connection method
@ -138,6 +149,9 @@ def sendByte_win(byte_to_send, num_bytes):
# Code dependant on this connection method
def sendList_win(data, chunk_size=8):
#Why? Idk. But it fixes it... :/
if(chunk_size > 0x3C):
chunk_size = 0x3C
num_iters = int(len(data)/chunk_size)
for i in range(num_iters):
p.write(bytes(data[i*chunk_size:(i+1)*chunk_size]))
@ -146,12 +160,56 @@ def sendList_win(data, chunk_size=8):
if (num_iters*chunk_size) != len(data):
p.write(bytes(data[num_iters*chunk_size:]))
def receiveByte_win(num_bytes):
recv = int.from_bytes(p.read(size=num_bytes), byteorder='big')
# Code dependant on this connection method
# The original was so slow, I had to rewrite it a bit to make it work for time sensitive applications
def read_win(self, size=None):
if not self.is_open:
return None
rx = [self._rxremaining]
length = len(self._rxremaining)
self._rxremaining = b''
end_timeout = time.time() + (self.timeout or 0.2)
if size:
super(ComPort, self).set_timeout(self._ep_in, (self.timeout or 0.2) * 10)
while length < size:
c = super(ComPort, self).read(self._ep_in, size-length)
if c is not None and len(c):
rx.append(c)
length += len(c)
if len(c) == self.maximum_packet_size:
end_timeout += (self.timeout or 0.2)
if time.time() > end_timeout:
break
else:
super(ComPort, self).set_timeout(self._ep_in, (self.timeout or 0.2))
while True:
c = super(ComPort, self).read(self._ep_in, self.maximum_packet_size)
if c is not None and len(c):
rx.append(c)
length += len(c)
if len(c) == self.maximum_packet_size:
end_timeout += (self.timeout or 0.2)
else:
break
else:
break
if time.time() > end_timeout:
break
chunk = b''.join(rx)
if size and len(chunk) >= size:
if self._rxremaining:
self._rxremaining = chunk[size:] + self._rxremaining
else:
self._rxremaining = chunk[size:]
chunk = chunk[0:size]
return chunk
def receiveByte_win(num_bytes=None):
recv = int.from_bytes(read_win(p, size=num_bytes), byteorder='big')
return recv
def receiveByte_raw_win(num_bytes):
return p.read(size=num_bytes)
def receiveByte_raw_win(num_bytes=None):
return read_win(p, size=num_bytes)
# Things for the USB connection part
def exit_gracefully():