Repository navigation
Expand file tree
/
Copy pathi2c.py
More file actions
122 lines (86 loc) · 3.49 KB
/
Copy pathi2c.py
File metadata and controls
122 lines (86 loc) · 3.49 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import sys
from contextlib import closing
import posix
from fcntl import ioctl
from .linux_i2c import *
import ctypes
# Try to make this support Python 2 as well.
assert sys.version_info.major >= 3, __name__ + " is only supported on Python 3"
DEFAULT_BUS = 1
class I2CMaster(object):
"""Performs I2C I/O transactions on an I2C bus.
Transactions are performed by passing one or more I2C I/O messages
to the transaction method of the I2CMaster. I2C I/O messages are
created with the reading, reading_into, writing and writing_bytes
functions defined in the microstackcommon.i2c module.
An I2CMaster acts as a context manager, allowing it to be used in a
with statement. The I2CMaster's file descriptor is closed at
the end of the with statement and the instance cannot be used for
further I/O.
For example:
from microstackcommon.i2c import I2CMaster, writing
with I2CMaster() as i2c:
i2c.transaction(
writing(0x20, bytes([0x01, 0xFF])))
"""
def __init__(self, bus=DEFAULT_BUS, extra_open_flags=0):
"""
bus -- the number of the bus (default is
the bus on the Raspberry Pi accessible
via the header pins).
"""
self.bus = bus
def __enter__(self):
self.open()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.close()
def open(self, extra_open_flags=0):
"""Opens the bus device.
Arguments:
extra_open_flags -- extra flags passed to posix.open when
opening the I2C bus device file (default 0;
e.g. no extra flags).
"""
self.fd = posix.open("/dev/i2c-{}".format(self.bus),
posix.O_RDWR | extra_open_flags)
def close(self):
"""
Closes the I2C bus device.
"""
posix.close(self.fd)
def transaction(self, *msgs):
"""
Perform an I2C I/O transaction.
Arguments:
*msgs -- I2C messages created by one of the reading, reading_into,
writing or writing_bytes functions.
Returns: a list of byte sequences, one for each read operation
performed.
"""
msg_count = len(msgs)
msg_array = (i2c_msg*msg_count)(*msgs)
ioctl_arg = i2c_rdwr_ioctl_data(msgs=msg_array, nmsgs=msg_count)
ioctl(self.fd, I2C_RDWR, ioctl_arg)
return [i2c_msg_to_bytes(m) for m in msgs if (m.flags & I2C_M_RD)]
def reading(addr, n_bytes):
"""An I2C I/O message that reads n_bytes bytes of data"""
return reading_into(addr, ctypes.create_string_buffer(n_bytes))
def reading_into(addr, buf):
"""An I2C I/O message that reads into an existing ctypes string buffer."""
return _new_i2c_msg(addr, I2C_M_RD, buf)
def writing_bytes(addr, *bytes):
"""An I2C I/O message that writes one or more bytes of data.
Each byte is passed as an argument to this function.
"""
return writing(addr, bytes)
def writing(addr, byte_seq):
"""An I2C I/O message that writes one or more bytes of data.
The bytes are passed to this function as a sequence.
"""
buf = bytes(byte_seq)
return _new_i2c_msg(addr, 0, ctypes.create_string_buffer(buf, len(buf)))
def _new_i2c_msg(addr, flags, buf):
return i2c_msg(addr=addr, flags=flags, len=ctypes.sizeof(buf), buf=buf)
def i2c_msg_to_bytes(m):
return ctypes.string_at(m.buf, m.len)