Skip to content

Migrate m3-python to Python 3.9 #28

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JHLeeECE wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Migrate m3-python to Python 3.9 #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0b5e508
fix(ice): migrate serial protocol to Python 3.9
JHLeeECE Aug 17, 2026
3c9f14d
fix(gdb): preserve remote packets on Python 3.9
JHLeeECE Aug 17, 2026
c120f2f
fix(common): normalize binary program handling
JHLeeECE Aug 17, 2026
34fc3dd
fix(mbus): preserve payload bytes on Python 3.9
JHLeeECE Aug 17, 2026
47c4d67
build: require CPython 3.9
JHLeeECE Aug 17, 2026
50422b0
readme update
JHLeeECE Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,17 @@ M3-Python

Utilities and software libraries for the [M3 ecosystem](http://cubeworks.us) and interfacing with the [ICE board](http://mbus.io/ice.html).

M3-Python supports CPython 3.9. Install the local checkout with:

```bash
$ pip install m3
$ # Program the board via the optical interface:
$ m3_ice goc flash program.bin
```

The `m3_ice` utility should handle most use cases, however users are free to write their scripts against the `ice` library directly.
The simulator and integration suite require `socat` and Unix PTYs. Run them on Linux, or use WSL instead of native Windows, with `tox -e py39`.

The `m3_ice` utility should handle most use cases, however users are free to write their scripts against the `m3.ice` library directly.
Developers are encouraged to consider some of the higher-level interfaces provided by `m3_common`.

m3_ice
Expand Down
63 changes: 27 additions & 36 deletions m3/ice.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,12 @@

################################################################################

# Coerce Py2k to act more like Py3k
from __future__ import (absolute_import, division, print_function, unicode_literals)
from builtins import (
ascii, bytes, chr, dict, filter, hex, input, int, isinstance, list, map,
next, object, oct, open, pow, range, round, str, super, zip,
)

import binascii
from copy import copy
from copy import deepcopy
import errno
import functools
import logging
import socket
import struct
import sys
Expand All @@ -22,7 +16,7 @@

try:
from . import m3_logging
except:
except ImportError:
import m3_logging
logger = m3_logging.getLogger(__name__)

Expand Down Expand Up @@ -52,7 +46,7 @@

class ICE(object):
VERSIONS = ((0,1),(0,2),(0,3),(0,4),(0,5))
ONEYEAR = 365 * 24 * 60 * 60
ONEYEAR = min(365 * 24 * 60 * 60, threading.TIMEOUT_MAX)

class ICE_Error(Exception):
'''
Expand Down Expand Up @@ -223,7 +217,7 @@ def find_baud(self, serial_device):
with serial.Serial(serial_device, baudrates[0],
timeout=0.05 ) as tmpSerial:

if not tmpSerial.isOpen():
if not tmpSerial.is_open:
raise self.ICE_Error("Failed to connect to temporary serial device")

for baudrate in baudrates:
Expand Down Expand Up @@ -266,7 +260,7 @@ def connect(self, serial_device, baudrate=115200):
logger.warn("Skipping baudrate?")
self.dev = serial.Serial(serial_device, timeout=0.5)

if self.dev.isOpen():
if self.dev.is_open:
logger.info("Connected to serial device at " + self.dev.portstr +
" at " + str(baudrate) + " baud")
else:
Expand Down Expand Up @@ -320,14 +314,14 @@ def spawn_handler(self, msg_type, event_id, length, msg):
logger.warn("WARNING: No handler registered for message type: " +
str(msg_type))
logger.warn("Known Types:")
for t,f in self.msg_handler.iteritems():
for t,f in self.msg_handler.items():
logger.warn("%s\t%s" % (t, str(f)))
logger.warn(" Dropping packet:")
logger.warn("")
logger.warn(" Type: %s" % (msg_type))
logger.warn("Event ID: %d" % (event_id))
logger.warn(" Length: %d" % (length))
logger.warn(" Message:" + msg.encode('hex'))
logger.warn(" Message:" + msg.hex())
except Exception as e:
logger.warn("Unhandled exception trying to report unknown message.")
logger.warn(str(e))
Expand All @@ -345,7 +339,7 @@ def spawn_handler(self, msg_type, event_id, length, msg):
logger.debug(" Type: %s" % (msg_type))
logger.debug("Event ID: %d" % (event_id))
logger.debug(" Length: %d" % (length))
logger.debug(" Message:" + msg.encode('hex'))
logger.debug(" Message:" + msg.hex())
except Exception as e:
logger.debug("Unhandled exception trying to report unknown message.")
logger.debug(str(e))
Expand All @@ -363,11 +357,11 @@ def useful_read(self, length, check_timeout = False):
rxBuf += rx

assert len(rxBuf) == length
logger.debug('Raw Read: ' + binascii.hexlify(rxBuf) )
logger.debug('Raw Read: ' + rxBuf.hex())
return rxBuf

def communicator(self):
while not self.communicator_stop_request.isSet():
while not self.communicator_stop_request.is_set():
try:
# Read has a timeout of .1 s. Polling is the easiest way to
# do x-platform cancellation
Expand All @@ -376,16 +370,13 @@ def communicator(self):
continue
except (serial.SerialException, OSError):
break
msg_type = ord(msg_type)
event_id = ord(event_id)
length = ord(length)
#print("Got msg type", msg_type, chr(msg_type), length)
try:
msg = self.useful_read(length, check_timeout = True)
except self.TimeoutError:
logger.warn("Timeout error occured, skipping rest of packet!")
continue
#print(msg.encode('hex'))
#print(msg.hex())

if event_id == self.last_event_id:
logger.warn("WARNING: Duplicate event_id! THIS IS A BUG [somewhere]!!")
Expand All @@ -394,7 +385,7 @@ def communicator(self):
logger.warn(" Type: %d" % (msg_type))
logger.warn("Event ID: %d" % (event_id))
logger.warn(" Length: %d" % (length))
logger.warn(" Message:" + msg.encode('hex'))
logger.warn(" Message:" + msg.hex())
else:
self.last_event_id = event_id

Expand All @@ -413,7 +404,7 @@ def communicator(self):
logger.warn(" Type: %s" % (["ACK","NAK"][msg_type]))
logger.warn("Event ID: %d" % (event_id))
logger.warn(" Length: %d" % (length))
logger.warn(" Message:" + msg.encode('hex'))
logger.warn(" Message:" + msg.hex())
else:
msg_type = chr(msg_type)
logger.debug("Got an async message of type: " + msg_type)
Expand Down Expand Up @@ -577,9 +568,9 @@ def common_bB_formatter(self, msg_type, event_id, length, msg, b_type):
try:
logger.warn("No handler registered for B++ (formatted, snooped MBus) messages")
logger.warn("Dropping message:")
logger.warn("\taddr: " + binascii.hexlify(addr))
logger.warn("\tdata: " + binascii.hexlify(data))
logger.warn("\tstat: " + binascii.hexlify(cb))
logger.warn("\taddr: " + addr.hex())
logger.warn("\tdata: " + data.hex())
logger.warn("\tstat: " + msg[-1:].hex())
logger.warn("")
except Exception as e:
logger.warn("Unhandled exception trying to report missing B++ handler.")
Expand Down Expand Up @@ -647,8 +638,8 @@ def negotiate_version(self):

logger.debug("Sending version probe")
resp = self.send_message_until_acked('V')
if (len(resp) is 0) or (len(resp) % 2):
raise self.FormatError("Version response: " + resp)
if (len(resp) == 0) or (len(resp) % 2):
raise self.FormatError("Version response: {}".format(resp))

logger.info("This ICE board supports versions...")
self.major = None
Expand Down Expand Up @@ -748,7 +739,7 @@ def ice_query_capabilities(self):
characters from the ICE board, which requires the caller to know the
ICE protocol.
'''
resp = self.send_message_until_acked('?', struct.pack("B", ord('?')))
resp = self.send_message_until_acked('?', struct.pack("B", ord('?'))).decode('ascii')
self.capabilities = resp
return resp

Expand Down Expand Up @@ -890,7 +881,7 @@ def goc_ein_get_freq_divisor_max_0_2(self):
resp = self.send_message_until_acked('O', struct.pack("B", ord('c')))
if len(resp) != 3:
raise self.FormatError("Wrong response length from `Oc': " + str(resp))
setting = struct.unpack("!I", "\x00"+resp)[0]
setting = struct.unpack("!I", b"\x00"+resp)[0]
return setting

@min_proto_version("0.3")
Expand All @@ -911,7 +902,7 @@ def goc_ein_get_freq_divisor(self):
@max_proto_version("0.2")
def goc_ein_set_freq_divisor_max_0_2(self, divisor):
packed = struct.pack("!I", divisor)
if packed[0] != '\x00':
if packed[0] != 0:
raise self.ParameterError("Out of range.")
msg = struct.pack("B", ord('c')) + packed[1:]
self.send_message_until_acked('o', msg)
Expand Down Expand Up @@ -1015,7 +1006,7 @@ def _goc_freq_in_hz_to_divisor(self, freq_in_hz):
NOMINAL = 2e6
else:
NOMINAL = 4e6
return NOMINAL / freq_in_hz;
return int(NOMINAL / freq_in_hz)

@min_proto_version("0.1")
@capability('o')
Expand Down Expand Up @@ -1099,7 +1090,7 @@ def i2c_get_speed(self):
raise self.FormatError
return struct.unpack("B", msg)[0] * 2

ret = ord(msg[0])
ret = msg[0]
msg = msg[1:]
if ret == errno.ENODEV:
# XXX Generalize me w.r.t. version?
Expand Down Expand Up @@ -1133,7 +1124,7 @@ def i2c_set_speed(self, speed):
if ack == 0:
return speed

ret = ord(msg[0])
ret = msg[0]
msg = msg[1:]
if ret == errno.EINVAL:
raise self.ICE_Error("ICE reports: Invalid argument.")
Expand Down Expand Up @@ -1683,7 +1674,7 @@ def _gpio_get_level_0_2(self):
resp = self.send_message_until_acked('G', struct.pack('B', ord('l')))
if len(resp) != 3:
raise self.FormatError("Bad response from `Gl':" + str(resp))
high,mid,low = map(ord, resp)
high,mid,low = resp
return low | (mid << 8) | (high << 16)

@min_proto_version("0.2")
Expand All @@ -1697,7 +1688,7 @@ def _gpio_get_direction_0_2(self):
resp = self.send_message_until_acked('G', struct.pack('B', ord('d')))
if len(resp) != 3:
raise self.FormatError("Bad response from `Gd#':" + str(resp))
high,mid,low = map(ord, resp)
high,mid,low = resp
return low | (mid << 8) | (high << 16)

@min_proto_version("0.2")
Expand Down Expand Up @@ -1747,7 +1738,7 @@ def gpio_get_interrupt_enable_mask(self):
resp = self.send_message_until_acked('G', struct.pack('B', ord('i')))
if len(resp) != 3:
raise self.FormatError("Bad response from `Gi':" + str(resp))
high,mid,low = map(ord, resp)
high,mid,low = resp
return low | (mid << 8) | (high << 16)

@min_proto_version("0.2")
Expand Down
Loading