Implementierung des HART-Protokolls ⇐ Python

[Guest]

Ich implementieren das HART-Protokoll mit Python.
Im ersten Schritt sende ich den Befehl 0, um die 5-Byte-Adresse des Slaves mit der Funktion connect() zu empfangen:

Code: Select all

def connect(port):
ser = serial.Serial()
ser.close()
ser = serial.Serial(port, 1200, xonxoff=True, timeout=2, parity=serial.PARITY_ODD)
ser.setRTS(False)
ser.setDTR(False)
time.sleep(0.002)
ser.setRTS(True)
ser.setDTR(False)
time.sleep(0.002)
data = bytearray([0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x02,
0x00, 0x00, 0x00, ])
bcc = 0x00
for i in range(10, len(data)):
bcc ^= data[i]

data.append(bcc)
ser.setRTS(False)
ser.setDTR(False)
time.sleep(0.002)
ser.setRTS(True)
ser.setDTR(False)
time.sleep(0.002)
ser.write(data)
time.sleep(0.138)
ser.setRTS(False)
ser.setDTR(True)
time.sleep(0.002)
response = ser.read(120)

index = response.find(0xfe)
device_address = bytearray([])

device_address.append(response[index + 1])
device_address.append(response[index + 2])
device_address.append(response[index + 9])
device_address.append(response[index + 10])
device_address.append(response[index + 11])

return device_address

Danach möchte ich die Nachricht des Geräts mit Befehl 12 lesen:

Code: Select all

def CMD_12(port_name, address):
ser = serial.Serial()
ser.close()
ser = serial.Serial(port_name, 1200, xonxoff=True, timeout=2, parity=serial.PARITY_ODD)
data = []
for i in range(5):
data.append(0xff)
data.append(0x82)
for i in range(len(address)):
data.append(address[i])

data.append(0x0C)
data.append(0x00)
bcc = 0x00
for i in range(5, len(data)):
bcc ^= data[i]
data.append(bcc)
ser.setRTS(False)
ser.setDTR(False)
time.sleep(0.002)
ser.setRTS(True)
ser.setDTR(False)
time.sleep(0.002)
# print("transmitted data : ", end=" ")
# for i in data:
#     print(hex(i), end=',')
# print()

ser.write(data)
time.sleep(0.138)
ser.setRTS(False)
ser.setDTR(True)
time.sleep(0.002)
response = ser.read(120)
# print("recieved data : ", end=" ")
# for i in response:
#     print(hex(i), end=',')
# print()
response = list(response)
try:
response = utility.parse_packet(response)
except:
pass
#     print("response was not receive")
# print(response)
status = response[0]
resp_data = response[3]

if status[0] == 0:
status = "Success"
elif status[0] == 100:
status = "Not Implemented"
elif status[0] == 6:
status = "Dev-specific cmd error"
elif status[0] == 8:
status = "Update Failure"
elif status[0] == 16:
status = "Access Restricted"
else:
status = "Undefined"
return [status, [utility.packed_bytes2str(resp_data)]]

und ich habe erfolgreich eine Nachricht vom Gerät erhalten.

Aber wenn ich eine Nachricht per Befehl 17 schreiben möchte, sendet mir das Gerät keine Antwort:

Code: Select all

def CMD_17(port_name, address, message):
ser = serial.Serial()
ser.close()
ser = serial.Serial(port_name, 1200, timeout=5, parity=serial.PARITY_ODD, rtscts=True,
xonxoff=False)  # RTS/CTS enabled
data = []
# Construct the data packet
for i in range(5):
data.append(0xff)
data.append(0x82)

# Add the address to the data packet
for i in range(len(address)):
data.append(address[i])

data.append(0x11)

# Ensure message length is 32 bytes
if len(message) < 32:
a = 32 - len(message)
for i in range(a):
message += "  "
message = utility.str2pack(message)  # Assuming this function packs the message
data.append(len(message))

print("Message:", message)

# Append the message data
data += message

# Calculate BCC
bcc = 0x00
for i in range(5, len(data)):
bcc ^= data[i]
data.append(bcc)

# Ensure RTS/CTS lines are correctly toggled
ser.setRTS(False)  # Make sure RTS is low at the start
ser.setDTR(False)  # DTR is low initially
time.sleep(0.002)  # Let the modem prepare
ser.setRTS(True)   # Toggle RTS high
ser.setDTR(False)
time.sleep(0.002)

print("Transmitting data:")
for i in data:
print(hex(i), end=', ')
print()

# Write data to the serial port
ser.write(data)
time.sleep(0.138)  # Wait for modem processing

# Set DTR to high to request the response from the modem
ser.setRTS(False)
ser.setDTR(True)
time.sleep(0.002)

# Read the response from the modem
response = ser.read(120)  # Try reading up to 120 bytes
print("Received data:")
for i in response:
print(hex(i), end=', ')
print()

# Check if we received any response
if len(response) == 0:
print("No response received!")
return ["No Response", []]

response = list(response)
try:
response = utility.parse_packet(response)  # Attempt to parse the response
except Exception as e:
print(f"Failed to parse response: {e}")
return ["Error", []]

print("Parsed response:", response)
status = response[0]
resp_data = response[3]

# Return the appropriate status message
if status[0] == 0:
status = "Success"
elif status[0] == 5:
status = "Too few data bytes received"
elif status[0] == 6:
status = "Dev-specific cmd error"
elif status[0] == 7:
status = "Device is write protected"
elif status[0] == 100:
status = "Not Implemented"
elif status[0] == 16:
status = "Access Restricted"
elif status[0] == 32:
status = "Busy"
else:
status = "Undefined"

res = utility.packed_bytes2str(data[:24])  # Assuming this decodes the first 24 bytes

return [status, [res]]

Ich denke, die Befehle, die etwas vom Gerät lesen, sind korrekt, aber es gibt Probleme mit den Befehlen, die etwas auf das Gerät schreiben. Was ist das Problem?
Alles im vorherigen Abschnitt erklärt