测试多根DS18B20传感器电缆

Question

我对python和一般的编码都是新手，所以我在这里寻求帮助。在工作中，我们需要测试许多连接了多达五个DS18B20传感器的电缆。我们有了这个程序，现在我们已经用它来测试raspberry Pi的电缆了。但它几乎没有什么问题。

程序需要检测连接的传感器的数量，看看它们是否工作，然后按序列号lowest>highest列出它们(这样我们就可以预热它们，看看哪个突出显示，一个接一个地物理编号)，然后重复。

它通常感觉很慢，而且经常崩溃。有了连接了5个传感器的电缆，它可以在最多2根电缆上找到传感器(如果幸运的话，可以找到更多)，当你连接第三根电缆时，它不会再找到它们，或者它会一直说有10个传感器连接。重启Pi可以修复它，但这需要时间。热插拔电缆可能是这里的主要罪魁祸首，但它是最快的方法。

使用3根传感器电缆，您可以检查其中4-5根电缆，直到电缆挂起。

欢迎任何可能出现问题的提示，或者我如何让事情变得更有效率和更快。

代码：

import os,sys,time
import curses
stdscr = curses.initscr()

begin_x = 5; begin_y = 5
height = 40; width = 40  

def temp_raw(sensor_str):
    f = open(sensor_str, 'r')
    lines = f.readlines()
    f.close()
    return lines

def read_temp(sensor_str):
    lines = temp_raw(sensor_str)
    while lines[0].strip()[-3:] != 'YES':
        lines = temp_raw(sensor_str)
    temp_output = lines[1].find('t')
    if temp_output != -1:
        temp_string = lines[1].strip()[temp_output+2:]
        temp_c = float(temp_string) / 1000.0
        return temp_c
init = 1
timer = 0
search = 1
curses.noecho()
curses.curs_set(0)
subdirectories = []
old_nbr_of_subdirs = 1
nbr_of_subdirs = 1
old_subdirectories = []
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')

while (1):
    timer = timer + 1

    while search == 1:
        stdscr.addstr(10,30, "Searching for sensors");
        subdirectories = os.listdir("/sys/bus/w1/devices/");
        nbr_of_subdirs = len(subdirectories);
        time.sleep(3)
        if nbr_of_subdirs > old_nbr_of_subdirs:
            stdscr.addstr(10,30, "Found %d new sensors           " % (nbr_of_subdirs-old_nbr_of_subdirs));
            stdscr.addstr(11,30, "Is this all sensors? (y/n)?");
            stdscr.refresh()
            c = -1;
            while(c == -1):
                c = stdscr.getch();
                if c == ord('y'):
                    stdscr.addstr(10,30, "                                   ");
                    stdscr.addstr(11,30, "                            ");
                    stdscr.refresh()
                    old_nbr_of_subdirs = nbr_of_subdirs;
                    init = 1;
                    search = 0;
                    timer = 0;
        else:
            stdscr.addstr(10,51 + timer, ". ");
            timer = timer + 1;
            timer = timer % 4;
            stdscr.refresh()



    if init == 1:
        init = 0
        alive_ticker = 0;
        temp_list = [];
        sensor_oldtemp = [0,0,0,0,0,0];
        sensor_temp = [0,0,0,0,0,0];
        active_sensor = [0,0,0,0,0,0];
        sensors = 0;
        alive_ticker = 0;
        active = 2;
        confirm = 0; 

        stdscr.addstr(17,0, "Number of subdirectories %d" % (len(subdirectories)));
        removed = 0;
        index = 0;
        length_old_subdirs = len(old_subdirectories);
        while (index < length_old_subdirs):
            length_old_subdirs = len(old_subdirectories);

            if old_subdirectories[index] in subdirectories:
                subdirectories.remove(old_subdirectories[index]);
                removed += 1;
                index += 1;     

            else:
                old_subdirectories.remove(old_subdirectories[index]);
                index = 0;


        stdscr.addstr(30,20, "Removed %d" % (removed))


        stdscr.addstr(18,0, "Number of CUT subdirectories %d" % (len(subdirectories)));


        stdscr.addstr(19,0, "Number of  old_subdirectories %d" % (len(old_subdirectories)));

        stdscr.addstr(13,20, "Press space to confirm active sensor!");
        stdscr.addstr(14,20, "Press r to reset!");

        for index in range(len(subdirectories)):
            if subdirectories[index].find("28") != -1:
                temp_list.append("/sys/bus/w1/devices/"+subdirectories[index]+"/w1_slave");
                sensor_temp[sensors] = 0;
                sensors = sensors + 1;
                old_subdirectories = old_subdirectories + subdirectories;
                temp_list.sort();

    if confirm == 1:
        stdscr.addstr(13,20, "Press space to confirm active sensor!");
        stdscr.addstr(14,20, "Press r to restart!");


    curses.halfdelay(1);
    c = stdscr.getch();

    if c == 32:
        confirm = 0;
        timer = 10
        for z in range(sensors):
            if active_sensor[z] > active:
                stdscr.addstr(z+5,20, "                                                                      " , curses.A_DIM)
                active_sensor[z] = 100  
                stdscr.refresh()
    elif c == ord('n'):
        init = 1
        continue

    elif c == ord('r'):
        for z in range(sensors):
            active_sensor[z] = 1;

    if (timer > 9):
        timer = 0
        nbr_active_sensors = 0
        for y in range(sensors):
            if 99 > active_sensor[y]:
                nbr_active_sensors = nbr_active_sensors + 1;
                sensor_temp[y] = read_temp(temp_list[y]);
                if ((sensor_oldtemp[y] + 0.1) < sensor_temp[y]):
                    active_sensor[y] = active_sensor[y] + 1;
                    sensor_oldtemp[y] = sensor_temp[y]

        stdscr.addstr(3,5, "nbr_active_sensors=%d" % (nbr_active_sensors))
        stdscr.addstr(4,5, "sensors=%d" % (sensors))

        if nbr_active_sensors == 0:
            search = 1
            timer = 0;
            continue

        for x in range(sensors):
            if (99 > active_sensor[x] and active_sensor[x] > active):
                confirm = 1         
                stdscr.addstr(x+5,20, "Sensor %d value %4.2f ID:%s : %d " % (x+1, sensor_temp[x], temp_list[x][20:-9], active_sensor[x]), curses.A_STANDOUT)
            elif active_sensor[x] <= active:
                stdscr.addstr(x+5,20, "Sensor %d value %4.2f ID:%s : %d " % (x+1, sensor_temp[x], temp_list[x][20:-9], active_sensor[x]))
        stdscr.refresh()
        alive_ticker = alive_ticker + 1
        stdscr.addstr(2, 55, "Alive ticker: %6d" % (alive_ticker))

Answer 1

insert ：当您插入第三根电缆(上面有2-3个温度传感器)时，它将找不到任何新的传感器。

如果我没记错的话，2线缆是有限制的。您是否验证过您的硬件支持更多电缆？

我建议重写代码和分离的代码到用户界面，传感器和电线部分。

例如：

class DS18B20(object):
    ...
    def read(self):
        ...

class One_Wire(object):
    def search(self):
        print("Searching sensors")
        ...

    def read(self):
        print('Reading sensors')
        for key in self.sensors:
            self.sensors[key].read()

    def refresh(self):
        print("Refreshing sensors")
        ...

    def history(self, alive=True):
        print('Reading History')
        ...

Usage

if __name__ == '__main__':
    oWire = One_Wire()

    oWire.search()
    print(oWire)

    oWire.refresh()
    print(oWire)

    oWire.read()
    print(oWire)

    history = oWire.history()
    print(history)

Output

Searching sensors
Sensors:5, active:5
id:i2c-0 CONNECT temp:None, id:i2c-3 CONNECT temp:None, id:i2c-1 CONNECT temp:None, id:i2c-5 CONNECT temp:None, id:i2c-6 CONNECT temp:None
Refreshing sensors
NEW Sensor:i2c-2
NEW Sensor:i2c-4
Sensors:7, active:6
id:i2c-0 CONNECT temp:None, id:i2c-2 CONNECT temp:None, id:i2c-3 CONNECT temp:None, id:i2c-1 disconnect temp:None, id:i2c-5 CONNECT temp:None, id:i2c-6 CONNECT temp:None, id:i2c-4 CONNECT temp:None
Reading sensors
Reading sensors
Reading sensors
Sensors:7, active:6
id:i2c-0 CONNECT temp:19, id:i2c-2 CONNECT temp:25, id:i2c-3 CONNECT temp:26, id:i2c-1 disconnect temp:None, id:i2c-5 CONNECT temp:25, id:i2c-6 CONNECT temp:18, id:i2c-4 CONNECT temp:30
Reading History
[{'i2c-0': [20, 27, 19]}, {'i2c-2': [30, 21, 25]}, {'i2c-3': [23, 29, 26]}, {'i2c-5': [30, 18, 25]}, {'i2c-6': [30, 21, 18]}, {'i2c-4': [24, 18, 30]}]