树莓派python485设备通信

Posted 2022-02-09 三明治开发社区

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了树莓派python485设备通信相关的知识，希望对你有一定的参考价值。

树莓派python485设备通信

1、程序设计简单概述

树莓派通过Modbus-Rtu协议采集温湿度传感器数据以及门磁的状态，以及控制继电器动作。

本demo完整工程

2、程序设计逻辑分析

在temp_hum_door.py文件中实现整个控制逻辑：

温湿度和门磁状态获取函数里面引用类（从new_relay_control文件中引用relay ），执行相应的控制逻辑；
通过控制id_value值执行不同的逻辑控制。
温湿度和门磁开关数据是循环获取的。

# -*- coding: utf-8 -*-
from new_relay_control import relay
from time import sleep

#温湿度获取

def temp_hum_sensor_get():
    temp_hum = relay()
    temp_hum.all_relay = 3
    temp_hum.relay_all_on_order = ['01 04 00 00 00 02 71 CB']
    return_str = temp_hum.ALL_ON()
    return return_str

#门磁开关获取

def door_sensor_get():
    door_sensor = relay()
    door_sensor.all_relay = 3
    door_sensor.relay_all_on_order = ['FE 01 00 00 00 02 A9 C4']
    return_str = door_sensor.ALL_ON()
    return return_str

id_value =2
action =1
time = 1

if id_value == 1:

    relay_open = relay()

    if action == 1:
        return_str = relay_open.ALL_ON()
    else:
        return_str = relay_open.ALL_OFF()

    relay_str = return_str[8:12]
    if relay_str == "FF00":
        relay_state = 1
    else:
        relay_state = 0

    print(relay_str)
    print(relay_state)

if id_value == 2:

    while True:

        return_str = temp_hum_sensor_get()
        print(return_str)

        get_str1 = return_str[6:10]
        get_str2 = return_str[10:14]
        try:
            temp_data = (int(get_str1, 16)) / 10
            hum_data = (int(get_str2, 16)) / 10
        except ValueError:
            pass
                
        final_str2 = get_str1 + ' ' + get_str2
        print(final_str2)
        print(temp_data)
        print(hum_data)
        sleep(time)

if id_value == 3:
    while True:
        return_str = door_sensor_get()
        get_str3 = return_str[6:8]
        print(get_str3)
        sleep(time)

这个三个值可以自己进行控制，id_value=1，执行继电器动作；id_value=2，循环获取温湿度数据；id_value=3，循环获取门磁开关状态。
action=1，打开继电器；action=0，关闭继电器。
time = 1，表示每隔1s读取温湿度或门磁开关数据。

id_value =2
action =1
time = 1

温湿度获取函数：

def temp_hum_sensor_get():
    temp_hum = relay()
    temp_hum.all_relay = 3
    temp_hum.relay_all_on_order = ['01 04 00 00 00 02 71 CB']
    return_str = temp_hum.ALL_ON()
    return return_str

门磁开关获取函数：

def door_sensor_get():
    door_sensor = relay()
    door_sensor.all_relay = 3
    door_sensor.relay_all_on_order = ['FE 01 00 00 00 02 A9 C4']
    return_str = door_sensor.ALL_ON()
    return return_str

使用class定义类，实现继电器的控制逻辑。类里面定义了串口收发的函数relay_send(self, send_order)以及继电器打开与关闭的控制逻辑。

# -*- coding: utf-8 -*-
import RPi.GPIO as GPIO
import serial
from time import sleep

'''2路继电器开关控制函数，单独继电器开关控制和全部开关控制'''

class relay(object):
    def __init__(self):
        self.relay_all_on_order = ['02 05 00 00 FF 00 8C 09', '02 05 00 01 FF 00 DD C9',
                                   '02 0F 00 00 00 08 01 FF FE C0']
        self.relay_all_off_order = ['02 05 00 00 00 00 CD F9', '02 05 00 01 00 00 9C 39',
                                    '02 0F 00 00 00 08 01 00 BE 80']
        self.relay1 = 1
        self.relay2 = 2
        self.all_relay = 3
        self.port = '/dev/ttyAMA0'

    def relay_send(self, send_order):
        if self.port:
            relay_serial = serial.Serial(self.port, 9600)
            GPIO.setmode(GPIO.BCM)
            GPIO.setup(17, GPIO.OUT)
            if not relay_serial.isOpen():
                relay_serial.Open()
            while True:
                GPIO.output(17, GPIO.HIGH)
                sleep(0.01)
                relay_serial.write(bytes.fromhex(send_order))
                # relay_serial.write(bytes(send_order))
                sleep(0.01)
                GPIO.output(17, GPIO.LOW)
                count = relay_serial.inWaiting()
                if count > 0:
                    GPIO.output(17, GPIO.LOW)
                    sleep(0.01)
                    recv = relay_serial.read(count)
                    GPIO.output(17, GPIO.HIGH)
                    sleep(0.01)
                    print("recv: ", recv)
                    # recv_bytes = binascii.b2a_hex(recv)
                    # recv_str = binascii.b2a_hex(recv_bytes).decode('utf-8')
                    recv_str = str(recv.hex())
                    print("recv_str: ", recv_str)
                    if recv_str == "00":
                        print("error")
                    else:
                        return recv_str
                sleep(0.5)

    def ALL_ON(self):
        send_order = self.relay_all_on_order[self.all_relay - 3]
        print(send_order)
        get_return = self.relay_send(send_order)
        return get_return

    def ALL_OFF(self):
        send_order = self.relay_all_off_order[self.all_relay - 3]
        get_return = self.relay_send(send_order)
        print("继电器控制: ALL_RELAY_OFF")
        return get_return

    def RELAY1_ON(self):
        send_order = self.relay_all_on_order[self.relay1 - 1]
        get_return = self.relay_send(send_order)
        print("继电器控制: RELAY1_ON")
        return get_return

    def RELAY1_OFF(self):
        send_order = self.relay_all_off_order[self.relay1 - 1]
        get_return = self.relay_send(send_order)
        print("继电器控制: RELAY1_OFF")
        return get_return

    def RELAY2_ON(self):
        send_order = self.relay_all_on_order[self.relay2 - 1]
        get_return = self.relay_send(send_order)
        print("继电器控制: RELAY2_ON")
        return get_return

    def RELAY2_OFF(self):
        send_order = self.relay_all_off_order[self.relay2 - 1]
        get_return = self.relay_send(send_order)
        print("继电器控制: RELAY2_OFF")
        return get_return

串口发送与接收函数：

采用的485转TTL模块，DIR脚控制接收与发送。当DIR高电平的时候为发送模式，当DIR为低电平的时候为接收模式。
DIR脚采用GPIO管脚控制即可。

 def relay_send(self, send_order):
        if self.port:
            relay_serial = serial.Serial(self.port, 9600)
            GPIO.setmode(GPIO.BCM)
            GPIO.setup(17, GPIO.OUT)
            if not relay_serial.isOpen():
                relay_serial.Open()
            while True:
                GPIO.output(17, GPIO.HIGH)
                sleep(0.01)
                relay_serial.write(bytes.fromhex(send_order))
                # relay_serial.write(bytes(send_order))
                sleep(0.01)
                GPIO.output(17, GPIO.LOW)
                count = relay_serial.inWaiting()
                if count > 0:
                    GPIO.output(17, GPIO.LOW)
                    sleep(0.01)
                    recv = relay_serial.read(count)
                    GPIO.output(17, GPIO.HIGH)
                    sleep(0.01)
                    print("recv: ", recv)
                    # recv_bytes = binascii.b2a_hex(recv)
                    # recv_str = binascii.b2a_hex(recv_bytes).decode('utf-8')
                    recv_str = str(recv.hex())
                    print("recv_str: ", recv_str)
                    if recv_str == "00":
                        print("error")
                    else:
                        return recv_str
                sleep(0.5)

该函数在类实例化时会自动调用。（两路继电器的控制指令，串口端口的设置等）

 def __init__(self):
        self.relay_all_on_order = ['02 05 00 00 FF 00 8C 09', '02 05 00 01 FF 00 DD C9',
                                   '02 0F 00 00 00 08 01 FF FE C0']
        self.relay_all_off_order = ['02 05 00 00 00 00 CD F9', '02 05 00 01 00 00 9C 39',
                                    '02 0F 00 00 00 08 01 00 BE 80']
        self.relay1 = 1
        self.relay2 = 2
        self.all_relay = 3
        self.port = '/dev/ttyAMA0'

3、调试过程中注意事项

子设备通信程序需要在python3以上版本进行编译运行，如果低版本，编译运行时会报错。
有些写法，低版本的不支持。例如下面写法，python2.7版本会报错，python3以上版本则不会。

        temp_value = int(get_str1, 16)
        hum_value = int(get_str2, 16)

下面编码声明在python3以上版本则不用，在低版本需要声明，不然运行会报错。

# -*- coding: utf-8 -*-

下面写法同样在python3以上版本适用，在低版本中则不支持，不然运行会报错。

 recv_str = str(recv.hex())

若在程序中遇到异常，可以添加捕获异常处理。

以上是关于树莓派python485设备通信的主要内容，如果未能解决你的问题，请参考以下文章

使用树莓派远程控制灯继电器开关，dht11温湿度网页显示，树莓派物联网

树莓派4B 串口通信

树莓派与Arduino Leonardo使用NRF24L01无线模块通信之基于RF24库树莓派查询子节点温湿度数据

树莓派使用1M CAN通讯

树莓派连接DHT11温湿度传感器（python）

物联网毕业设计树莓派(大学记录)