add source code and description

This commit is contained in:
King Kévin 2015-05-26 20:32:29 +02:00
commit 555aecf318
6 changed files with 612 additions and 0 deletions

45
README.md Normal file
View File

@ -0,0 +1,45 @@
scripts to communicate with EA-PS 2000 B family power supplies (i.e. EA-PS 2084-03 B)
links
=====
manufacturer: EA elektro-automatik http://www.elektroautomatik.de/en/
product: PS 2000 B http://shop.elektroautomatik.de/shop/shop__Series%20PS%202000B__1::4::14::42__en_GB
driver: USB ACM http://www.elektroautomatik.de/de/ps2000b.html
tool: easyPS2000 http://www.elektroautomatik.de/en/easyps2000-en.html
programming manual: http://www.elektroautomatik.de/files/eautomatik/treiber/ps2000b/programming_ps2000b.zip
files
=====
telegram.rb
-----------
library to create, parse, and decode telegrams used to communicate with the power supply
demo.rb
-------
demonstration script.
read power supply information, set to 42 V for 10 s, and read every second the set and actual values
control.rb
----------
control the power supply to go from 0 V to 84 V in 0.1 V steps (@ 1 A) and read set, actual, and measured values using two multi-meters.
the two DMM are UNI-T UT61E, one connected using the UT02 cable and measuring voltage, the other using the UT04 cable and measuring amapere.
to read the DMM values sigrock-cli is used.
probe.rb
--------
test which object in telegrams exist.
mitm.rb
-------
small script to monitor the communication:
- connect a windows compute over serial to this compute (using port ttyUSB0)
- connect the power supply to this computer (using port ttyACM0)
- start the mitm.rb script on this computer
- start easyPS2000 software on windows computer (it will use the serial link to this computer)

119
control.rb Executable file
View File

@ -0,0 +1,119 @@
#!/usr/bin/env ruby
# encoding: utf-8
# ruby: 2.2
require 'serialport'
require './telegram'
DEBUG = false
@serial = SerialPort.open("/dev/ttyACM0",{ baud: 57600, databits: 8, parity: SerialPort::ODD, stop_bit: 1, flow_control: SerialPort::NONE})
# send query/send telegram and return answer
def comm(query,data=nil)
telegram = Telegram.new(query,data)
puts("< "+(telegram.pack.bytes.collect{|b| "%02x" % b})*" ") if DEBUG
@serial.write(telegram.pack)
answer = @serial.readpartial(3+16+1+2)
raise "no answer received to query #{query}" if !answer or answer.empty?
puts("> "+(answer.bytes.collect{|b| "%02x" % b})*" ") if DEBUG
telegram = Telegram.parse(answer)
raise "malformed answer" unless telegram
raise telegram.to_s if (telegram.object==0 and telegram.data.length==2 and telegram.data[0]==0xff) or (telegram.object==255 and telegram.data[0]!=0)
return telegram
end
# get the nominal values (voltage, current)
def nominal_values
return [comm(2).data.pack("C*").unpack("g")[0],comm(3).data.pack("C*").unpack("g")[0]]
end
# get actual values (voltage, current)
def actual_values
telegram = comm(71)
voltage = telegram.data[2,2].pack("C*").unpack("n")[0]/25600.0
current = telegram.data[4,2].pack("C*").unpack("n")[0]/25600.0
nominal = nominal_values
return [voltage*nominal_values[0],current*nominal_values[1]]
end
# get protection values (voltage, current)
def protection_values(voltage=nil,current=nil)
return set_values(voltage,current,true)
end
# set values or get set values (voltage, current)
def set_values (voltage=nil,current=nil,protection=false)
# set query obj and limit
query = (protection ? [38,39,1.1] : [50,51,1.0])
nominal = nominal_values
if voltage then
raise "#{voltage} out of voltage range [0-#{query[2]*nominal[0]}]" if voltage<0 or voltage>query[2]*nominal[0]
comm(54,[0x10,0x10]) # enable remote
value = voltage*25600.0/nominal[0]
comm(query[0],[value].pack('n').bytes)
end
if current then
comm(54,[0x10,0x10]) # enable remote
raise "#{current} out of current range [0-#{query[2]*nominal[1]}]" if current<0 or current>query[2]*nominal[1]
value = current*25600.0/nominal[1]
comm(query[1],[value].pack('n').bytes)
end
voltage = comm(query[0]).data.pack("C*").unpack("n")[0]/25600.0
current = comm(query[1]).data.pack("C*").unpack("n")[0]/25600.0
nominal = nominal_values
return [voltage*nominal_values[0],current*nominal_values[1]]
end
# get device information
def device_info
info = []
[8,19,0,6,1,9].each do |query|
info << comm(query).to_s[2..-1]
end
return info
end
# set output
def output(on)
comm(54,[0x10,0x10]) # enable remote
comm(54,[0x01, on ? 0x01: 0x00]) # set on
end
# set output to on
def on
output(true)
end
# set output
def off
output(false)
end
########
# main #
########
# print device information
puts device_info*"\n"
# set to 0 and let the capacitor discharge
puts "setting to 0"
off
set_values(0.0,1.0)
on
sleep 10
puts "increasing voltage"
puts "set voltage (V),actual voltage (V), measured voltage (V),set ampere (A),actual ampere (A),measured ampere (A)"
voltage = 0.0 # start value
while voltage<=84.0 do
set_values(voltage)
sleep 3
actual = actual_values
set = set_values
measured_voltage = `sigrok-cli --driver uni-t-ut61e-ser:conn=/dev/ttyUSB0 --samples 1 -O analog`.split(" ")[1].to_f
measured_ampere = `sigrok-cli --driver uni-t-ut61e:conn=1a86.e008 --samples 1 -O analog`.split(" ")[1].to_f
puts [set[0],actual[0],measured_voltage,set[1],actual[1],measured_ampere]*","
voltage += 0.1
end
off

115
demo.rb Executable file
View File

@ -0,0 +1,115 @@
#!/usr/bin/env ruby
# encoding: utf-8
# ruby: 2.2
require 'serialport'
require './telegram'
DEBUG = false
@serial = SerialPort.open("/dev/ttyACM0",{ baud: 57600, databits: 8, parity: SerialPort::ODD, stop_bit: 1, flow_control: SerialPort::NONE})
# send query/send telegram and return answer
def comm(query,data=nil)
telegram = Telegram.new(query,data)
puts("< "+(telegram.pack.bytes.collect{|b| "%02x" % b})*" ") if DEBUG
@serial.write(telegram.pack)
answer = @serial.readpartial(3+16+1+2)
raise "no answer received to query #{query}" if !answer or answer.empty?
puts("> "+(answer.bytes.collect{|b| "%02x" % b})*" ") if DEBUG
telegram = Telegram.parse(answer)
raise "malformed answer" unless telegram
raise telegram.to_s if (telegram.object==0 and telegram.data.length==2 and telegram.data[0]==0xff) or (telegram.object==255 and telegram.data[0]!=0)
return telegram
end
# get the nominal values (voltage, current)
def nominal_values
return [comm(2).data.pack("C*").unpack("g")[0],comm(3).data.pack("C*").unpack("g")[0]]
end
# get actual values (voltage, current)
def actual_values
telegram = comm(71)
voltage = telegram.data[2,2].pack("C*").unpack("n")[0]/25600.0
current = telegram.data[4,2].pack("C*").unpack("n")[0]/25600.0
nominal = nominal_values
return [voltage*nominal_values[0],current*nominal_values[1]]
end
# get protection values (voltage, current)
def protection_values(voltage=nil,current=nil)
return set_values(voltage,current,true)
end
# set values or get set values (voltage, current)
def set_values (voltage=nil,current=nil,protection=false)
# set query obj and limit
query = (protection ? [38,39,1.1] : [50,51,1.0])
nominal = nominal_values
if voltage then
raise "#{voltage} out of voltage range [0-#{query[2]*nominal[0]}]" if voltage<0 or voltage>query[2]*nominal[0]
comm(54,[0x10,0x10]) # enable remote
value = voltage*25600.0/nominal[0]
comm(query[0],[value].pack('n').bytes)
end
if current then
comm(54,[0x10,0x10]) # enable remote
raise "#{current} out of current range [0-#{query[2]*nominal[1]}]" if current<0 or current>query[2]*nominal[1]
value = current*25600.0/nominal[1]
comm(query[1],[value].pack('n').bytes)
end
voltage = comm(query[0]).data.pack("C*").unpack("n")[0]/25600.0
current = comm(query[1]).data.pack("C*").unpack("n")[0]/25600.0
nominal = nominal_values
return [voltage*nominal_values[0],current*nominal_values[1]]
end
# get device information
def device_info
info = []
[8,19,0,6,1,9].each do |query|
info << comm(query).to_s[2..-1]
end
return info
end
# set output
def output(on)
comm(54,[0x10,0x10]) # enable remote
comm(54,[0x01, on ? 0x01: 0x00]) # set on
end
# set output to on
def on
output(true)
end
# set output
def off
output(false)
end
########
# main #
########
# print device information
puts device_info*"\n"
# set to 0 and let the capacitor discharge
puts "setting to 42V for 10s"
off
set_values(42.0,0.1)
on
sleep 10
off
comm(54,[0x10,0x00])
puts "tracking values"
while true
actual = actual_values
set = set_values
protection = protection_values
puts "voltage (V): #{actual[0].round(2).to_s}/#{set[0].round(2).to_s} (max #{protection[0].round(2).to_s}), current (A): #{actual[1].round(2).to_s}/#{set[1].round(2).to_s} (max #{protection[1].round(2).to_s})"
sleep 1
end

33
mitm.rb Executable file
View File

@ -0,0 +1,33 @@
#!/usr/bin/env ruby
# encoding: utf-8
# ruby: 2.2
require 'serialport'
require './telegram'
pc_serial = SerialPort.open("/dev/ttyUSB0",{ baud: 57600, databits: 8, parity: SerialPort::ODD, stop_bit: 1, flow_control: SerialPort::NONE})
ps_serial = SerialPort.open("/dev/ttyACM0",{ baud: 115200, databits: 8, parity: SerialPort::ODD, stop_bit: 1, flow_control: SerialPort::NONE})
while true
activities = IO.select([pc_serial,ps_serial])
activities[0].each do |activity|
data = activity.readpartial(16+1+5)
line = data.unpack("C*").collect { |b| sprintf("%02X ",b) }.join
begin
telegram = Telegram.parse(data)
rescue Exception => e
puts e.to_s
end
if activity==pc_serial then
puts "pc->ps: "+line
puts "pc->ps: "+telegram.to_s if telegram
ps_serial.write(data)
elsif activity==ps_serial then
puts "ps->pc: "+line
puts "ps->pc: "+telegram.to_s if telegram
pc_serial.write(data)
else
raise "unknown source"
end
end
end

40
probe.rb Executable file
View File

@ -0,0 +1,40 @@
#!/usr/bin/env ruby
# encoding: utf-8
# ruby: 2.2
require 'serialport'
require './telegram'
@serial = SerialPort.open("/dev/ttyACM0",{ baud: 115200, databits: 8, parity: SerialPort::ODD, stop_bit: 1, flow_control: SerialPort::NONE})
@serial.dtr = 1
=begin
# start queries
[0,1,6,8,9,19,71,2,3,4,38,39,50,51,54,71,72].each do |query|
telegram = Telegram.new(query)
puts telegram
@serial.write(telegram.pack)
data = @serial.readpartial(16+1+5)
if data and !data.empty? then
telegram = Telegram.parse(data)
puts telegram
else
puts "empty"
end
end
=end
256.times do |object|
telegram = Telegram.new(object)
@serial.write(telegram.pack)
data = @serial.readpartial(16+1+5)
if data and !data.empty? then
#puts data.unpack("C*").collect { |b| sprintf("%02X ",b) }.join
telegram = Telegram.parse(data)
if telegram.object==0xff and telegram.data[0]!=0x07 then
puts "#{object}: #{telegram.to_s}"
end
else
puts "empty"
end
end

260
telegram.rb Normal file
View File

@ -0,0 +1,260 @@
# encoding: utf-8
# ruby: 2.2
class Telegram
attr_accessor :direction, :cast, :transmission, :node, :object, :data
# transmission types
RESERVED = 0
QUERY = 1
ANSWER = 2
SEND = 3
# object types
OBJECTS = []
OBJECTS[0] = "device type"
OBJECTS[1] = "serial no."
OBJECTS[2] = "nominal voltage (V)"
OBJECTS[3] = "nominal current (A)"
OBJECTS[4] = "nominal power (W)"
OBJECTS[6] = "article no."
OBJECTS[8] = "manufacturer"
OBJECTS[9] = "software version"
OBJECTS[19] = "device class"
OBJECTS[38] = "over voltage protection threshold (%V)"
OBJECTS[39] = "over current protection threshold (%A)"
OBJECTS[50] = "set voltage (%V)"
OBJECTS[51] = "set current (%I)"
OBJECTS[54] = "power supply control"
OBJECTS[71] = "status and actual values"
OBJECTS[72] = "status and set values"
OBJECTS[255] = "error"
# length of the object type
LENGTHS = []
LENGTHS[0] = 16
LENGTHS[1] = 16
LENGTHS[2] = 4
LENGTHS[3] = 4
LENGTHS[4] = 4
LENGTHS[6] = 16
LENGTHS[8] = 16
LENGTHS[9] = 16
LENGTHS[19] = 2
LENGTHS[38] = 2
LENGTHS[39] = 2
LENGTHS[50] = 2
LENGTHS[51] = 2
LENGTHS[54] = 2
LENGTHS[71] = 6
LENGTHS[72] = 6
LENGTHS[255] = 1
# possibles errors
ERRORS = []
ERRORS[0] = "no error"
ERRORS[3] = "checksum incorrect"
ERRORS[4] = "start delimiter incorrect"
ERRORS[5] = "wrong address for output"
ERRORS[7] = "object not defined"
ERRORS[8] = "object length incorrect"
ERRORS[9] = "no access permission"
ERRORS[15] = "device in lock state"
ERRORS[48] = "upper limit exceeded"
ERRORS[49] = "lower limit exceeded"
# create a query or send telegram for this object
# query if data is nil, else send data
def initialize (object, data=nil)
# telegram direction: true = control unit to device, false = device to control unit
@direction = true
# cast type: true = query, false = answer
@cast = true
# device node
@node = 0
# set object
@object = object
# verify data
if data==nil or data.empty? then
@transmission = QUERY
@data = []
else
raise "wrong data length" if LENGTHS[@object] and data.length!=LENGTHS[@object]
@transmission = SEND
@data = data
end
end
# create a Telegram from the raw telegram data
def Telegram.parse (telegram)
# check there are at least 5 bytes (minimum message size)
return nil if telegram==nil
return nil if telegram.length<5
bytes = telegram.bytes # get bytes
to_return = new(bytes[2]) # new Telegram
# parse start delimiter (SD)
length = bytes[0]&0x0f # get length
to_return.direction = (bytes[0]&0x10!=0)
to_return.cast = (bytes[0]&0x20!=0)
to_return.transmission = ((bytes[0]>>6)&0x03)
# parse device node (DN)
to_return.node = bytes[1]
# parse object (OBJ)
to_return.object = bytes[2]
# parse data field
to_return.data = bytes[3..-3]
# parse checksum (CS)
checksum = (bytes[-2]<<8)+bytes[-1]
# run some checks
raise "wrong length. expected #{length}, got #{to_return.data.length-1}" if (to_return.transmission==SEND or to_return.transmission==ANSWER) and length != to_return.data.length-1
raise "wrong checksum. expected #{to_return.checksum}, got #{checksum}" if checksum != to_return.checksum
return to_return
end
# is the telegram from the control unit to the device
def to_device?
return @direction
end
# is the telegram from the device to the control unit
def to_control?
return !@direction
end
# is the telegram a query
def query?
return @cast
end
# is the telegram an answer
def answer?
return !@cast
end
# pack all except checksum
def pack_data
# make start delimiter
raise "too much data" if data.length-1>0xf
length = nil
if @transmission==ANSWER or @transmission==RESERVED then
raise "wrong data field length. expected #{LENGTHS[@object]}, got #{data.length}" if LENGTHS[@object] and data.length>LENGTHS[@object]
end
if @data and !@data.empty? then
start = data.length-1
elsif LENGTHS[@object] then
start = LENGTHS[@object]-1
else
start = 0x0f
end
start += 1<<4 if @direction
start += 1<<5 if @cast
start += @transmission<<6
# add rest
return [start,@node,@object]+@data
end
# pack telegram as string
def pack
# calculate checksum
data = pack_data
data += [checksum>>8,checksum&0xff]
return data.pack("C*")
end
# calculate checksum
def checksum
# calculate checksum
data = pack_data
cs = 0
data.each { |b| cs += b }
return cs
end
# check packet
def check_direction
raise "wrong direction" if (@direction and !@cast) or (@cast and !(@transmission==QUERY or @transmission==SEND)) or (!@direction and @cast) or (!@cast and !(@transmission==ANSWER or @transmission==RESERVED))
end
def to_s
str = @direction ? "<" : ">"
if OBJECTS[@object] then
if @object==0 and @data.length==2 and @data[0]==0xFF then # bug in the firmware. error should use object FF but uses object 00 and first byte is FF. second byte is error"
str += " error"
else
str += " "+OBJECTS[@object]
end
else
str += " #{@object}"
end
if @data and !@data.empty? then
str += ": "
case @object
when 0 # string or error (that's a bug)
if @data.length==2 and @data[0]==0xFF then # error
str += (ERRORS[@data[1]] or "unknown")
else # string
data = @data.pack("C*")
str_end = data.index("\0")
str_end = data.length unless str_end
str += data[0,str_end]
end
when 1,6,8,9 # strings
data = @data.pack("C*")
str_end = data.index("\0")
str_end = data.length unless str_end
str += data[0,str_end]
when 2,3,4 # float
str += @data.pack("C*").unpack("g")[0].to_s
when 19 # id
str += if @data == [0x00,0x10] then
"PS 2000 B Single"
elsif @data == [0x00,0x18] then
"PS 2000 B Triple"
else
"unknown"
end
when 38,39,50,51 # percentage
str += (@data.pack("C*").unpack("n")[0]/256.0).round(3).to_s
when 54 # changes
changes = []
changes << ((@data[1]&0x01)==0 ? "output off" : "output on")
changes[-1] += " (changed)" if (@data[0]&0x01)!=0
changes << ((@data[1]&0x0A)==0 ? nil : "acknowledge alarm")
changes[-1] += " (changed)" if (@data[0]&0x0A)!=0
changes << ((@data[1]&0x10)==0 ? "manual control" : "remote control")
changes[-1] += " (changed)" if (@data[0]&0x10)!=0
changes << "tracking on" if @data[1]&0xF0 == 0xF0
changes << "tracking off" if @data[1]&0xF0 == 0xF0
changes[-1] += " (changed)" if (@data[0]&0xF0)!=0
str += changes.compact*", "
when 71,72 # status + values
status = []
status << if @data[0]&0x03 == 0x00 then
"free access"
elsif @data[0]&0x03 == 0x01 then
"free access"
else
"unknown access"
end
status << ((@data[1]&(1<<1))==0 ? "output off" : "output on")
status << if @data[1]&0x06 == 0x00 then
"constant voltage"
elsif @data[1]&0x06 == 0x04 then
"constant current"
else
"unknown controller state"
end
status << ((@data[1]&(1<<3))==0 ? "tracking off" : "tracking on")
status << ((@data[1]&(1<<4))==0 ? "over-voltage protection off" : "over-voltage protection on")
status << ((@data[1]&(1<<5))==0 ? "over-current protection off" : "over-current protection on")
status << ((@data[1]&(1<<6))==0 ? "over-power protection off" : "over-power protection on")
status << ((@data[1]&(1<<7))==0 ? "over-temperature protection off" : "over-temperature protection on")
str += status.compact*", "
str += ", voltage %: "+(@data[2,2].pack("C*").unpack("n")[0]/256.0).round(3).to_s
str += ", current %: "+(@data[4,2].pack("C*").unpack("n")[0]/256.0).round(3).to_s
when 255
str += (ERRORS[@data[0]] or "unknown")
else
str += @data.collect { |b| sprintf("%02X ",b) }.join if @data and !@data.empty?
end
end
return str
end
end