|
| 1 | +# Test machine.I2CTarget. |
| 2 | +# |
| 3 | +# IMPORTANT: This test requires hardware connections: a SoftI2C instance must be |
| 4 | +# wired to a hardware I2C target. See pin definitions below. |
| 5 | + |
| 6 | +import sys |
| 7 | + |
| 8 | +try: |
| 9 | + from machine import Pin, SoftI2C, I2CTarget |
| 10 | +except ImportError: |
| 11 | + print("SKIP") |
| 12 | + raise SystemExit |
| 13 | + |
| 14 | +import unittest |
| 15 | + |
| 16 | +ADDR = 67 |
| 17 | + |
| 18 | +kwargs_target = {} |
| 19 | + |
| 20 | +# Configure pins based on the target. |
| 21 | +if sys.platform == "alif" and sys.implementation._build == "ALIF_ENSEMBLE": |
| 22 | + args_controller = {"scl": "P1_1", "sda": "P1_0"} |
| 23 | + args_target = (0,) # on pins P0_3/P0_2 |
| 24 | +elif sys.platform == "esp32": |
| 25 | + args_controller = {"scl": 5, "sda": 6} |
| 26 | + args_target = (0,) # on pins 9/8 for C3 and S3, 18/19 for others |
| 27 | + kwargs_target = {"scl": 9, "sda": 8} |
| 28 | +elif sys.platform == "rp2": |
| 29 | + args_controller = {"scl": 5, "sda": 4} |
| 30 | + args_target = (1,) |
| 31 | +elif sys.platform == "pyboard": |
| 32 | + if sys.implementation._build == "NUCLEO_WB55": |
| 33 | + args_controller = {"scl": "B8", "sda": "B9"} |
| 34 | + args_target = (3,) |
| 35 | + else: |
| 36 | + args_controller = {"scl": "X1", "sda": "X2"} |
| 37 | + args_target = ("X",) |
| 38 | +elif "zephyr-nucleo_wb55rg" in sys.implementation._machine: |
| 39 | + # PB8=I2C1_SCL, PB9=I2C1_SDA (on Arduino header D15/D14) |
| 40 | + # PC0=I2C3_SCL, PC1=I2C3_SDA (on Arduino header A0/A1) |
| 41 | + args_controller = {"scl": Pin(("gpiob", 8)), "sda": Pin(("gpiob", 9))} |
| 42 | + args_target = ("i2c3",) |
| 43 | +elif "zephyr-rpi_pico" in sys.implementation._machine: |
| 44 | + args_controller = {"scl": Pin(("gpio0", 5)), "sda": Pin(("gpio0", 4))} |
| 45 | + args_target = ("i2c1",) # on gpio7/gpio6 |
| 46 | +elif sys.platform == "mimxrt": |
| 47 | + if "Teensy" in sys.implementation._machine: |
| 48 | + args_controller = {"scl": "A6", "sda": "A3"} # D20/D17 |
| 49 | + else: |
| 50 | + args_controller = {"scl": "D0", "sda": "D1"} |
| 51 | + args_target = (0,) # pins 19/18 On Teensy 4.x |
| 52 | +elif sys.platform == "samd": |
| 53 | + args_controller = {"scl": "D5", "sda": "D1"} |
| 54 | + args_target = () |
| 55 | +else: |
| 56 | + print("Please add support for this test on this platform.") |
| 57 | + raise SystemExit |
| 58 | + |
| 59 | + |
| 60 | +def config_pull_up(): |
| 61 | + Pin(args_controller["scl"], Pin.OPEN_DRAIN, Pin.PULL_UP) |
| 62 | + Pin(args_controller["sda"], Pin.OPEN_DRAIN, Pin.PULL_UP) |
| 63 | + |
| 64 | + |
| 65 | +class TestMemory(unittest.TestCase): |
| 66 | + @classmethod |
| 67 | + def setUpClass(cls): |
| 68 | + cls.mem = bytearray(8) |
| 69 | + cls.i2c = SoftI2C(**args_controller) |
| 70 | + cls.i2c_target = I2CTarget(*args_target, **kwargs_target, addr=ADDR, mem=cls.mem) |
| 71 | + config_pull_up() |
| 72 | + |
| 73 | + @classmethod |
| 74 | + def tearDownClass(cls): |
| 75 | + cls.i2c_target.deinit() |
| 76 | + |
| 77 | + def test_scan(self): |
| 78 | + self.assertIn(ADDR, self.i2c.scan()) |
| 79 | + |
| 80 | + def test_write(self): |
| 81 | + self.mem[:] = b"01234567" |
| 82 | + self.i2c.writeto_mem(ADDR, 0, b"test") |
| 83 | + self.assertEqual(self.mem, bytearray(b"test4567")) |
| 84 | + self.i2c.writeto_mem(ADDR, 4, b"TEST") |
| 85 | + self.assertEqual(self.mem, bytearray(b"testTEST")) |
| 86 | + |
| 87 | + def test_write_wrap(self): |
| 88 | + self.mem[:] = b"01234567" |
| 89 | + self.i2c.writeto_mem(ADDR, 6, b"test") |
| 90 | + self.assertEqual(self.mem, bytearray(b"st2345te")) |
| 91 | + |
| 92 | + @unittest.skipIf(sys.platform == "esp32", "write lengths larger than buffer unsupported") |
| 93 | + def test_write_wrap_large(self): |
| 94 | + self.mem[:] = b"01234567" |
| 95 | + self.i2c.writeto_mem(ADDR, 0, b"testTESTmore") |
| 96 | + self.assertEqual(self.mem, bytearray(b"moreTEST")) |
| 97 | + |
| 98 | + def test_read(self): |
| 99 | + self.mem[:] = b"01234567" |
| 100 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 0, 4), b"0123") |
| 101 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 4, 4), b"4567") |
| 102 | + |
| 103 | + def test_read_wrap(self): |
| 104 | + self.mem[:] = b"01234567" |
| 105 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 0, 4), b"0123") |
| 106 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 2, 4), b"2345") |
| 107 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 6, 4), b"6701") |
| 108 | + |
| 109 | + @unittest.skipIf(sys.platform == "esp32", "read lengths larger than buffer unsupported") |
| 110 | + def test_read_wrap_large(self): |
| 111 | + self.mem[:] = b"01234567" |
| 112 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 0, 12), b"012345670123") |
| 113 | + |
| 114 | + def test_write_read(self): |
| 115 | + self.mem[:] = b"01234567" |
| 116 | + self.assertEqual(self.i2c.writeto(ADDR, b"\x02"), 1) |
| 117 | + self.assertEqual(self.i2c.readfrom(ADDR, 4), b"2345") |
| 118 | + |
| 119 | + @unittest.skipIf(sys.platform == "esp32", "read after read unsupported") |
| 120 | + def test_write_read_read(self): |
| 121 | + self.mem[:] = b"01234567" |
| 122 | + self.assertEqual(self.i2c.writeto(ADDR, b"\x02"), 1) |
| 123 | + self.assertEqual(self.i2c.readfrom(ADDR, 4), b"2345") |
| 124 | + self.assertEqual(self.i2c.readfrom(ADDR, 4), b"7012") |
| 125 | + |
| 126 | + |
| 127 | +@unittest.skipUnless(hasattr(I2CTarget, "IRQ_END_READ"), "IRQ unsupported") |
| 128 | +class TestMemoryIRQ(unittest.TestCase): |
| 129 | + @staticmethod |
| 130 | + def irq_handler(i2c_target): |
| 131 | + flags = i2c_target.irq().flags() |
| 132 | + TestMemoryIRQ.events[TestMemoryIRQ.num_events] = flags |
| 133 | + TestMemoryIRQ.events[TestMemoryIRQ.num_events + 1] = i2c_target.memaddr |
| 134 | + TestMemoryIRQ.num_events += 2 |
| 135 | + |
| 136 | + @classmethod |
| 137 | + def setUpClass(cls): |
| 138 | + cls.mem = bytearray(8) |
| 139 | + cls.events = [0] * 8 |
| 140 | + cls.num_events = 0 |
| 141 | + cls.i2c = SoftI2C(**args_controller) |
| 142 | + cls.i2c_target = I2CTarget(*args_target, **kwargs_target, addr=ADDR, mem=cls.mem) |
| 143 | + cls.i2c_target.irq(TestMemoryIRQ.irq_handler) |
| 144 | + config_pull_up() |
| 145 | + |
| 146 | + @classmethod |
| 147 | + def tearDownClass(cls): |
| 148 | + cls.i2c_target.deinit() |
| 149 | + |
| 150 | + @unittest.skipIf(sys.platform == "esp32", "scan doesn't trigger IRQ_END_WRITE") |
| 151 | + def test_scan(self): |
| 152 | + TestMemoryIRQ.num_events = 0 |
| 153 | + self.i2c.scan() |
| 154 | + self.assertEqual(self.events[: self.num_events], [I2CTarget.IRQ_END_WRITE, 0]) |
| 155 | + |
| 156 | + def test_write(self): |
| 157 | + TestMemoryIRQ.num_events = 0 |
| 158 | + self.mem[:] = b"01234567" |
| 159 | + self.i2c.writeto_mem(ADDR, 2, b"test") |
| 160 | + self.assertEqual(self.mem, bytearray(b"01test67")) |
| 161 | + self.assertEqual(self.events[: self.num_events], [I2CTarget.IRQ_END_WRITE, 2]) |
| 162 | + |
| 163 | + def test_read(self): |
| 164 | + TestMemoryIRQ.num_events = 0 |
| 165 | + self.mem[:] = b"01234567" |
| 166 | + self.assertEqual(self.i2c.readfrom_mem(ADDR, 2, 4), b"2345") |
| 167 | + self.assertEqual(self.events[: self.num_events], [I2CTarget.IRQ_END_READ, 2]) |
| 168 | + |
| 169 | + |
| 170 | +@unittest.skipUnless(hasattr(I2CTarget, "IRQ_WRITE_REQ"), "IRQ unsupported") |
| 171 | +@unittest.skipIf(sys.platform == "mimxrt", "not working") |
| 172 | +@unittest.skipIf(sys.platform == "pyboard", "can't queue more than one byte") |
| 173 | +@unittest.skipIf(sys.platform == "samd", "not working") |
| 174 | +@unittest.skipIf(sys.platform == "zephyr", "must call readinto/write in IRQ handler") |
| 175 | +class TestPolling(unittest.TestCase): |
| 176 | + @staticmethod |
| 177 | + def irq_handler(i2c_target, buf=bytearray(1)): |
| 178 | + flags = i2c_target.irq().flags() |
| 179 | + if flags & I2CTarget.IRQ_READ_REQ: |
| 180 | + i2c_target.write(b"0123") |
| 181 | + |
| 182 | + @classmethod |
| 183 | + def setUpClass(cls): |
| 184 | + cls.i2c = SoftI2C(**args_controller) |
| 185 | + cls.i2c_target = I2CTarget(*args_target, addr=ADDR) |
| 186 | + cls.i2c_target.irq( |
| 187 | + TestPolling.irq_handler, |
| 188 | + I2CTarget.IRQ_WRITE_REQ | I2CTarget.IRQ_READ_REQ, |
| 189 | + hard=True, |
| 190 | + ) |
| 191 | + config_pull_up() |
| 192 | + |
| 193 | + @classmethod |
| 194 | + def tearDownClass(cls): |
| 195 | + cls.i2c_target.deinit() |
| 196 | + |
| 197 | + def test_read(self): |
| 198 | + # Can't write data up front, must wait until IRQ_READ_REQ. |
| 199 | + # self.assertEqual(self.i2c_target.write(b"abcd"), 4) |
| 200 | + self.assertEqual(self.i2c.readfrom(ADDR, 4), b"0123") |
| 201 | + |
| 202 | + def test_write(self): |
| 203 | + # Can do the read outside the IRQ, but requires IRQ_WRITE_REQ trigger to be set. |
| 204 | + self.assertEqual(self.i2c.writeto(ADDR, b"0123"), 4) |
| 205 | + buf = bytearray(8) |
| 206 | + self.assertEqual(self.i2c_target.readinto(buf), 4) |
| 207 | + self.assertEqual(buf, b"0123\x00\x00\x00\x00") |
| 208 | + |
| 209 | + |
| 210 | +@unittest.skipUnless(hasattr(I2CTarget, "IRQ_ADDR_MATCH_READ"), "IRQ unsupported") |
| 211 | +class TestIRQ(unittest.TestCase): |
| 212 | + @staticmethod |
| 213 | + def irq_handler(i2c_target, buf=bytearray(1)): |
| 214 | + flags = i2c_target.irq().flags() |
| 215 | + TestIRQ.events[TestIRQ.num_events] = flags |
| 216 | + TestIRQ.num_events += 1 |
| 217 | + if flags & I2CTarget.IRQ_READ_REQ: |
| 218 | + i2c_target.write(b"Y") |
| 219 | + if flags & I2CTarget.IRQ_WRITE_REQ: |
| 220 | + i2c_target.readinto(buf) |
| 221 | + TestIRQ.events[TestIRQ.num_events] = buf[0] |
| 222 | + TestIRQ.num_events += 1 |
| 223 | + |
| 224 | + @classmethod |
| 225 | + def setUpClass(cls): |
| 226 | + cls.events = [0] * 8 |
| 227 | + cls.num_events = 0 |
| 228 | + cls.i2c = SoftI2C(**args_controller) |
| 229 | + cls.i2c_target = I2CTarget(*args_target, addr=ADDR) |
| 230 | + cls.i2c_target.irq( |
| 231 | + TestIRQ.irq_handler, |
| 232 | + I2CTarget.IRQ_ADDR_MATCH_READ |
| 233 | + | I2CTarget.IRQ_ADDR_MATCH_WRITE |
| 234 | + | I2CTarget.IRQ_WRITE_REQ |
| 235 | + | I2CTarget.IRQ_READ_REQ |
| 236 | + | I2CTarget.IRQ_END_READ |
| 237 | + | I2CTarget.IRQ_END_WRITE, |
| 238 | + hard=True, |
| 239 | + ) |
| 240 | + config_pull_up() |
| 241 | + |
| 242 | + @classmethod |
| 243 | + def tearDownClass(cls): |
| 244 | + cls.i2c_target.deinit() |
| 245 | + |
| 246 | + def test_scan(self): |
| 247 | + TestIRQ.num_events = 0 |
| 248 | + self.i2c.scan() |
| 249 | + self.assertEqual( |
| 250 | + self.events[: self.num_events], |
| 251 | + [ |
| 252 | + I2CTarget.IRQ_ADDR_MATCH_WRITE, |
| 253 | + I2CTarget.IRQ_END_WRITE, |
| 254 | + ], |
| 255 | + ) |
| 256 | + |
| 257 | + def test_write(self): |
| 258 | + TestIRQ.num_events = 0 |
| 259 | + self.i2c.writeto(ADDR, b"XYZ") |
| 260 | + self.assertEqual( |
| 261 | + self.events[: self.num_events], |
| 262 | + [ |
| 263 | + I2CTarget.IRQ_ADDR_MATCH_WRITE, |
| 264 | + I2CTarget.IRQ_WRITE_REQ, |
| 265 | + ord(b"X"), |
| 266 | + I2CTarget.IRQ_WRITE_REQ, |
| 267 | + ord(b"Y"), |
| 268 | + I2CTarget.IRQ_WRITE_REQ, |
| 269 | + ord(b"Z"), |
| 270 | + I2CTarget.IRQ_END_WRITE, |
| 271 | + ], |
| 272 | + ) |
| 273 | + |
| 274 | + def test_read(self): |
| 275 | + TestIRQ.num_events = 0 |
| 276 | + self.assertEqual(self.i2c.readfrom(ADDR, 1), b"Y") |
| 277 | + self.assertEqual( |
| 278 | + self.events[: self.num_events], |
| 279 | + [ |
| 280 | + I2CTarget.IRQ_ADDR_MATCH_READ, |
| 281 | + I2CTarget.IRQ_READ_REQ, |
| 282 | + I2CTarget.IRQ_READ_REQ, |
| 283 | + I2CTarget.IRQ_END_READ, |
| 284 | + ], |
| 285 | + ) |
| 286 | + |
| 287 | + def test_write_read(self): |
| 288 | + TestIRQ.num_events = 0 |
| 289 | + self.i2c.writeto(ADDR, b"X", False) |
| 290 | + self.assertEqual(self.i2c.readfrom(ADDR, 1), b"Y") |
| 291 | + self.assertEqual( |
| 292 | + self.events[: self.num_events], |
| 293 | + [ |
| 294 | + I2CTarget.IRQ_ADDR_MATCH_WRITE, |
| 295 | + I2CTarget.IRQ_WRITE_REQ, |
| 296 | + ord(b"X"), |
| 297 | + I2CTarget.IRQ_END_WRITE, |
| 298 | + I2CTarget.IRQ_ADDR_MATCH_READ, |
| 299 | + I2CTarget.IRQ_READ_REQ, |
| 300 | + I2CTarget.IRQ_READ_REQ, |
| 301 | + I2CTarget.IRQ_END_READ, |
| 302 | + ], |
| 303 | + ) |
| 304 | + |
| 305 | + |
| 306 | +if __name__ == "__main__": |
| 307 | + unittest.main() |
0 commit comments