import atexit
import codecs
-VERSION = "1.8"
+VERSION = "1.9"
exceptions = True
[PI_SPI_XFER_FAILED , "SPI xfer/read/write failed"],
]
+class _socklock:
+ """
+ A class to store socket and lock.
+ """
+ def __init__(self):
+ self.s = None
+ self.l = threading.Lock()
+
class error(Exception):
"""pigpio module exception"""
def __init__(self, value):
else:
return v
-def _pigpio_command(sock, cmd, p1, p2):
+def _pigpio_command(sl, cmd, p1, p2):
"""
Runs a pigpio socket command.
- sock:= command socket.
- cmd:= the command to be executed.
- p1:= command parameter 1 (if applicable).
+ sl:= command socket and lock.
+ cmd:= the command to be executed.
+ p1:= command parameter 1 (if applicable).
p2:= command parameter 2 (if applicable).
"""
- sock.send(struct.pack('IIII', cmd, p1, p2, 0))
- dummy, res = struct.unpack('12sI', sock.recv(16))
+ sl.l.acquire()
+ sl.s.send(struct.pack('IIII', cmd, p1, p2, 0))
+ dummy, res = struct.unpack('12sI', sl.s.recv(16))
+ sl.l.release()
return res
-def _pigpio_command_ext(sock, cmd, p1, p2, p3, extents):
+def _pigpio_command_ext(sl, cmd, p1, p2, p3, extents):
"""
Runs an extended pigpio socket command.
- sock:= command socket.
+ sl:= command socket and lock.
cmd:= the command to be executed.
p1:= command parameter 1 (if applicable).
p2:= command parameter 2 (if applicable).
ext.extend(_b(x))
else:
ext.extend(x)
- sock.sendall(ext)
- dummy, res = struct.unpack('12sI', sock.recv(16))
+ sl.l.acquire()
+ sl.s.sendall(ext)
+ dummy, res = struct.unpack('12sI', sl.s.recv(16))
+ sl.l.release()
return res
class _callback_ADT:
"""Initialises notifications."""
threading.Thread.__init__(self)
self.control = control
+ self.sl = _socklock()
self.go = False
self.daemon = True
self.monitor = 0
self.callbacks = []
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.sock.connect((host, port))
- self.handle = _pigpio_command(self.sock, _PI_CMD_NOIB, 0, 0)
+ self.sl.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.sl.s.connect((host, port))
+ self.handle = _pigpio_command(self.sl, _PI_CMD_NOIB, 0, 0)
self.go = True
self.start()
"""Stops notifications."""
if self.go:
self.go = False
- self.sock.send(struct.pack('IIII', _PI_CMD_NC, self.handle, 0, 0))
+ self.sl.s.send(struct.pack('IIII', _PI_CMD_NC, self.handle, 0, 0))
def append(self, callb):
"""Adds a callback to the notification thread."""
while self.go:
- buf = self.sock.recv(MSG_SIZ)
+ buf = self.sl.s.recv(MSG_SIZ)
while self.go and len(buf) < MSG_SIZ:
- buf += self.sock.recv(MSG_SIZ-len(buf))
+ buf += self.sl.s.recv(MSG_SIZ-len(buf))
if self.go:
seq, flags, tick, level = (struct.unpack('HHII', buf))
if cb.gpio == gpio:
cb.func(cb.gpio, TIMEOUT, tick)
- self.sock.close()
+ self.sl.s.close()
class _callback:
"""A class to provide gpio level change callbacks."""
def _rxbuf(self, count):
"""Returns count bytes from the command socket."""
- ext = bytearray(self._control.recv(count))
+ ext = bytearray(self.sl.s.recv(count))
while len(ext) < count:
- ext.extend(self._control.recv(count - len(ext)))
+ ext.extend(self.sl.s.recv(count - len(ext)))
return ext
def set_mode(self, gpio, mode):
pi.set_mode(24, pigpio.ALT2) # gpio 24 as ALT2
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_MODES, gpio, mode))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_MODES, gpio, mode))
def get_mode(self, gpio):
"""
4
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_MODEG, gpio, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_MODEG, gpio, 0))
def set_pull_up_down(self, gpio, pud):
"""
pi.set_pull_up_down(24, pigpio.PUD_DOWN)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PUD, gpio, pud))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PUD, gpio, pud))
def read(self, gpio):
"""
1
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_READ, gpio, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_READ, gpio, 0))
def write(self, gpio, level):
"""
1
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WRITE, gpio, level))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WRITE, gpio, level))
def set_PWM_dutycycle(self, user_gpio, dutycycle):
"""
...
"""
return _u2i(_pigpio_command(
- self._control, _PI_CMD_PWM, user_gpio, int(dutycycle)))
+ self.sl, _PI_CMD_PWM, user_gpio, int(dutycycle)))
def set_PWM_range(self, user_gpio, range_):
"""
pi.set_PWM_range(9, 3000) # now 750 1/4, 1500 1/2, 2250 3/4 on
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PRS, user_gpio, range_))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PRS, user_gpio, range_))
def get_PWM_range(self, user_gpio):
"""
500
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PRG, user_gpio, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PRG, user_gpio, 0))
def get_PWM_real_range(self, user_gpio):
"""
250
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PRRG, user_gpio, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PRRG, user_gpio, 0))
def set_PWM_frequency(self, user_gpio, frequency):
"""
8000
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PFS, user_gpio, frequency))
+ return _u2i(
+ _pigpio_command(self.sl, _PI_CMD_PFS, user_gpio, frequency))
def get_PWM_frequency(self, user_gpio):
"""
800
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PFG, user_gpio, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PFG, user_gpio, 0))
def set_servo_pulsewidth(self, user_gpio, pulsewidth):
"""
...
"""
return _u2i(_pigpio_command(
- self._control, _PI_CMD_SERVO, user_gpio, int(pulsewidth)))
+ self.sl, _PI_CMD_SERVO, user_gpio, int(pulsewidth)))
def notify_open(self):
"""
pi.notify_begin(h, 1234)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_NO, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_NO, 0, 0))
def notify_begin(self, handle, bits):
"""
pi.notify_begin(h, 1234)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_NB, handle, bits))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_NB, handle, bits))
def notify_pause(self, handle):
"""
...
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_NB, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_NB, handle, 0))
def notify_close(self, handle):
"""
...
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_NC, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_NC, handle, 0))
def set_watchdog(self, user_gpio, wdog_timeout):
"""
...
"""
return _u2i(_pigpio_command(
- self._control, _PI_CMD_WDOG, user_gpio, int(wdog_timeout)))
+ self.sl, _PI_CMD_WDOG, user_gpio, int(wdog_timeout)))
def read_bank_1(self):
"""
0b10010100000011100100001001111
...
"""
- return _pigpio_command(self._control, _PI_CMD_BR1, 0, 0)
+ return _pigpio_command(self.sl, _PI_CMD_BR1, 0, 0)
def read_bank_2(self):
"""
0b1111110000000000000000
...
"""
- return _pigpio_command(self._control, _PI_CMD_BR2, 0, 0)
+ return _pigpio_command(self.sl, _PI_CMD_BR2, 0, 0)
def clear_bank_1(self, bits):
"""
pi.clear_bank_1(int("111110010000",2))
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_BC1, bits, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_BC1, bits, 0))
def clear_bank_2(self, bits):
"""
pi.clear_bank_2(0x1010)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_BC2, bits, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_BC2, bits, 0))
def set_bank_1(self, bits):
"""
pi.set_bank_1(int("111110010000",2))
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_BS1, bits, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_BS1, bits, 0))
def set_bank_2(self, bits):
"""
pi.set_bank_2(0x303)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_BS2, bits, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_BS2, bits, 0))
def get_current_tick(self):
"""
t2 = pi.get_current_tick()
...
"""
- return _pigpio_command(self._control, _PI_CMD_TICK, 0, 0)
+ return _pigpio_command(self.sl, _PI_CMD_TICK, 0, 0)
def get_hardware_revision(self):
"""
2
...
"""
- return _pigpio_command(self._control, _PI_CMD_HWVER, 0, 0)
+ return _pigpio_command(self.sl, _PI_CMD_HWVER, 0, 0)
def get_pigpio_version(self):
"""
v = pi.get_pigpio_version()
...
"""
- return _pigpio_command(self._control, _PI_CMD_PIGPV, 0, 0)
+ return _pigpio_command(self.sl, _PI_CMD_PIGPV, 0, 0)
def wave_clear(self):
"""
pi.wave_clear()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVCLR, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVCLR, 0, 0))
def wave_add_new(self):
"""
pi.wave_add_new()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVNEW, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVNEW, 0, 0))
def wave_add_generic(self, pulses):
"""
ext.extend(struct.pack("III", p.gpio_on, p.gpio_off, p.delay))
extents = [ext]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_WVAG, 0, 0, len(pulses)*12, extents))
+ self.sl, _PI_CMD_WVAG, 0, 0, len(pulses)*12, extents))
else:
return 0
if len(data):
extents = [struct.pack("I", offset), data]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_WVAS, user_gpio, bb_baud, len(data)+4, extents))
+ self.sl, _PI_CMD_WVAS, user_gpio, bb_baud, len(data)+4, extents))
else:
return 0
wid = pi.wave_create()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVCRE, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVCRE, 0, 0))
def wave_delete(self, wave_id):
"""
pi.wave_delete(0) # delete all waves
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVDEL, wave_id, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVDEL, wave_id, 0))
def wave_tx_start(self): # DEPRECATED
"""
Use [*wave_create*]/[*wave_send_**] instead.
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVGO, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVGO, 0, 0))
def wave_tx_repeat(self): # DEPRECATED
"""
Use [*wave_create*]/[*wave_send_**] instead.
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVGOR, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVGOR, 0, 0))
def wave_send_once(self, wave_id):
"""
cbs = pi.wave_send_once(wid)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVTX, wave_id, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVTX, wave_id, 0))
def wave_send_repeat(self, wave_id):
"""
cbs = pi.wave_send_repeat(wid)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVTXR, wave_id, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVTXR, wave_id, 0))
def wave_tx_busy(self):
"""
pi.wave_send_once(1) # send next waveform
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVBSY, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVBSY, 0, 0))
def wave_tx_stop(self):
"""
pi.wave_tx_stop()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVHLT, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVHLT, 0, 0))
def wave_get_micros(self):
"""
micros = pi.wave_get_micros()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVSM, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVSM, 0, 0))
def wave_get_max_micros(self):
"""
micros = pi.wave_get_max_micros()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVSM, 2, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVSM, 2, 0))
def wave_get_pulses(self):
"""
pulses = pi.wave_get_pulses()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVSP, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVSP, 0, 0))
def wave_get_max_pulses(self):
"""
pulses = pi.wave_get_max_pulses()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVSP, 2, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVSP, 2, 0))
def wave_get_cbs(self):
"""
cbs = pi.wave_get_cbs()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVSC, 0, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVSC, 0, 0))
def wave_get_max_cbs(self):
"""
cbs = pi.wave_get_max_cbs()
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_WVSC, 2, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_WVSC, 2, 0))
def i2c_open(self, i2c_bus, i2c_address, i2c_flags=0):
"""
# I i2c_flags
extents = [struct.pack("I", i2c_flags)]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CO, i2c_bus, i2c_address, 4, extents))
+ self.sl, _PI_CMD_I2CO, i2c_bus, i2c_address, 4, extents))
def i2c_close(self, handle):
"""
pi.i2c_close(h)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_I2CC, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_I2CC, handle, 0))
def i2c_read_device(self, handle, count):
"""
(count, data) = pi.i2c_read_device(h, 12)
...
"""
- bytes = _u2i(_pigpio_command(self._control, _PI_CMD_I2CRD, handle, count))
+ bytes = _u2i(_pigpio_command(self.sl, _PI_CMD_I2CRD, handle, count))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
# s len data bytes
if len(data):
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CWD, handle, 0, len(data), [data]))
+ self.sl, _PI_CMD_I2CWD, handle, 0, len(data), [data]))
else:
return 0
pi.i2c_write_quick(3, 0) # send 0 to device 3
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_I2CWQ, handle, bit))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_I2CWQ, handle, bit))
def i2c_write_byte(self, handle, byte_val):
"""
...
"""
return _u2i(
- _pigpio_command(self._control, _PI_CMD_I2CWS, handle, byte_val))
+ _pigpio_command(self.sl, _PI_CMD_I2CWS, handle, byte_val))
def i2c_read_byte(self, handle):
"""
b = pi.i2c_read_byte(2) # read a byte from device 2
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_I2CRS, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_I2CRS, handle, 0))
def i2c_write_byte_data(self, handle, reg, byte_val):
"""
# I byte_val
extents = [struct.pack("I", byte_val)]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CWB, handle, reg, 4, extents))
+ self.sl, _PI_CMD_I2CWB, handle, reg, 4, extents))
def i2c_write_word_data(self, handle, reg, word_val):
"""
# I word_val
extents = [struct.pack("I", word_val)]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CWW, handle, reg, 4, extents))
+ self.sl, _PI_CMD_I2CWW, handle, reg, 4, extents))
def i2c_read_byte_data(self, handle, reg):
"""
b = pi.i2c_read_byte_data(0, 1)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_I2CRB, handle, reg))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_I2CRB, handle, reg))
def i2c_read_word_data(self, handle, reg):
"""
w = pi.i2c_read_word_data(2, 7)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_I2CRW, handle, reg))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_I2CRW, handle, reg))
def i2c_process_call(self, handle, reg, word_val):
"""
# I word_val
extents = [struct.pack("I", word_val)]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CPC, handle, reg, 4, extents))
+ self.sl, _PI_CMD_I2CPC, handle, reg, 4, extents))
def i2c_write_block_data(self, handle, reg, data):
"""
# s len data bytes
if len(data):
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CWK, handle, reg, len(data), [data]))
+ self.sl, _PI_CMD_I2CWK, handle, reg, len(data), [data]))
else:
return 0
# process read failure
...
"""
- bytes = _u2i(_pigpio_command(self._control, _PI_CMD_I2CRK, handle, reg))
+ bytes = _u2i(_pigpio_command(self.sl, _PI_CMD_I2CRK, handle, reg))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
## extension ##
# s len data bytes
bytes = _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CPK, handle, reg, len(data), [data]))
+ self.sl, _PI_CMD_I2CPK, handle, reg, len(data), [data]))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
# s len data bytes
if len(data):
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CWI, handle, reg, len(data), [data]))
+ self.sl, _PI_CMD_I2CWI, handle, reg, len(data), [data]))
else:
return 0
# I count
extents = [struct.pack("I", count)]
bytes = _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_I2CRI, handle, reg, 4, extents))
+ self.sl, _PI_CMD_I2CRI, handle, reg, 4, extents))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
# I spi_flags
extents = [struct.pack("I", spi_flags)]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_SPIO, spi_channel, spi_baud, 4, extents))
+ self.sl, _PI_CMD_SPIO, spi_channel, spi_baud, 4, extents))
def spi_close(self, handle):
"""
pi.spi_close(h)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_SPIC, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_SPIC, handle, 0))
def spi_read(self, handle, count):
"""
...
"""
bytes = _u2i(_pigpio_command(
- self._control, _PI_CMD_SPIR, handle, count))
+ self.sl, _PI_CMD_SPIR, handle, count))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
## extension ##
# s len data bytes
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_SPIW, handle, 0, len(data), [data]))
+ self.sl, _PI_CMD_SPIW, handle, 0, len(data), [data]))
def spi_xfer(self, handle, data):
"""
## extension ##
# s len data bytes
bytes = _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_SPIX, handle, 0, len(data), [data]))
+ self.sl, _PI_CMD_SPIX, handle, 0, len(data), [data]))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
## extension ##
# s len data bytes
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_SERO, ser_baud, ser_flags, len(tty), [tty]))
+ self.sl, _PI_CMD_SERO, ser_baud, ser_flags, len(tty), [tty]))
def serial_close(self, handle):
"""
pi.serial_close(h1)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_SERC, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_SERC, handle, 0))
def serial_read_byte(self, handle):
"""
b = pi.serial_read_byte(h1)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_SERRB, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_SERRB, handle, 0))
def serial_write_byte(self, handle, byte_val):
"""
...
"""
return _u2i(
- _pigpio_command(self._control, _PI_CMD_SERWB, handle, byte_val))
+ _pigpio_command(self.sl, _PI_CMD_SERWB, handle, byte_val))
def serial_read(self, handle, count):
"""
# process read data
...
"""
- bytes = _u2i(_pigpio_command(self._control, _PI_CMD_SERR, handle, count))
+ bytes = _u2i(_pigpio_command(self.sl, _PI_CMD_SERR, handle, count))
if bytes > 0:
return bytes, self._rxbuf(bytes)
# s len data bytes
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_SERW, handle, 0, len(data), [data]))
+ self.sl, _PI_CMD_SERW, handle, 0, len(data), [data]))
def serial_data_available(self, handle):
"""
(b, d) = pi.serial_read(h1, rdy)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_SERDA, handle, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_SERDA, handle, 0))
def gpio_trigger(self, user_gpio, pulse_len=10, level=1):
"""
# I level
extents = [struct.pack("I", level)]
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_TRIG, user_gpio, pulse_len, 4, extents))
+ self.sl, _PI_CMD_TRIG, user_gpio, pulse_len, 4, extents))
def store_script(self, script):
"""
# s len data bytes
if len(script):
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_PROC, 0, 0, len(script), [script]))
+ self.sl, _PI_CMD_PROC, 0, 0, len(script), [script]))
else:
return 0
nump = 0
extents = []
return _u2i(_pigpio_command_ext(
- self._control, _PI_CMD_PROCR, script_id, 0, nump*4, extents))
+ self.sl, _PI_CMD_PROCR, script_id, 0, nump*4, extents))
def script_status(self, script_id):
"""
(s, pars) = pi.script_status(sid)
...
"""
- status = _u2i(_pigpio_command(self._control, _PI_CMD_PROCP, script_id, 0))
+ status = _u2i(_pigpio_command(self.sl, _PI_CMD_PROCP, script_id, 0))
if status > 0:
- params = struct.unpack('I10i', self._control.recv(44))
+ params = struct.unpack('I10i', self.sl.s.recv(44))
return params[0], params[1:]
return status, ()
status = pi.stop_script(sid)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PROCS, script_id, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PROCS, script_id, 0))
def delete_script(self, script_id):
"""
status = pi.delete_script(sid)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_PROCD, script_id, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_PROCD, script_id, 0))
def bb_serial_read_open(self, user_gpio, bb_baud):
"""
...
"""
return _u2i(_pigpio_command(
- self._control, _PI_CMD_SLRO, user_gpio, bb_baud))
+ self.sl, _PI_CMD_SLRO, user_gpio, bb_baud))
def bb_serial_read(self, user_gpio):
"""
(count, data) = pi.bb_serial_read(4)
...
"""
- bytes = _u2i(_pigpio_command(self._control, _PI_CMD_SLR, user_gpio, 10000))
+ bytes = _u2i(_pigpio_command(self.sl, _PI_CMD_SLR, user_gpio, 10000))
if bytes > 0:
return bytes, self._rxbuf(bytes)
return bytes, ""
status = pi.bb_serial_read_close(17)
...
"""
- return _u2i(_pigpio_command(self._control, _PI_CMD_SLRC, user_gpio, 0))
+ return _u2i(_pigpio_command(self.sl, _PI_CMD_SLRC, user_gpio, 0))
def callback(self, user_gpio, edge=RISING_EDGE, func=None):
"""
"""
self.connected = True
- self._control = None
+ self.sl = _socklock()
self._notify = None
self._host = ''
self._host = host
self._port = int(port)
- self._control = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.sl.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Disable the Nagle algorithm.
- self._control.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+ self.sl.s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
try:
- self._control.connect((self._host, self._port))
- self._notify = _callback_thread(self._control, self._host, self._port)
+ self.sl.s.connect((self._host, self._port))
+ self._notify = _callback_thread(self.sl, self._host, self._port)
except socket.error:
self.connected = False
- if self._control is not None:
- self._control = None
+ if self.sl.s is not None:
+ self.sl.s = None
if self._host == '':
h = "localhost"
else:
self._notify.stop()
self._notify = None
- if self._control is not None:
- self._control.close()
- self._control = None
+ if self.sl.s is not None:
+ self.sl.s.close()
+ self.sl.s = None
def xref():
"""
projects / pigpio.git / commitdiff