Article Index

Raspberry Pi IoT In Python Using Linux Drivers 
Second Edition

Programs

DriverPython2e360

We have decided not to make the programs available as a download because this is not the point of the book - the programs are not finished production code but something you should type in and study.

The best solution is to provide the source code of the programs in a form that can be copied and pasted into a NetBeans of VS Code project. 

 

Note: The VS Code task listings are at the very end of this page.

The only downside is that you have to create a project to paste the code into.

To do this follow the instructionin the book-- in particular make sure you have added any libraries to the link step and are running as root if required. 

All of the programs below were copy and pasted from working programs in the IDE. They have been formatted using the built in formatter and hence are not identical in layout to the programs in the book. This is to make copy and pasting them easier. The programs in the book are formatted to be easy to read on the page.

If anything you consider important is missing or if you have any requests or comments  contact:

This email address is being protected from spambots. You need JavaScript enabled to view it. 


Page 41

import io
from time import sleep

fd = io.open("/sys/class/leds/ACT/trigger","w")
fd.write("none")
fd.close()

fd = io.open("/sys/class/leds/ACT/brightness", "w")
while(True):
    fd.write("0")
    fd.flush()
    sleep(1)
    fd.write("1")
    fd.flush()
    sleep(1)

 

Page 43

import io
from time import sleep

fd = io.open("/sys/class/leds/ACT/trigger","w")
fd.write("timer")
fd.close()
fd = io.open("/sys/class/leds/ACT/delay_on", "w")
fd.write("2000")
fd.close()
fd = io.open("/sys/class/leds/ACT/delay_off", "w")
fd.write("3000")
fd.close()
 

 

Page 53

import gpiod
chip = gpiod.Chip("0")
print(chip.label())
print(chip.name())
print(chip.num_lines())
chip.close()
 
change 0 to 4 for a Pi 5

Page 57

import gpiod

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
while(True):
    line.set_value(1)
    line.set_value(0)
 
change 0 to 4 for a Pi 5
 

Page 57

import gpiod

chip = gpiod.Chip("0")
lines = chip.get_lines([4, 17])

lines.request(consumer="myprog.py", type=gpiod.LINE_REQ_DIR_OUT, default_vals=[0, 1])
while(True):
    lines.set_values([1, 0])
    lines.set_values([0, 1])
 

Remember the need to change the chip number to "4" for a Pi 5.

Page 59

import gpiod
from time import sleep, time_ns

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="resistor.py", type=gpiod.LINE_REQ_DIR_IN)
sleep(0.010)
line.release()

line.request(consumer="resistor.py", type=gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
sleep(0.010)
line.release()

line.request(consumer="resistor.py", type=gpiod.LINE_REQ_DIR_IN)
t = time_ns()
while(line.get_value() == 0):
    pass
print((time_ns()-t)/1000)

 

Remember the need to change the chip number to "4" for a Pi 5. 

 


Page 67

import fcntl
import struct
import io
GPIO_GET_CHIPINFO_IOCTL = 0x8044B401

f = io.open("/dev/gpiochip0", "rb", buffering=0)

gpiochip_info = struct.Struct("32s 32s L")
buffer = gpiochip_info.pack(b' ', b' ', 0)

result = fcntl.ioctl(f, GPIO_GET_CHIPINFO_IOCTL, buffer)

name, label, lines = gpiochip_info.unpack(result)

print(name.rstrip(b'\0').decode("utf-8"))
print(label.rstrip(b'\0').decode("utf-8"))
print(lines)
 
To run it on a Pi 5 change gpiochip0 to gpiochip4

Page 70

import fcntl
import struct
import io

GPIO_GET_CHIPINFO_IOCTL = 0x8044B401
GPIO_GET_LINEHANDLE_IOCTL = 0xC16CB403
GPIOHANDLE_SET_LINE_VALUES_IOCTL = 0xC040B409
GPIOHANDLE_GET_LINE_VALUES_IOCTL = 0xC040B408

GPIOHANDLE_REQUEST_OUTPUT = 0x02
GPIOHANDLE_REQUEST_INPUT = 0x01
GPIOHANDLE_REQUEST_ACTIVE_LOW = 0x04
GPIOHANDLE_REQUEST_OPEN_DRAIN = 0x08
GPIOHANDLE_REQUEST_OPEN_SOURCE = 0x10

gpiochip_info = struct.Struct("32s 32s L")
gpiohandle_request = struct.Struct("64L L 64B 32s L L")
gpiohandle_data = struct.Struct("64B")

lines = [0]*64
lines[0] = 4
lines[1] = 17
values = [0]*64
buffer = gpiohandle_request.pack(*lines, GPIOHANDLE_REQUEST_OUTPUT, *values, b"Output test", 2, 0)

f = io.open("/dev/gpiochip0", "rb", buffering=0)
result = fcntl.ioctl(f, GPIO_GET_LINEHANDLE_IOCTL, buffer)
f.close()
fL = struct.unpack_from("L", buffer, 360)[0]

values[0] = 1
values[1] = 0
state1 = gpiohandle_data.pack(*values)

values[0] = 0
values[1] = 1
state2 = gpiohandle_data.pack(*values)

while(True):
    result = fcntl.ioctl(fL, GPIOHANDLE_SET_LINE_VALUES_IOCTL, state1)
    result = fcntl.ioctl(fL, GPIOHANDLE_SET_LINE_VALUES_IOCTL, state2)
 

Remember to change gpiochip0 to gpiochip4 if you are running on a Pi 5.

Page 72

import fcntl
import struct
import io
import os

GPIO_GET_CHIPINFO_IOCTL = 0x8044B401
GPIO_GET_LINEHANDLE_IOCTL = 0xC16CB403
GPIOHANDLE_SET_LINE_VALUES_IOCTL = 0xC040B409
GPIOHANDLE_GET_LINE_VALUES_IOCTL = 0xC040B408

GPIOHANDLE_REQUEST_OUTPUT = 0x02
GPIOHANDLE_REQUEST_INPUT = 0x01
GPIOHANDLE_REQUEST_ACTIVE_LOW = 0x04
GPIOHANDLE_REQUEST_OPEN_DRAIN = 0x08
GPIOHANDLE_REQUEST_OPEN_SOURCE = 0x10

gpiochip_info = struct.Struct("32s 32s L")
gpiohandle_request = struct.Struct("64L L 64B 32s L L")
gpiohandle_data = struct.Struct("64B")

lines = [0]*64
lines[0] = 4
lines[1] = 17
values = [0]*64
buffer = gpiohandle_request.pack(*lines, GPIOHANDLE_REQUEST_INPUT, *values, b"Input test", 2, 0)

f = io.open("/dev/gpiochip0", "rb", buffering=0)
result = fcntl.ioctl(f, GPIO_GET_LINEHANDLE_IOCTL, buffer)
f.close()
fL = struct.unpack_from("L", result, 360)[0]
inputBuffer = gpiohandle_data.pack(*values)
result = fcntl.ioctl(fL, GPIOHANDLE_GET_LINE_VALUES_IOCTL, inputBuffer)
os.close(fL)
print(result[0], result[1])
 

Remember to change gpiochip0 to gpiochip4 if you are running on a Pi 5.

Page 73

import fcntl
import struct
import io
import os
from time import sleep, time_ns
GPIO_GET_CHIPINFO_IOCTL = 0x8044B401
GPIO_GET_LINEHANDLE_IOCTL = 0xC16CB403
GPIOHANDLE_SET_LINE_VALUES_IOCTL = 0xC040B409
GPIOHANDLE_GET_LINE_VALUES_IOCTL = 0xC040B408

GPIOHANDLE_REQUEST_OUTPUT = 0x02
GPIOHANDLE_REQUEST_INPUT = 0x01
GPIOHANDLE_REQUEST_ACTIVE_LOW = 0x04
GPIOHANDLE_REQUEST_OPEN_DRAIN = 0x08
GPIOHANDLE_REQUEST_OPEN_SOURCE = 0x10

gpiochip_info = struct.Struct("32s 32s L")
gpiohandle_request = struct.Struct("64L L 64B 32s L L")
gpiohandle_data = struct.Struct("64B")

lines = [0]*64
lines[0] = 4
values = [0]*64

buffer = gpiohandle_request.pack(*lines, GPIOHANDLE_REQUEST_INPUT,*values, b"Resistance test", 1, 0)

f = io.open("/dev/gpiochip0", "rb", buffering=0)
result = fcntl.ioctl(f, GPIO_GET_LINEHANDLE_IOCTL, buffer)
fL = struct.unpack_from("L", result, 360)[0]
state = gpiohandle_data.pack(*values)
result = fcntl.ioctl(fL, GPIOHANDLE_GET_LINE_VALUES_IOCTL, state)
sleep(0.01)
os.close(fL)

buffer = gpiohandle_request.pack(*lines, GPIOHANDLE_REQUEST_OUTPUT, *values, b"Resistance test", 1, 0)
result = fcntl.ioctl(f, GPIO_GET_LINEHANDLE_IOCTL, buffer)
fL = struct.unpack_from("L", result, 360)[0]
fcntl.ioctl(fL, GPIOHANDLE_SET_LINE_VALUES_IOCTL, state)
sleep(0.01)
os.close(fL)

buffer = gpiohandle_request.pack(*lines, GPIOHANDLE_REQUEST_INPUT,*values, b"Resistance test", 1, 0)
result = fcntl.ioctl(f, GPIO_GET_LINEHANDLE_IOCTL, buffer)
fL = struct.unpack_from("L", result, 360)[0]
t = time_ns()
result = fcntl.ioctl(fL, GPIOHANDLE_GET_LINE_VALUES_IOCTL, state)
while(result[0] == 0):
    result = fcntl.ioctl(fL, GPIOHANDLE_GET_LINE_VALUES_IOCTL, state)
print((time_ns()-t)/1000)
os.close(fL)
f.close()
 

Remember to change gpiochip0 to gpiochip4 if you are running on a Pi 5.


 

Page 80

import gpiod

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_EV_BOTH_EDGES)

event = line.event_wait(sec=1)
if event:
    print("Event on line 4")
else:
    print("Time out")
 

Remember to change the chip number from “0” to “4” if running on a Pi 5.

Page 81

import gpiod
from time import sleep, time_ns

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="resistor.py", type=gpiod.LINE_REQ_DIR_IN)
sleep(0.010)
line.release()

line.request(consumer="resistor.py", type=gpiod.LINE_REQ_DIR_OUT, default_vals=[0])
sleep(0.010)
line.release()

line.request(consumer="resistor.py", type=gpiod.LINE_REQ_EV_RISING_EDGE)
t = time_ns()
event = line.event_read()
print((event.sec*1000000000+event.nsec-t)/1000)
 

Remember to change the chip number from “0” to “4” if running on a Pi 5.

 

Page 82

import gpiod

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_EV_BOTH_EDGES)

while(True):
    while True:
        event1 = line.event_read()
        if event1.type == event1.RISING_EDGE:
            break
    while True:
        event2 = line.event_read()
        if event2.type == gpiod.LineEvent.FALLING_EDGE:
            break
    print((event2.sec-event2.sec)*1000000000 +
          (event2.nsec-event1.nsec))
 

Remember to change the chip number from “0” to “4” if running on a Pi 5.

 

Page 783a

import gpiod
from time import sleep

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_DIR_IN)
print("Press the button", flush=True)
sleep(20)
if line.get_value() == 1:
    print("button pressed")
else:
    print("button not pressed")
 

Don’t forget to change "0" to "4" if running on a Pi 5.

Page 83b

import gpiod
from time import sleep

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_EV_BOTH_EDGES)
print("Press the button", flush=True)
sleep(20)
event = line.event_wait(sec=0)
if event:
    print("button pressed")
else:
    print("button not pressed")
 

Don’t forget to change "0" to "4" if running on a Pi 5.

Page 84

import gpiod

def event_poll(line):
    event = line.event_wait(sec=0)
    if event:
        event = line.event_read()
    return event


chip = gpiod.Chip("0")
line = chip.get_line(4)
line.request(consumer="myprog.py", type=gpiod.LINE_REQ_EV_BOTH_EDGES)
while True:
    event = event_poll(line)
    if event:
        print(event.sec, flush=True)
 

Don’t forget to change "0" to "4" if running on a Pi 5.

Page 87

import fcntl
import struct
import io
import os
from time import sleep, time_ns
import select

GPIO_GET_CHIPINFO_IOCTL = 0x8044B401
GPIO_GET_LINEHANDLE_IOCTL = 0xC16CB403
GPIOHANDLE_SET_LINE_VALUES_IOCTL = 0xC040B409
GPIOHANDLE_GET_LINE_VALUES_IOCTL = 0xC040B408
GPIO_GET_LINEEVENT_IOCTL = 0xC030B404

GPIOHANDLE_REQUEST_OUTPUT = 0x02
GPIOHANDLE_REQUEST_INPUT = 0x01
GPIOHANDLE_REQUEST_ACTIVE_LOW = 0x04
GPIOHANDLE_REQUEST_OPEN_DRAIN = 0x08
GPIOHANDLE_REQUEST_OPEN_SOURCE = 0x10

GPIOEVENT_REQUEST_RISING_EDGE = 0x01
GPIOEVENT_REQUEST_FALLING_EDGE = 0x02
GPIOEVENT_REQUEST_BOTH_EDGES = 0x03

gpiochip_info = struct.Struct("32s 32s L")
gpiohandle_request = struct.Struct("64L L 64B 32s L L")
gpiohandle_data = struct.Struct("64B")
gpioevent_request = struct.Struct("L L L 32s L")
gpioevent_data = struct.Struct("Q L")

buffer = gpioevent_request.pack(4, GPIOHANDLE_REQUEST_INPUT, GPIOEVENT_REQUEST_BOTH_EDGES, b"Event test", 0)

f = io.open("/dev/gpiochip0", "rb", buffering=0)
result = fcntl.ioctl(f, GPIO_GET_LINEEVENT_IOCTL, buffer)
f.close()
fL = struct.unpack_from("L", result, 44)[0]

e = select.epoll()
e.register(fL)
while True:
    events = e.poll(1)
    if len(events) > 0:
        buffer = os.read(fL, 100)
        timestamp, id = gpioevent_data.unpack(buffer[0:12])
        print(timestamp, id, flush=True)
 

Don’t forget to change "0" to "4" if running on a Pi 5.

Page 88

import gpiod
import struct
import io
import os
import select

gpioevent_data = struct.Struct("Q L")

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_EV_BOTH_EDGES)

fL = line.event_get_fd()
e = select.epoll()
e.register(fL)
while True:
    events = e.poll(1)
    if len(events) > 0:
        buffer = os.read(fL, 100)
        timestamp, id = gpioevent_data.unpack(buffer[0:12])
        print(timestamp, id, flush=True)
 

Don’t forget to change "0" to "4" if running on a Pi 5.

Page 90

import gpiod
from time import sleep
import threading


def waitForInterrupt(line):
    while(True):
        event = line.event_read()
        print(event.sec*1000000000+event.nsec, flush=True)


chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_EV_BOTH_EDGES)

IntThread = threading.Thread(target=waitForInterrupt, args=(line,))
IntThread.start()


while True:
    sleep(2)
    print("running", flush=True)
 
Don’t forget to change "0" to "4" if running on a Pi 5.

 

Page 100

import io

fdr = io.open("/sys/bus/iio/devices/iio:device0/name", "rb", buffering=0)
name = fdr.readline().decode("utf-8")
print("name=", name)
fdr.close()

fdr = io.open("/sys/bus/iio/devices/iio:device0/in_temp_input",  "rb", buffering=0)
temp = fdr.readline().decode("utf-8")
print("temp=", int(temp)/1000, "C")
fdr.close()

fdr = io.open("/ sys/bus/iio/devices/iio: device0 /in_humidityrelative_input", "rb", buffering=0)
hum = fdr.readline().decode("utf-8")
print("Humidity=", int(hum)/1000, "%")
fdr.close()

Page 102

import subprocess
import io
import fcntl


def checkDht11():
    indicator = "dht11  gpiopin=4"
    command = ["sudo", "dtoverlay", "dht11", "gpiopin=4"]
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    print(output)
    if output.find(indicator) == -1:
        temp = subprocess.Popen(command, stdout=subprocess.PIPE)
        output = str(temp.communicate())
        print(output)
    return


checkDht11()

fdr = io.open("/sys/bus/iio/devices/iio:device0/name", "rb", buffering=0)
name = fdr.readline().decode("utf-8")
print("name=", name)
fdr.close()

fdr = io.open("/sys/bus/iio/devices/iio:device0/in_temp_input",  "rb", buffering=0)
temp = fdr.readline().decode("utf-8")
print("temp=", int(temp)/1000, "C")
fdr.close()

fdr = io.open( "/sys/bus/iio/devices/iio:device0/in_humidityrelative_input", "rb", buffering=0)
hum = fdr.readline().decode("utf-8")
print("Humidity=", int(hum)/1000, "%")
fdr.close()
 

 

Page 123a

import gpiod
from time import sleep

chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_DIR_OUT, default_vals=[0])

period = 20
duty = 25

ontime = period/1000 * duty / 100
offtime = period/1000 - ontime

while(True):
    line.set_value(1)
    sleep(ontime)
    line.set_value(0)
    sleep(offtime)
 

Remember to change "0" to "4" if you are running on a Pi 5.

Page 123b

import gpiod
from time import sleep
import threading


def softPWM(line, period, duty):
    ontime = period/1000 * duty / 100
    offtime = period/1000 - ontime
    while(True):
        line.set_value(1)
        sleep(ontime)
        line.set_value(0)
        sleep(offtime)


chip = gpiod.Chip("0")
line = chip.get_line(4)

line.request(consumer="myprog.py", type=gpiod.LINE_REQ_DIR_OUT, default_vals=[0])

period = 20
duty = 75

IntThread = threading.Thread(target=softPWM, args=(line, period, duty))
IntThread.start()

while(True):
    print("working", flush=True)
    sleep(2)
 

Remember to change "0" to "4" if you are running on a Pi 5.

 

Page 127

import subprocess
import io

def checkPWM():
    indicator = "pwm-2chan"
    command = ["sudo", "dtoverlay", "pwm-2chan"]
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    print(output)
    if output.find(indicator) == -1:
        temp = subprocess.Popen(command, stdout=subprocess.PIPE)
        output = str(temp.communicate())
        print(output)
    return


checkPWM()

fdw = io.open("/sys/class/pwm/pwmchip0/export", "wb", buffering=0)
fdw.write(b"0")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip0/pwm0/period", "wb", buffering=0)
fdw.write(b"10000000")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip0/pwm0/duty_cycle", "wb", buffering=0)
fdw.write(b"8000000")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip0/pwm0/enable", "wb", buffering=0)
fdw.write(b"1")
fdw.close()

Page 128

import subprocess
import io

def checkPWM():
    indicator = "pwm-2chan"
    command =["sudo", "dtoverlay", "pwm-2chan","pin=18", "func=2", "pin2=12","func2=4"]
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout = subprocess.PIPE)
    output = str(temp.communicate())
    print(output)
    if output.find(indicator)!=-1:
        temp = subprocess.Popen(["sudo","dtoverlay", "-R", "pwm-2chan"], stdout = subprocess.PIPE)
    temp = subprocess.Popen(command, stdout = subprocess.PIPE)
    output = str(temp.communicate())
    print(output)
    return

checkPWM()

fdw = io.open("/sys/class/pwm/pwmchip2/export", "wb", buffering=0)
fdw.write(b"0")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/pwm0/period","wb", buffering=0)
fdw.write(b"10000000")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/pwm0/duty_cycle", "wb", buffering=0)
fdw.write(b"8000000")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/pwm0/enable", "wb", buffering=0)
fdw.write(b"1")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/export", "wb", buffering=0)
fdw.write(b"2")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/pwm2/period",  "wb", buffering=0)
fdw.write(b"10000000")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/pwm2/duty_cycle", "wb", buffering=0)
fdw.write(b"8000000")
fdw.close()

fdw = io.open("/sys/class/pwm/pwmchip2/pwm2/enable","wb", buffering=0)
fdw.write(b"1")
fdw.close()
 
 

Page 130

import subprocess
import io
from time import sleep
import os
 
Pi5=True
 
class Pwm:
    def __init__(self, channel,f,duty,Pi5=False):
        if not(channel==0 or channel==1):
            return
        if Pi5:
            self.path="/sys/class/pwm/pwmchip2/"
        else:
            self.path="/sys/class/pwm/pwmchip0/"

        self.chan = str(channel)
   
        indicator = "pwm-2chan  pin=12 func=4 pin2=13 func2=4"
        command =["sudo", "dtoverlay", "pwm-2chan","pin=12", "func=4", "pin2=13","func2=4"]
        temp = subprocess.Popen(["sudo", "dtparam", "-l"],  stdout = subprocess.PIPE)
        output = str(temp.communicate())
        print(output)
        if output.find(indicator)==-1:
            temp = subprocess.Popen(command, stdout = subprocess.PIPE)
            output = str(temp.communicate())
            print(output)    

        if not(os.path.exists(self.path+"pwm"+self.chan)):
            fdw = io.open(self.path+"export", "w")
            fdw.write(self.chan)
            fdw.close()

        while not(os.path.exists(self.path+"pwm"+self.chan+  "/enable")):
            pass
             
        self.fdwp = io.open(self.path+"pwm"+self.chan+ "/period", "w")
        self.setFreq(f)
        self.fdwd = io.open(self.path+"pwm"+self.chan+ "/duty_cycle", "w")
        self.setDuty(duty)

    def setFreq(self,f):
        self.f=int(1000000000/f)  
        self.fdwp.write(str(self.f))
        self.fdwp.flush()
   
    def setDuty(self,duty):
        self.duty=int(self.f*duty/100)
        self.fdwd.write(str(self.duty))
        self.fdwd.flush()

    def enableChan(self):
        fdw = io.open(self.path+"pwm"+self.chan+"/enable", "w")
        fdw.write("1")
        fdw.close()
   
    def disableChan(self):
        fdw = io.open(self.path+"pwm"+self.chan+"/enable", "w")
        fdw.write("0")
        fdw.close()
   
    def closeChan(self):
        self.fdwd.close()
        self.fdwp.close()
        fdw = io.open(self.path+"unexport", "w")
        fdw.write(self.chan)        
        fdw.close()
 

Page 141

import subprocess
import io
from time import sleep
import os
 
Pi5=True
 
class Pwm:
    def __init__(self, channel,f,duty,Pi5=False):
        if not(channel==0 or channel==1):
            return
        if Pi5:
            self.path="/sys/class/pwm/pwmchip2/"
        else:
            self.path="/sys/class/pwm/pwmchip0/"

        self.chan = str(channel)
        indicator = "pwm-2chan"
        command =["sudo", "dtoverlay", "pwm-2chan"]
        temp = subprocess.Popen(["sudo", "dtparam", "-l"],  stdout = subprocess.PIPE)
        output = str(temp.communicate())
        print(output,flush=True)

        indicator = "pwm-2chan  pin=12 func=4 pin2=13 func2=4"
        command =["sudo", "dtoverlay", "pwm-2chan"]
        temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout = subprocess.PIPE)
        output = str(temp.communicate())
        print(output,flush=True)
        command =["sudo", "dtoverlay", "pwm-2chan","pin=12", "func=4", "pin2=13","func2=4"]
        temp = subprocess.Popen(["sudo", "dtparam", "-l"],  stdout = subprocess.PIPE)
        output = str(temp.communicate())
        print(output)
        if output.find(indicator)==-1:
            temp = subprocess.Popen(command,
                                         stdout = subprocess.PIPE)
            output = str(temp.communicate())
            print(output)
   
        if not(os.path.exists(self.path+"pwm"+self.chan)):
            fdw = io.open(self.path+"export", "w")
            fdw.write(self.chan)
            fdw.close()
        while not(os.path.exists(self.path+"pwm"+self.chan+"/enable")):
            pass
             
        self.fdwp = io.open(self.path+"pwm"+self.chan+"/period", "w")
        self.setFreq(f)
        self.fdwd = io.open(self.path+"pwm"+self.chan+"/duty_cycle", "w")
        self.setDuty(duty)

    def setFreq(self,f):
        self.f=int(1000000000/f)  
        t=str(self.f)
        self.fdwp.write(str(self.f))
        self.fdwp.flush()
   
    def setDuty(self,duty):
        self.duty=int(self.f*duty/100)
        self.fdwd.write(str(self.duty))
        self.fdwd.flush()

    def enableChan(self):
        fdw = io.open(self.path+"pwm"+self.chan+"/enable", "w")
        fdw.write("1")
        fdw.close()
   
    def disableChan(self):
        fdw = io.open(self.path+"pwm"+self.chan+"/enable", "w")
        fdw.write("0")
        fdw.close()
   
    def closeChan(self):
        self.fdwd.close()
        self.fdwp.close()
        fdw = io.open(self.path+"unexport", "w")
        fdw.write(self.chan)        
        fdw.close()
   
    def inverChan(self):
        fdw = io.open(self.path+ "pwm"+self.chan+"/polarity", "w")
        fdw.write("inversed")
        fdw.close()
   
    def normalChan(self):
        fdw = io.open(self.path+ "pwm"+self.chan+"/polarity", "w")
        fdw.write("normal")
        fdw.close(

 


Page 151

import subprocess
import spidev
from time import sleep


def checkSPI():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    lastSPI = output.rfind("spi")
    if lastSPI != -1:
        lastSPI = output.find("spi=on", lastSPI)
    if lastSPI == -1:
        temp = subprocess.Popen(
            ["sudo", "dtparam", "spi=on"], stdout=subprocess.PIPE)
        output = str(temp.communicate())


checkSPI()

spi = spidev.SpiDev()
spi.open(0, 0)
spi.max_speed_hz = 5000
Data = [0xAA]
print(hex(Data[0]))
data = spi.xfer(Data)
print(hex(Data[0]))
if Data == 0xAA:
    print("match")
spi.close()

Page 161

import subprocess
import spidev

def checkSPI():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    lastSPI = output.rfind("spi")
    if lastSPI != -1:
        lastSPI = output.find("spi=on", lastSPI)
    if lastSPI == -1:
        temp = subprocess.Popen(
            ["sudo", "dtparam", "spi=on"], stdout=subprocess.PIPE)
        output = str(temp.communicate())


checkSPI()

spi = spidev.SpiDev()
spi.open(0, 0)
spi.max_speed_hz = 100000
spi.mode = 0
spi.bits_per_word = 8

Data = [0x01, 0x80, 0x00]
spi.xfer(Data)
raw = (Data[1] & 0x03) << 8 | Data[2]
print(raw)
volts = raw * 3.3 / 1023.0
print(volts, "V")
spi.close()
 

 

Page 181

import subprocess
import io
import fcntl


def checkI2CBus():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    lasti2c = output.rfind("i2c_arm")
    if lasti2c != -1:
        lasti2c = output.find("i2c_arm=on", lasti2c)
    if lasti2c == -1:
        temp = subprocess.Popen(["sudo", "dtparam", "i2c_arm=on"],  stdout=subprocess.PIPE)
        output = str(temp.communicate())
    return


checkI2CBus()

I2C_SLAVE = 0x0703

fdr = io.open("/dev/i2c-1", "rb", buffering=0)
fdw = io.open("/dev/i2c-1", "wb", buffering=0)
fcntl.ioctl(fdr, I2C_SLAVE, 0x40)
fcntl.ioctl(fdw, I2C_SLAVE, 0x40)
fdw.write(bytearray([0xE7]))
data = fdr.read(1)
print(data)
fdr.close()
fdw.close()

Page 183

import subprocess
import io
import fcntl
from time import sleep

checkI2CBus()
I2C_SLAVE = 0x0703

fdr = io.open("/dev/i2c-1", "rb", buffering=0)
fdw = io.open("/dev/i2c-1", "wb", buffering=0)
fcntl.ioctl(fdr, I2C_SLAVE, 0x40)
fcntl.ioctl(fdw, I2C_SLAVE, 0x40)

fdw.write(bytearray([0xF3]))
while(True):
    try:
        data = fdr.read(3)
        break
    except:
        sleep(0.01)
msb = data[0]
lsb = data[1]
crc = data[2]
print("msb=", msb, " lsb=", lsb, " crc=", crc)

fdw.close()
fdr.close()

Page 187

import subprocess
import io
import fcntl
from time import sleep


def crcCheck(msb, lsb, check):
    data32 = (msb << 16) | (lsb << 8) | check
    divisor = 0x988000
    for i in range(16):
        if(data32 & 1 << (23 - i)):
            data32 ^= divisor
        divisor >>= 1
    return data32


def checkI2CBus():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"],  stdout=subprocess.PIPE)
    output = str(temp.communicate())
    print(output)
    lasti2c = output.rfind("i2c_arm")
    if lasti2c != -1:
        lasti2c = output.find("i2c_arm=on", lasti2c)
    if lasti2c == -1:
        temp = subprocess.Popen(["sudo", "dtparam", "i2c_arm=on"],  stdout=subprocess.PIPE)
        output = str(temp.communicate())
    return


checkI2CBus()
I2C_SLAVE = 0x0703

fdr = io.open("/dev/i2c-1", "rb", buffering=0)
fdw = io.open("/dev/i2c-1", "wb", buffering=0)
fcntl.ioctl(fdr, I2C_SLAVE, 0x40)
fcntl.ioctl(fdw, I2C_SLAVE, 0x40)

fdw.write(bytearray([0xF3]))
while(True):
    try:
        data = fdr.read(3)
        break
    except:
        sleep(0.01)
msb = data[0]
lsb = data[1]
crc = data[2]

data16 = (msb << 8) | (lsb & 0xFC)
temp = -46.85 + (175.72 * data16 / 65536)
print("Temperature=", temp, "C")

fdw.write(bytearray([0xF5]))
while(True):
    try:
        data = fdr.read(3)
        break
    except:
        sleep(0.01)

msb = data[0]
lsb = data[1]
crc = data[2]
data16 = (msb << 8) | (lsb & 0xFC)
hum = -6 + 125.0 * data16 / 65536

print("humidity=", hum, "%")

print(crcCheck(msb, lsb, crc))
fdw.close()
fdr.close()
 

 

Page 197

import subprocess
import io
import fcntl
from time import sleep

def checkI2CBus():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout = subprocess.PIPE)
    output = str(temp.communicate())
    lasti2c=output.rfind("i2c_arm")
    if lasti2c!=-1:
        lasti2c=output.find("i2c_arm=on",lasti2c)
    if lasti2c==-1:
        temp = subprocess.Popen(["sudo", "dtparam", "i2c_arm=on"],  stdout = subprocess.PIPE)
        output = str(temp.communicate())
    return

I2C_SLAVE=0x0703

fdr = io.open("/dev/i2c-1", "rb", buffering=0)
fdw = io.open("/dev/i2c-1", "wb", buffering=0)
fcntl.ioctl(fdr, I2C_SLAVE, 0x40)
fcntl.ioctl(fdw, I2C_SLAVE, 0x40)

fdw.write( bytearray([0xE3]))

data=fdr.read(3)
msb=data[0]
lsb=data[1]
crc=data[2]
print("msb=",msb," lsb=",lsb," crc=",crc)

fdw.close()
fdr.close()

Page 200

import subprocess
from ctypes import *
import os
import fcntl
import io


def checkI2CBus():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"],  stdout=subprocess.PIPE)
    output = str(temp.communicate())
    lasti2c = output.rfind("i2c_arm")
    if lasti2c != -1:
        lasti2c = output.find("i2c_arm=on", lasti2c)
    if lasti2c == -1:
        temp = subprocess.Popen(["sudo", "dtparam", "i2c_arm=on"], stdout=subprocess.PIPE)
        output = str(temp.communicate())
    return


checkI2CBus()
I2C_SLAVE = 0x0703
I2C_FUNCS = 0x0705

I2C_FUNC_I2C = 0x00000001
I2C_FUNC_10BIT_ADDR = 0x00000002
I2C_FUNC_PROTOCOL_MANGLING = 0x00000004
I2C_FUNC_NOSTART = 0x00000010
I2C_FUNC_SLAVE = 0x00000020

i2cfd = os.open("/dev/i2c-1", os.O_RDWR | os.O_SYNC)
support = c_uint32(0)
fcntl.ioctl(i2cfd, I2C_FUNCS, support)
support = support.value
if support & I2C_FUNC_I2C:
    print("I2C Support")
if support & I2C_FUNC_10BIT_ADDR:
    print("10 bit address Support")
if support & I2C_FUNC_PROTOCOL_MANGLING:
    print("I2C Mangling Support")
if support & I2C_FUNC_NOSTART:
    print("I2C Nostart Support")
if support & I2C_FUNC_SLAVE:
    print("I2C Slave Support")

Page 204

import fcntl
import os
from ctypes import *


class Msgs(Structure):
    _fields_ = [("addr", c_uint16),
                ("flags", c_uint16),
                ("len", c_uint16),
                ("buf", POINTER(c_uint8))
                ]


class MsgSet(Structure):
    _fields_ = [("msgs", POINTER(Msgs)),
                ("nmsgs", c_uint32)
                ]


def i2cReadRegister(i2cfd, slaveaddr, reg, n):
    I2C_RDWR = 0x0707
    I2C_M_RD = 0x0001

    msgs = (Msgs*2)()

    msgs[0].addr = c_uint16(slaveaddr)
    msgs[0].flags = c_uint16(0)
    msgs[0].len = c_uint16(1)
    msgs[0].buf = POINTER(c_uint8)(c_uint8(reg))

    msgs[1].addr = c_uint16(slaveaddr)
    msgs[1].flags = c_uint16(I2C_M_RD)
    msgs[1].len = c_uint16(n)
    buf = (c_uint8*n)(0)
    msgs[1].buf = POINTER(c_uint8)(buf)

    msgset = MsgSet()
    msgset.msgs = POINTER(Msgs)(msgs)
    msgset.nmsgs = c_uint32(2)

    libc = CDLL("libc.so.6")
    libc.ioctl(i2cfd, I2C_RDWR, byref(msgset))

    return msgs[1].buf


def crcCheck(msb, lsb, check):
    data32 = (msb << 16) | (lsb << 8) | check
    divisor = 0x988000
    for i in range(16):
        if(data32 & 1 << (23 - i)):
            data32 ^= divisor
        divisor >>= 1
    return data32


i2cfd = os.open("/dev/i2c-1", os.O_RDWR | os.O_SYNC)
data = i2cReadRegister(i2cfd, 0x40, 0xE3, 3)
msb = data[0]
lsb = data[1]
crc = data[2]

data16 = (msb << 8) | (lsb & 0xFC)
temp = -46.85 + (175.72 * data16 / 65536)
print("Temperature=", temp, "C")
print(hex(msb), hex(lsb), hex(crc))
print(crcCheck(msb, lsb, crc))
 

 

Page 212

import subprocess
import io


def checkLM75():
    indicator1 = "i2c_arm=on"
    command1 = ["sudo", "dtparam", "i2c_arm=on"]
    indicator2 = "lm75"
    command2 = ["sudo", "dtoverlay", "i2c-sensor", "lm75", "addr=0x48"]
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    print(output, flush=True)
    if output.find(indicator1) == -1:
        temp = subprocess.Popen(command1, stdout=subprocess.PIPE)
        output = str(temp.communicate())
        print(output, flush=True)
    if output.find(indicator2) == -1:
        temp = subprocess.Popen(command2, stdout=subprocess.PIPE)
        output = str(temp.communicate())
        print(output, flush=True)


checkLM75()

fdr = io.open("/sys/class/hwmon/hwmon2/temp1_input", "r")
buf = fdr.read()
print(buf)
temp = int(buf) / 1000
print(temp)
fdr.close()

fdw = io.open("/sys/class/hwmon/hwmon2/temp1_max", "w")
fdw.write("19000")
fdw.close()

fdw = io.open("/sys/class/hwmon/hwmon2/temp1_max_hyst", "w")
fdw.write("18000")
fdw.close()

Page 215

import subprocess
import io

def loadHtu21():
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    print(output)
    if output.find("htu21") == -1:
        temp = subprocess.Popen( ["sudo", "dtoverlay", "i2c-sensor", "htu21"], stdout=subprocess.PIPE)
        output = str(temp.communicate())
        print(output)
    return

loadHtu21()

fdrt = io.open("/sys/bus/iio/devices/iio:device0/in_temp_input", "r")
print(int(fdrt.read())/1000)
fdrh = io.open( "/sys/bus/iio/devices/iio:device0/in_humidityrelative_input", "r")
print(int(fdrh.read())/1000)
 

 

Page 229

import subprocess
import io


def load1w(pin):
    indicator = "w1-gpio"
    command = ["sudo", "dtoverlay", "w1-gpio", "gpiopin="+str(pin)]
    temp = subprocess.Popen(["sudo", "dtparam", "-l"], stdout=subprocess.PIPE)
    output = str(temp.communicate())
    print(output, flush=True)
    if output.find(indicator) == -1:
        temp = subprocess.Popen(command, stdout=subprocess.PIPE)
        output = str(temp.communicate())
        print(output, flush=True)


def getDevices():
    fdr = io.open(
        "/sys/bus/w1/drivers/w1_master_driver/w1_bus_master1/w1_master_slaves", "r")
    buffer = fdr.read()
    fdr.close()
    return buffer.split()


def getData(dev, name):
    fdr = io.open("/sys/bus/w1/devices/"+dev+"/"+name, "r")
    data = fdr.read()
    fdr.close()
    return data


load1w(4)
devs = getDevices()
name = getData(devs[0], "name")
print("Name ", name)
resolution = getData(devs[0], "resolution")
print("Resolution ", resolution)
w1_slave = getData(devs[0], "w1_slave")
print("w1_slave ", w1_slave)
temperature = getData(devs[0], "temperature")
print("temperature ", temperature)
temperature = int(temperature) / 1000
print("temperature ", temperature, "C")
alarms = getData(devs[0], "alarms")
print("Alarms ", alarms)

Page 240

from ctypes import *
from socket import *
from struct import *
import os
import subprocess
import io
from time import sleep


NLMSG_DONE = 0x3

CN_W1_IDX = 0x3
CN_W1_VAL = 0x1

W1_SLAVE_ADD = 0
W1_SLAVE_REMOVE = 1
W1_MASTER_ADD = 2
W1_MASTER_REMOVE = 3
W1_MASTER_CMD = 4
W1_SLAVE_CMD = 5
W1_LIST_MASTERS = 6

W1_CMD_READ = 0
W1_CMD_WRITE = 1
W1_CMD_SEARCH = 2
W1_CMD_ALARM_SEARCH = 3
W1_CMD_TOUCH = 4
W1_CMD_RESET = 5
W1_CMD_SLAVE_ADD = 6
W1_CMD_SLAVE_REMOVE = 7
W1_CMD_LIST_SLAVES = 8
W1_CMD_MAX = 9

NETLINK_CONNECTOR = 11

nl_seq = 0


class nlmsghdr(Structure):
    _fields_ = [("nlmsg_len", c_uint32),
                ("nlmsg_type", c_uint16),
                ("nlmsg_flags", c_uint16),
                ("nlmsg_seq", c_uint32),
                ("nlmsg_pid", c_uint32)
                ]


cnh = nlmsghdr()
cnh.nlmsg_seq = c_uint32(nl_seq)
nl_seq += 1
cnh.nlmsg_pid = c_uint32(os.getpid())
cnh.nlmsg_type = c_uint16(NLMSG_DONE)
cnh.nlmsg_flags = c_uint16(0)


class cn_msg(Structure):
    _fields_ = [("idx", c_uint32),
                ("val", c_uint32),
                ("seq", c_uint32),
                ("ack", c_uint32),
                ("len", c_uint16),
                ("flags", c_uint16),
                ]


cmsg = cn_msg()
cmsg.idx = c_uint32(CN_W1_IDX)
cmsg.val = c_uint32(CN_W1_VAL)
cmsg.seq = c_uint32(cnh.nlmsg_seq)
cmsg.ack = c_uint32(0)


class w1_netlink_msg(Structure):
    _fields_ = [("type", c_uint8),
                ("status", c_uint8),
                ("len", c_uint16),
                ("id", c_uint8*8),
                ]


msg = w1_netlink_msg()
msg.type = c_uint8(W1_MASTER_CMD)
msg.id = (c_uint8*8).from_buffer_copy(c_uint64(1))


class w1_netlink_cmd(Structure):
    _fields_ = [("cmd", c_uint8),
                ("res", c_uint8),
                ("len", c_uint16),
                ]


w1_cmd = w1_netlink_cmd()
w1_cmd.cmd = c_uint8(W1_CMD_SEARCH)
w1_cmd.len = c_uint16(0)


msg.len = c_uint16(len(bytearray(w1_cmd)) + w1_cmd.len)
cmsg.len = c_uint16(len(bytearray(msg)) + msg.len)
cnh.nlmsg_len = c_uint32(len(bytearray(cnh))+len(bytearray(cmsg))+cmsg.len)


s = socket(AF_NETLINK, SOCK_DGRAM, NETLINK_CONNECTOR)
s.connect((0, AF_NETLINK))

buffer = bytearray(cnh)+bytearray(cmsg)+bytearray(msg)+bytearray(w1_cmd)
n = s.send(buffer)
print("Bytes sent", n, flush=True)
buffer2 = s.recv(65535)
print("Bytes received", len(buffer2), flush=True)

p = 0
cn_cnhr = nlmsghdr.from_buffer_copy(buffer2, p)
p = p+len(bytes(cn_cnhr))
cmsgr = cn_msg.from_buffer_copy(buffer2, p)
p = p+len(bytes(cmsgr))
msgr = w1_netlink_msg.from_buffer_copy(buffer2, p)
p = p+len(bytes(msgr))
w1_cmdr = w1_netlink_cmd.from_buffer_copy(buffer2, p)

num = int(w1_cmdr.len/8)
print("number of slaves", num)

p = p+len(bytes(w1_cmdr))
slaves = []
for i in range(num):
    slaves.append(c_uint64.from_buffer_copy(buffer2, p+i*8).value)
print([hex(slave) for slave in slaves])

# GET ACK
buffer2 = s.recv(65535)
print("length of ack received ", len(buffer2), flush=True)

# READ TEMPERATURE
# SET UP SLAVE COMMAND
print()
print("Read Temperature")

msg.type = W1_SLAVE_CMD

id = slaves[0].to_bytes(8, "little")
msg.id = (c_uint8*8).from_buffer_copy(id)


# SETUP AND SEND WRITE 0x44 = CONVERT
w1_cmd.cmd = c_uint8(W1_CMD_WRITE)
w1_cmd_data = c_uint8(0x44)


cnh.nlmsg_seq = c_uint32(nl_seq)
nl_seq += 1
cmsg.seq = c_uint32(cnh.nlmsg_seq)

w1_cmd.len = c_uint16(1)
msg.len = c_uint16(len(bytearray(w1_cmd)) + w1_cmd.len)
cmsg.len = c_uint16(len(bytearray(msg)) + msg.len)
cnh.nlmsg_len = c_uint32(len(bytearray(cnh))+len(bytearray(cmsg))+cmsg.len)

buffer = bytearray(cnh)+bytearray(cmsg)+bytearray(msg) + \
    bytearray(w1_cmd)+bytearray(w1_cmd_data)

n = s.send(buffer)
print("Bytes sent", n, flush=True)
buffer2 = s.recv(65535)
print("length of ack received ", len(buffer2), flush=True)
sleep(1)

# READ SCRATCH PAD
cnh.nlmsg_seq = nl_seq
nl_seq += 1
cmsg.seq = cnh.nlmsg_seq

w1_cmd.cmd = W1_CMD_WRITE
w1_cmd_data = c_uint8(0xBE)

w1_cmd2 = w1_netlink_cmd()
w1_cmd2.cmd = c_uint8(W1_CMD_READ)


w1_cmd.len = c_uint16(1)
w1_cmd2.len = c_uint16(9)
msg.len = c_uint16(len(bytearray(w1_cmd)) +
                   len(bytearray(w1_cmd2)) + w1_cmd.len+w1_cmd2.len)
cmsg.len = c_uint16(len(bytearray(msg)) + msg.len)
cnh.nlmsg_len = c_uint32(len(bytearray(cnh))+len(bytearray(cmsg))+cmsg.len)

w1_cmd2_data = (c_uint8*9)(0)
buffer = bytearray(cnh)+bytearray(cmsg)+bytearray(msg)+bytearray(w1_cmd) + \
    bytearray(w1_cmd_data)+bytearray(w1_cmd2)+bytearray(w1_cmd2_data)
n = s.send(buffer)
print("Bytes sent", n, flush=True)

buffer2 = s.recv(65535)
print("length of ack received ", len(buffer2), flush=True)

buffer2 = s.recv(65535)
print("length of data received ", len(buffer2), flush=True)
p = 0
cn_cnhr = nlmsghdr.from_buffer_copy(buffer2, p)
# print(cn_cnhr.nlmsg_len)
p = p+len(bytes(cn_cnhr))
cmsgr = cn_msg.from_buffer_copy(buffer2, p)
p = p+len(bytes(cmsgr))
msgr = w1_netlink_msg.from_buffer_copy(buffer2, p)
p = p+len(bytes(msgr))
w1_cmdr = w1_netlink_cmd.from_buffer_copy(buffer2, p)
p = p+len(bytes(w1_cmdr))

temp = c_uint16.from_buffer_copy(buffer2, p)

print("number of bytes found ", w1_cmdr.len, flush=True)
print(temp.value/16)
 
 

 

VSCODE TASKS

 

settings.json

which has to be customized to your IP, user name and folder:

{
    "sshUser": "pi",
    "sshEndpoint": "192.168.11.151",
    "remoteDirectory": "/home/pi/Documents/",
}

tasks.json

{
    "version": "2.0.0",
    "tasks": [
        {
            "label": "copyToRemote",
            "type": "shell",
            "command": "scp -r ${fileDirname} ${config:sshUser}@${config:sshEndpoint}:${config:remoteDirectory}/",
            "problemMatcher": [],
            "presentation": {
                "showReuseMessage": false,
                "clear": true
            }
        },
        {
            "label": "makeRemoteWorkSpace",
            "type": "shell",
            "command": "ssh ${config:sshUser}@${config:sshEndpoint} 'mkdir  ${config:remoteDirectory}'",
            "problemMatcher": [],
            "presentation": {
                "showReuseMessage": false,
            }
        },
        {
            "label": "RunR",
            "type": "shell",
            "command": "ssh ${config:sshUser}@${config:sshEndpoint} 'python3 ${config:remoteDirectory}/${relativeFileDirname}/${fileBasename}'",
            "problemMatcher": [],
            "presentation": {
                "showReuseMessage": false,
            }
        },
        {
            "label": "RunRemote",
            "dependsOrder": "sequence",
            "dependsOn": [
                "copyToRemote",
                "RunR"
            ],
            "problemMatcher": [],
            "group": {
                "kind": "build",
                "isDefault": true
            },
        },
        {
            "label": "StopREmotePython",
            "type": "shell",
            "command": "ssh ${config:sshUser}@${config:sshEndpoint} 'pkill python3'",
            "problemMatcher": [],
            "presentation": {
                "showReuseMessage": true,
            }
        },
        {
            "label": "wait",
            "type": "shell",
            "command": "timeout 10"
        },
        {
            "label": "tunnel",
            "type": "shell",
            "command": "ssh -2 -L 5678:localhost:5678  ${config:sshUser}@${config:sshEndpoint}",
            "problemMatcher": [],
            "presentation": {
                "showReuseMessage": false,
            }
        },
        {
            "label": "startDebug",
            "type": "shell",
            "command": "ssh -2 ${config:sshUser}@${config:sshEndpoint} 'nohup python3 -m debugpy --listen 0.0.0.0:5678 --wait-for-client ${config:remoteDirectory}/${relativeFileDirname}/${fileBasename} > /dev/null 2>&1 &'",
            "problemMatcher": [],
            "presentation": {
                "showReuseMessage": false,
            }
        },
        {
            "label": "copyAndDebug",
            "dependsOrder": "sequence",
            "dependsOn": [
                "copyToRemote",
                "startDebug",
                "wait"
            ],
            "presentation": {
                "showReuseMessage": false,
            },
        },
    ]
}
launch.json
{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: Remote Attach",
            "type": "python",
            "request": "attach",
            "connect": {
                "host": "localhost",
                "port": 5678
            },
            "pathMappings": [
                {
                    "localRoot": "${workspaceFolder}/${relativeFileDirname}/",
                    "remoteRoot": "${config:remoteDirectory}/${relativeFileDirname}"
                }
            ],
            "preLaunchTask": "copyAndDebug",
            "postDebugTask": "StopREmotePython"
        }
    ]
}