#!/usr/bin/env python2 # -*- encoding: utf-8 -*- import optparse import os, sys import serial import numpy as N import struct import binascii import time DEBUG = False crc_test = "\x0d\x01\x00\x62\x00\x33" crc_expected = 0xddd def crc(message): """ (from "Modbus_over_serial_line_V1_02.pdf" at http://www.modbus.org) 6.2.2 CRC Generation ==================== The Cyclical Redundancy Checking (CRC) field is two bytes, containing a 16–bit binary value. The CRC value is calculated by the transmitting device, which appends the CRC to the message. The device that receives recalculates a CRC during receipt of the message, and compares the calculated value to the actual value it received in the CRC field. If the two values are not equal, an error results. The CRC is started by first preloading a 16–bit register to all 1’s. Then a process begins of applying successive 8–bit bytes of the message to the current contents of the register. Only the eight bits of data in each character are used for generating the CRC. Start and stop bits and the parity bit, do not apply to the CRC. During generation of the CRC, each 8–bit character is exclusive ORed with the register contents. Then the result is shifted in the direction of the least significant bit (LSB), with a zero filled into the most significant bit (MSB) position. The LSB is extracted and examined. If the LSB was a 1, the register is then exclusive ORed with a preset, fixed value. If the LSB was a 0, no exclusive OR takes place. This process is repeated until eight shifts have been performed. After the last (eighth) shift, the next 8–bit character is exclusive ORed with the register’s current value, and the process repeats for eight more shifts as described above. The final content of the register, after all the characters of the message have been applied, is the CRC value. A procedure for generating a CRC is: 1. Load a 16–bit register with FFFF hex (all 1’s). Call this the CRC register. 2. Exclusive OR the first 8–bit byte of the message with the low–order byte of the 16–bit CRC register, putting the result in the CRC register. 3. Shift the CRC register one bit to the right (toward the LSB), zero–filling the MSB. Extract and examine the LSB. 4. (If the LSB was 0): Repeat Step 3 (another shift). (If the LSB was 1): Exclusive OR the CRC register with the polynomial value 0xA001 (1010 0000 0000 0001). 5. Repeat Steps 3 and 4 until 8 shifts have been performed. When this is done, a complete 8–bit byte will have been processed. 6. Repeat Steps 2 through 5 for the next 8–bit byte of the message. Continue doing this until all bytes have been processed. 7. The final content of the CRC register is the CRC value. 8. When the CRC is placed into the message, its upper and lower bytes must be swapped as described below. Placing the CRC into the Message When the 16–bit CRC (two 8–bit bytes) is transmitted in the message, the low-order byte will be transmitted first, followed by the high- order byte. """ crc = 0xffff # step 1 for byte in message: if type(byte) == type(str): crc ^= ord(byte) # step 2: xor with lower byte of crc else: crc ^= byte # step 2: xor with lower byte of crc for i in range(8): # step 5: repeat 2-4 until 8 shifts were performed if (crc & 0x1) == 0: # step 4: check LSB mask = 0x0 else: mask = 0xa001 crc >>= 1 # step 3: shift one bit to the right crc ^= mask if DEBUG: print("Message:", [i for i in message], " CRC:", hex(crc)) return chr(crc & 0xff) + chr( (crc >> 8) & 0xff) # return lower byte, upper byte (shift to the right, return last byte) i class Flow: def __init__(self, device="/dev/ttyUSB0"): self.s = serial.Serial(port=device, timeout=0.02) self.s.readline() self.s.timeout = 0.2 self.address = bytearray([0x3]) self.rcommands = { "flow": [bytearray([0x3, 0x0, 0x0, 0x0, 0x2]), ">f"], "temperature": [bytearray([0x3, 0x0, 0x2, 0x0, 0x2]), ">f"], "setflow": [bytearray([0x3, 0x0, 0x6, 0x0, 0x2]), ">f"], "total": [bytearray([0x3, 0x0, 0x8, 0x0, 0x2]), ">f"], "alarm": [bytearray([0x3, 0x0, 0xc, 0x0, 0x1]), ">H"], "hwerror": [bytearray([0x3, 0x0, 0xd, 0x0, 0x1]), ">H"], "unit_flow": [bytearray([0x3, 0x0, 0x16, 0x0, 0x4]), ">8s"], "medium": [bytearray([0x3, 0x00, 0x1a, 0x0, 0x4]), ">8s"], "device": [bytearray([0x3, 0x00, 0x23, 0x0, 0x4]), ">8s"], "unit_total": [bytearray([0x3, 0x40, 0x48, 0x0, 0x4]), ">8s"], } self.wcommands = { "flow": [bytearray([0x06, 0x0, 0x6]), ">f"] } def send_data(self, data_and_type): """ send data to the device at self.address. """ data, type = data_and_type if DEBUG: print("send", data_and_type, "data:", binascii.hexlify(data), "type:", type, self.address) msg = self.address + data msg_crc = msg + crc(msg) if DEBUG: print("send:", repr(str(msg_crc))) nbytes_sent = self.s.write(str(msg_crc)) if DEBUG: print("send:", self.s) if DEBUG: print("Sent bytes:", nbytes_sent) time.sleep(0.1) # this is necessary, otherwise no data can be received return msg_crc def recv_data(self): """ Reading Data ============ Function: 0x3 Format of the resulting string Addr Func NoBytes Byte1...N CRCHi CRCLo """ message = bytearray(self.s.read(3)) if DEBUG: print("recv_data: header:", binascii.hexlify(message)) if message == "": return addr, func, nbytes = message if func == 0x03: data = self.s.read(nbytes) if DEBUG: print("recv_data: data: %i %s" % (nbytes, binascii.hexlify(data))) message += data crc_data = self.s.read(2) if DEBUG: print("recv_data: crc", repr(str(crc_data))) if crc(message) != crc_data: print("mismatch", crc(message), crc_data) return data def get_var(self, named_type="flow"): """ get variable from device, variables are definde in self.rcommands as list [bytearray[hiAd, loAd, numHi, numLo], type] """ cmd,fmt = self.rcommands[named_type] if DEBUG: print("get_var", cmd,fmt) self.send_data([cmd,fmt]) d = self.recv_data() if d: d = struct.unpack(fmt, d)[0] return d # msg = self.recv_data() # print str(msg) # float struct.unpack('>f', "4byte string") # ushort struct.unpack('>H', "2byte string") def current_flow(self): # start register 0x0000 # 2 bytes cmd = self.address + bytearray([0x3, 0x0, 0x0, 0x0, 0x2]) return cmd + crc(cmd) def current_temperature(self): # start register 0x0002 # # cmd = self.address + bytearray([0x3, 0x0, 0x2, 0x0, 0x2]) return cmd + crc(cmd) def set_flow(self, flow): """ write multiple registers: 0x10 startHi startLo numregHi numregLo numbytes """ _flow = struct.pack(">f", flow) # cmd = address + bytearray([0x10,0x0,0x6, 0x0, 0x2, 0x2]) + _flow cmd1 = self.address + bytearray([0x06, 0x0, 0x6]) + _flow self.s.write(cmd1 + crc(cmd1)) time.sleep(0.01) # print self.s.readline() # self.recv_data() def cmd(self, data): cmds = self.address + data for i in cmds: print(hex(ord(i)), end=' ') return cmds + crc(cmds) if __name__ == "__main__": """ This is the command line program to control the red-y flow control series. Usage: ./flow_control.py [FLOW] Several parameters are printed on every execution The default serial device is /dev/ttyUSB0. If you want to change that create a file called "~/.flow" with the first line containing an alternate path, i.e. /dev/ttyUSB1 If you want to enable debug output edit the script and set DEBUG=True. Sample output: user@pfg # flow Device: GSCC9SA Medium: N2B Current Flow: 0.00 Current Set Flow: 0.00 Flow Unit: ln/min Total Gas: 0.00 Total Unit: lnF Current Temperature: 22.38 C Alarm: 0 HW Error: 0 """ conf = os.path.expanduser('~/.flow') if os.path.exists(conf): dev = open(conf).readlines()[0].strip() redy = Flow(device=dev) print("Device: %6s" % (redy.get_var("device"))) print("Medium: %6s" % (redy.get_var("medium"))) print("Current Flow: %6.2f" % (redy.get_var("flow"))) print("Current Set Flow: %6.2f" % (redy.get_var("setflow"))) print("Flow Unit: %6s" % (redy.get_var("unit_flow"))) print("Total Gas: %6.2f" % (redy.get_var("total"))) print("Total Unit: %6s" % (redy.get_var("unit_total"))) print("Current Temperature: %6.2f C" % (redy.get_var("temperature"))) print("Alarm: %8i" % (redy.get_var("alarm"))) print("HW Error: %8i" % (redy.get_var("hwerror"))) if len(sys.argv) > 1: try: float(sys.argv[1]) except: raise SyntaxError("usage: %s # to set flow or empty to get current flow\nfirst line in ~/.flow defines serial tty, default: /dev/ttyUSB0" % (sys.argv[0])) print("Set flow: %6.2f"%(float(sys.argv[1]))) redy.set_flow(float(sys.argv[1]))