Note
This module provides a class to control the RK6006 power supply via serial port using Modbus RTU protocol.
It allows reading (and writing registers - ⚠️ NOT TESTED YET), getting device status, and handling errors gracefully.
Important
Ensure you have the serialport and jsmodbus packages installed in your Node.js environment to use this module.
Warning
The code is in very early stage and it is unstable.
This is only ported python version from https://github.com/dzwer/RK6006-Python-Module/blob/main/RK6006_module.py and slightly modified.
// License: MIT
const { SerialPort } = require('serialport');
const Modbus = require('jsmodbus');
const EventEmitter = require('events');
class RK6006 extends EventEmitter {
constructor(portPath, baudRate = 115200, address = 1, modbusTimeout = 200) {
super();
this.portPath = portPath;
this.address = address;
this.baudRate = baudRate;
this.modbusTimeout = modbusTimeout;
this.voltsResolution = 100;
this.ampsResolution = 1000;
this.powerResolution = 100;
this.inVoltsResolution = 100;
this.model = "RK6006";
this.maxSetCurrent = 6;
this.maxOcpCurrent = 6.2;
this.registersMaxLen = 120;
this.port = new SerialPort({ path: this.portPath, baudRate: this.baudRate, autoOpen: false });
this.client = new Modbus.client.RTU(this.port, this.address, this.modbusTimeout);
this.port.open(async (err) => {
if (err) throw err;
await this.initialize();
console.log(`RK6006 initialized: SN=${this.sn}, FW=${this.fw}, Type=${this.type} ✅`);
this.emit('ready');
});
this.errorCounter = 0;
process.on('unhandledRejection', (reason, promise) => {
console.error('❗️Unhandled Rejection at:', promise, 'reason:', reason);
this.emit('error', reason);
});
}
handleModbusError({ err, cb, repeatCount }) {
if (Modbus.errors.isUserRequestError(err)) {
switch (err.err) {
case 'OutOfSync':
case 'Protocol':
case 'Timeout':
case 'ManuallyCleared':
case 'ModbusException':
case 'Offline':
case 'crcMismatch':
this.errorCounter++;
if (repeatCount < 3) {
console.warn(`🔁 Retrying... (${repeatCount}/3) (❗️error: ${err.err})`);
cb && typeof cb === 'function' && cb();
} else {
throw err;
}
break;
}
} else if (Modbus.errors.isInternalException(err)) {
console.error('❌ Error Message: ' + err.message, 'Error' + 'Error Name: ' + err.name, err.stack)
} else {
console.error('❌ Unknown Error', err);
}
}
async initialize() {
console.log(`Initializing RK6006 on port ${this.portPath}...`);
const regs = await this.readRegisters(0, 15);
this.sn = (regs[1] << 16) | regs[2];
this.fw = regs[3] / 100;
this.type = regs[0];
this.maxSetVoltage = Math.round((regs[14] / this.inVoltsResolution) / 1.1 - 1.5, 2);
if (this.type !== 60066) {
const error = new Error(`Detected Type: ${this.type}, Expected Type: 60066`);
this.emit('error', error);
throw error;
}
}
async readRegister(reg) {
try {
const res = await this.client.readHoldingRegisters(reg, 1);
return res.response._body.valuesAsArray[0];
} catch (e) {
return this.readRegister(reg);
}
}
async readRegisters(start, length, repeatCount = 0) {
try {
const res = await this.client.readHoldingRegisters(start, length);
return res.response._body.valuesAsArray;
} catch (e) {
this.handleModbusError({
err: e,
cb: () => this.readRegisters(start, length, repeatCount + 1),
repeatCount: repeatCount + 1
});
}
}
async writeRegister(reg, value) {
try {
await this.client.writeSingleRegister(reg, value);
} catch (e) {
return this.writeRegister(reg, value);
}
}
async getInputVoltage() {
const value = await this.readRegister(14);
this.maxSetVoltage = Math.round((value / this.inVoltsResolution) / 1.1 - 1.5, 2);
return value / this.inVoltsResolution;
}
async getSetVoltage() {
const value = await this.readRegister(8);
return value / this.voltsResolution;
}
async setVoltage(val) {
if (val < 0) val = 0;
if (val > this.maxSetVoltage) val = this.maxSetVoltage;
await this.writeRegister(8, Math.floor(val * this.voltsResolution));
}
async getOutputVoltage() {
const value = await this.readRegister(10);
return value / this.voltsResolution;
}
async getSetCurrent() {
const value = await this.readRegister(9);
return value / this.ampsResolution;
}
async setCurrent(val) {
if (val < 0) val = 0;
if (val > this.maxSetCurrent) val = this.maxSetCurrent;
await this.writeRegister(9, Math.floor(val * this.ampsResolution));
}
async getCurrentPower() {
const regs = await this.readRegisters(0, this.registersMaxLen);
if (regs) {
const power = (regs[12] << 16 | regs[13]) / this.powerResolution;
console.log(`Power: ${power}W`);
return power;
}
return null;
}
async printStatus() {
console.log("=== Print Full Status ===");
try {
const regs = await this.readRegisters(0, this.registersMaxLen);
// console.log('regs:', regs);
this.type = regs[0];
if (this.type !== 60066) {
console.log("Detected Type: ", this.type);
console.log("Expected Type: 60066");
console.log("Exit the program!");
process.exit(0);
}
console.log("=== Device ===");
console.log(`Model : ${this.model}`);
console.log(`SN : ${((regs[1] << 16) | regs[2]).toString().padStart(8, '0')}`); // SN is 4 bytes
console.log(`FW : V${regs[3] / 100}`);
console.log(`Input : ${regs[14] / this.inVoltsResolution}V`);
let sign = regs[4] ? -1 : 1;
console.log(`Int.Temp: ${sign * regs[5]}°C`);
sign = regs[34] ? -1 : 1;
let extTemp = sign * regs[35];
if (extTemp < -40) extTemp = "--";
console.log(`Ext.Temp: ${extTemp}°C`);
console.log("=== Output ===");
console.log(`Voltage : ${regs[10] / this.voltsResolution}V`);
console.log(`Current : ${regs[11] / this.ampsResolution}A`);
console.log(`Power : ${(regs[12] << 16 | regs[13]) / this.powerResolution}W`);
console.log("=== V&A SET ===");
console.log(`Voltage : ${regs[8] / this.voltsResolution}V`);
console.log(`Current : ${regs[9] / this.ampsResolution}A`);
console.log("=== OXP SET ===");
console.log(`Voltage : ${regs[82] / this.voltsResolution}V`);
console.log(`Current : ${regs[83] / this.ampsResolution}A`);
console.log("=== Energy ===");
console.log(`Charge : ${(regs[38] << 16 | regs[39]) / 1000}Ah`);
console.log(`Energy : ${(regs[40] << 16 | regs[41]) / 1000}Wh`);
console.log("=== Memories ===");
for (let m = 0; m < 10; m++) {
const offset = m * 4 + 80;
console.log(`M${m}: ${(regs[offset] / this.voltsResolution).toFixed(2)}V, ` +
`${(regs[offset + 1] / this.ampsResolution).toFixed(3)}A, ` +
`OVP: ${(regs[offset + 2] / this.voltsResolution).toFixed(2)}V, ` +
`OCP: ${(regs[offset + 3] / this.ampsResolution).toFixed(3)}A`);
}
console.log("=== End Full Status ===");
} catch (error) {
console.error("Error reading registers:", error);
}
}
}
module.exports = RK6006;
// Example usage
(async function main() {
const rk = new RK6006('/dev/tty.usbserial-21230');
rk.on('ready', async () => {
try {
await rk.printStatus();
} catch (error) {
console.error('❌ Error printing status:', error);
}
const currentPower = await rk.getCurrentPower();
console.log(`Current Power: ${currentPower}W`);
const counter = { count: 0, countOk: 0 };
setInterval(async () => {
try {
counter.count++;
const power = await rk.getCurrentPower();
power && counter.countOk++;
} catch (error) {
console.error('❌ Error getting current power:', error);
} finally {
if (counter.count % 10 === 0) {
console.log(`----------------: total-readings/ok-reads: ${counter.count} / ${counter.countOk} :----------------`);
}
}
}, 200);
});
rk.on('error', (err) => {
console.error('Error ❗️:', err);
});
process.on('SIGINT', () => {
console.log(' Ω Exiting...');
rk.port.close(() => {
console.log('Port closed.');
process.exit(0);
});
});
})();example output:
click to show output
node rk6006.js
Initializing RK6006 on port /dev/tty.usbserial-21230...
RK6006 initialized: SN=16186, FW=1.11, Type=60066 ✅
=== Print Full Status ===
=== Device ===
Model : RK6006
SN : 00016186
FW : V1.11
Input : 19.04V
Int.Temp: 39°C
Ext.Temp: --°C
=== Output ===
Voltage : 12.52V
Current : 0.188A
Power : 2.35W
=== V&A SET ===
Voltage : 12.5V
Current : 6.05A
=== OXP SET ===
Voltage : 62V
Current : 6.2A
=== Energy ===
Charge : 3.862Ah
Energy : 48.356Wh
=== Memories ===
M0: 12.50V, 6.050A, OVP: 62.00V, OCP: 6.200A
M1: 5.25V, 6.050A, OVP: 62.00V, OCP: 6.200A
M2: 12.20V, 6.050A, OVP: 62.00V, OCP: 6.200A
M3: 5.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
M4: 19.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
M5: 5.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
M6: 5.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
M7: 5.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
M8: 5.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
M9: 5.00V, 6.100A, OVP: 62.00V, OCP: 6.200A
=== End Full Status ===
Power: 2.35W
Power: 2.35W
Power: 2.35W
Power: 2.24W
Power: 2.24W
Power: 2.24W
Power: 2.24W
Power: 2.4W
Power: 2.42W
Power: 2.64W
----------------: total-readings/ok-reads: 10 / 10 :----------------
Power: 2.76W
Power: 2.87W
Power: 2.84W
Power: 2.96W
Power: 2.95W
Power: 2.87W
Power: 2.81W
Power: 2.71W
Power: 2.56W
Power: 2.36W
----------------: total-readings/ok-reads: 20 / 20 :----------------
Power: 2.31W
Power: 2.26W
Power: 2.24W
Power: 2.24W
Power: 2.24W
Power: 2.31W
Power: 2.31W
Power: 2.34W
Power: 2.34W
Power: 2.34W
----------------: total-readings/ok-reads: 30 / 30 :----------------
Power: 2.26W
^C Ω Exiting...
Port closed.original python code
https://github.com/dzwer/RK6006-Python-Module/blob/main/RK6006_module.py
import minimalmodbus
class RK6006:
def __init__(self, port, baudrate=115200, address=1, modbus_timeout=0.5):
self.port = port
self.address = address
self.instrument = minimalmodbus.Instrument(port=port, slaveaddress=address)
self.instrument.serial.baudrate = baudrate
minimalmodbus.TIMEOUT = modbus_timeout
regs = self._read_registers(0, 15)
self.sn = regs[1] << 16 | regs[2]
self.fw = regs[3] / 100
self.type = int(regs[0])
self.volts_resolution = 100
self.amps_resolution = 1000
self.power_resolution = 100
self.in_volts_resolution = 100
self.model = "RK6006"
self.max_set_voltage = round((regs[14] / self.in_volts_resolution)/1.1 - 1.5, 2) # max possible output voltage
self.max_set_current = 6 # max possible output current for the module
self.max_ocp_current = 6.2 # max OverCurrent Protection current
self.registers_max_len = 120 # https://github.com/Baldanos/rd6006/blob/master/registers.md
if self.type != 60066:
print("Detected Type: ", self.type)
print("Expected Type: 60066")
print("Exit the program!")
exit(0)
def __repr__(self):
return f"Model: {self.model}, SN:{self.sn}, FW:{self.fw}"
def _read_register(self, register):
try:
return self.instrument.read_register(register)
except minimalmodbus.NoResponseError:
return self._read_register(register)
def _read_registers(self, start, length):
try:
return self.instrument.read_registers(start, length)
except minimalmodbus.NoResponseError:
return self._read_registers(start, length)
except minimalmodbus.InvalidResponseError:
return self._read_registers(start, length)
def _write_register(self, register, value):
try:
return self.instrument.write_register(register, value)
except minimalmodbus.NoResponseError:
return self._write_register(register, value)
def update_max_set_voltage(self):
""" Updates the value of max_set_voltage according the current module input voltage """
self.max_set_voltage = round(self.get_input_voltage()/1.1 - 1.5, 2)
def print_saved_memory(self, mem=0):
""" Reads the 4 register of a Memory[0-9] and print on a single line"""
regs = self._read_registers(mem * 4 + 80, 4)
print(
f"M{mem}: {regs[0] / self.volts_resolution: 2.2f}V, "
f"{regs[1] / self.amps_resolution:1.3f}A, "
f"OVP: {regs[2] / self.volts_resolution:2.2f}V, "
f"OCP: {regs[3] / self.amps_resolution:1.3f}A"
)
def get_saved_memory(self, mem=0):
""" Reads the 4 register of a Memory[0-9] and returns a tuple (Voltage, Current, OVP, OCP)"""
regs = self._read_registers(mem * 4 + 80, 4)
mem_tuple = ((regs[0] / self.volts_resolution),
(regs[1] / self.amps_resolution),
(regs[2] / self.volts_resolution),
(regs[3] / self.amps_resolution))
return mem_tuple
def print_status(self):
""" Reads all registers and prints most of them"""
regs = self._read_registers(0, self.registers_max_len)
self.type = int(regs[0])
if self.type != 60066:
print("Detected Type: ", self.type)
print("Expected Type: 60066")
print("Exit the program!")
exit(0)
print("=== Print Full Status ===")
print("=== Device ===")
print(f"Model : {self.model}")
print(f"SN : {(regs[1] << 16 | regs[2]):08}") # SN is 4 bytes
print(f"FW : V{regs[3] / 100}")
print(f"Input : {regs[14] / self.in_volts_resolution}V")
if regs[4]:
sign = -1
else:
sign = +1
print(f"Int.Temp: {sign * regs[5]}°C")
if regs[34]:
sign = -1
else:
sign = +1
ext_temp = sign * regs[35]
if ext_temp < -40: # When external temp. sensor is missing, returns -71°C
ext_temp = "--"
print(f"Ext.Temp: {ext_temp}°C")
print("=== Output ===")
print(f"Voltage : {regs[10] / self.volts_resolution}V")
print(f"Current : {regs[11] / self.amps_resolution}A")
print(f"Power : {(regs[12] <<16 | regs[13]) / self.power_resolution}W")
print("=== V&A SET ===")
print(f"Voltage : {regs[8] / self.volts_resolution}V")
print(f"Current : {regs[9] / self.amps_resolution}A")
print("=== OXP SET ===")
print(f"Voltage : {regs[82] / self.volts_resolution}V")
print(f"Current : {regs[83] / self.amps_resolution}A")
print("=== Energy ===")
print(f"Charge : {(regs[38] <<16 | regs[39])/1000}Ah")
print(f"Energy : {(regs[40] <<16 | regs[41])/1000}Wh")
print("=== Memories ===")
for m in range(10):
offset = m * 4 + 80
print(f"M{m}: {regs[offset] / self.volts_resolution:2.2f}V, "
f"{regs[offset+1] / self.amps_resolution:1.3f}A, "
f"OVP: {regs[offset+2] / self.volts_resolution:2.2f}V, "
f"OCP: {regs[offset+3] / self.amps_resolution:1.3f}A")
print("=== End Full Status ===")
def get_input_voltage(self):
""" Returns the current board input voltage and updates the max_set_voltage value """
input_voltage = self._read_register(14) / self.in_volts_resolution
self.max_set_voltage = round(input_voltage/1.1 - 1.5, 2)
return input_voltage
def get_set_voltage(self):
""" Returns the set output voltage, not the current displayed voltage! """
return self._read_register(8) / self.volts_resolution
def set_voltage(self, value):
""" Sets the output voltage """
if value < 0:
value = 0
if value > self.max_set_voltage:
value = self.max_set_voltage
self._write_register(8, int(value * self.volts_resolution))
def get_output_voltage(self):
""" Returns the actual output voltage, not the set voltage! """
return self._read_register(10) / self.volts_resolution
def get_set_current(self):
""" Returns the set current, not the actual output current! """
return self._read_register(9) / self.amps_resolution
def set_current(self, value):
""" Sets the max output current """
if value < 0:
value = 0
if value > self.max_set_current:
value = self.max_set_current
self._write_register(9, int(value * self.amps_resolution))
def get_output_current(self):
""" Returns the actual output current, not the set current! """
return self._read_register(11) / self.amps_resolution
def get_output_power(self):
""" Returns the actual output power """
return (self._read_register(12) << 16 | self._read_register(13)) / self.power_resolution
def get_capacity_ah(self):
""" Returns the actual consumed capacity in Ah from the start of the module """
return (self._read_register(38) << 16 | self._read_register(39)) / 1000
def get_energy_wh(self):
""" Returns the actual consumed energy in Wh from the start of the module """
return (self._read_register(40) << 16 | self._read_register(41)) / 1000
def get_ovp_voltage(self):
""" Returns the set OverVoltage Protection voltage"""
return self._read_register(82) / self.volts_resolution
def set_ovp_voltage(self, value):
""" Sets the OverVoltage Protection voltage"""
if value < 0:
value = 0
if value > self.max_set_voltage + 2:
value = self.max_set_voltage + 2
self._write_register(82, int(value * self.volts_resolution))
def get_ocp_current(self):
""" Returns the set OverCurrent Protection current"""
return self._read_register(83) / self.amps_resolution
def set_ocp_current(self, value):
""" Sets the OverCurrent Protection current"""
if value < 0:
value = 0
if value > self.max_ocp_current:
value = self.max_ocp_current
self._write_register(83, int(value * self.amps_resolution))
def get_temp_internal(self):
""" Returns board temperature in Celsius"""
if self._read_register(4):
return -1 * self._read_register(5)
else:
return 1 * self._read_register(5)
def get_temp_f_internal(self):
""" Returns board temperature in Fahrenheit"""
if self._read_register(6):
return -1 * self._read_register(7)
else:
return 1 * self._read_register(7)
def get_temp_external(self):
""" Returns external temperature in Celsius"""
if self._read_register(34):
return -1 * self._read_register(35)
else:
return 1 * self._read_register(35)
def get_temp_f_external(self):
""" Returns external temperature in Fahrenheit"""
if self._read_register(36):
return -1 * self._read_register(37)
else:
return 1 * self._read_register(37)
def get_enable_state(self):
""" Returns the actual module output state - 0: OFF, 1: ON """
return self._read_register(18)
def set_enable_state(self, value):
""" Sets the module output state - 0/False: OFF, 1/True: ON """
value = int(value)
if value < 0:
value = 0
if value > 1:
value = 1
self._write_register(18, value)
def get_protection_status(self):
""" Returns the Protection Status - 0: OK, 1: OVP, 2: OCP """
return self._read_register(16)
def get_current_output_mode(self):
""" Returns the current CV/CC mode - 0: CV, 1: CC """
return self._read_register(17)
def get_backlight(self):
""" Returns the current backlight level: 0..5 """
return self._read_register(72)
def set_backlight(self, value):
""" Sets the backlight level: 0..5 """
if value < 0:
value = 0
if value > 5:
value = 5
self._write_register(72, value)
def get_take_out(self):
""" Returns the current Take_Out state - 0: OFF, 1: ON """
return self._read_register(67)
def set_take_out(self, value):
""" Sets the Take_Out state - 0: OFF, 1: ON """
value = int(value)
if value < 0:
value = 0
if value > 1:
value = 1
self._write_register(67, value)
def get_boot_power(self):
""" Returns the current Boot Power state - 0: OFF, 1: ON """
return self._read_register(68)
def set_boot_power(self, value):
""" Sets the Boot Power state - 0: OFF, 1: ON """
value = int(value)
if value < 0:
value = 0
if value > 1:
value = 1
self._write_register(68, value)
def get_buzzer(self):
""" Returns the current Buzzer state - 0: OFF, 1: ON """
return self._read_register(69)
def set_buzzer(self, value):
""" Sets the Buzzer state - 0: OFF, 1: ON """
value = int(value)
if value < 0:
value = 0
if value > 1:
value = 1
self._write_register(69, value)
# Serial port timeouts
@property
def read_timeout(self):
return self.instrument.serial.timeout
@read_timeout.setter
def read_timeout(self, value):
self.instrument.serial.timeout = value
@property
def write_timeout(self):
return self.instrument.serial.write_timeout
@write_timeout.setter
def write_timeout(self, value):
self.instrument.serial.write_timeout = value
# You can run the module as a self check script
if __name__ == "__main__":
import serial.tools.list_ports
ports = list(serial.tools.list_ports.comports())
for p in ports:
if "VID:PID=1A86:7523" in p[2]:
print(p)
rd = RK6006(p[0])
break
else:
raise Exception("Port not found")
rd.print_status()
power consumption of Odroid H4:
power consumption log