Skip to content

feat: add register specifications for new functions #45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mirzak merged 1 commit into from
Nov 5, 2025
Merged

feat: add register specifications for new functions #45

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
247 changes: 243 additions & 4 deletions src/em511/em511.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,10 +125,91 @@ class Em511:
INPUT_MAX_VALUE_32 = 0x7FFFFFFF
INPUT_MAX_VALUE_16 = 0x7FFF

EM511_REGISTER_RESET_TOT_ENERGY_AND_RUN_HOUR_COUNTER = 0x4003
EM511_REGISTER_RESET_PARTIAL_ENERGY_AND_HOUR_COUNTER = 0x4004
EM511_REGISTER_RESET_DMD_AND_DMD_MAX = 0x4005
EM511_REGISTER_RESET_TO_FACTORY_SETTINGS = 0x4020
EM511_REGISTER_FIRMWARE_AND_REVISION = 0x0302

_register_specs: Final[dict[str, RegisterSpec]] = {
"V": RegisterSpec(address=0x0000, count=2, decimals=1, scale=10),
"A": RegisterSpec(address=0x0002, count=2, decimals=3, scale=1000),
"W": RegisterSpec(address=0x0004, count=2, decimals=1, scale=10),
"V": RegisterSpec(
address=0x0000,
count=2,
decimals=1,
scale=10,
),
"A": RegisterSpec(
address=0x0002,
count=2,
decimals=3,
scale=1000,
),
"A_dmd": RegisterSpec(
address=0x003A,
count=2,
decimals=3,
scale=1000,
),
"A_dmd_peak": RegisterSpec(
address=0x003C,
count=2,
decimals=3,
scale=1000,
),
"W": RegisterSpec(
address=0x0004,
count=2,
decimals=1,
scale=10,
),
"W_dmd": RegisterSpec(
address=0x000A,
count=2,
decimals=1,
scale=10,
),
"W_dmd_peak": RegisterSpec(
address=0x000C,
count=2,
decimals=1,
scale=10,
),
"Hz": RegisterSpec(
address=0x000F,
count=1,
decimals=1,
scale=10,
),
"kwh_tot": RegisterSpec(
address=0x0010,
count=2,
decimals=1,
scale=10,
),
"kwh_partial": RegisterSpec(
address=0x0014,
count=2,
decimals=1,
scale=10,
),
"hour_counter": RegisterSpec(
address=0x002C,
count=2,
decimals=2,
scale=100,
),
"lifetime_counter": RegisterSpec(
address=0x0030,
count=2,
decimals=2,
scale=100,
),
"hour_counter_part": RegisterSpec(
address=0x0036,
count=2,
decimals=2,
scale=100,
),
"password": RegisterSpec(
address=0x1000,
count=1,
Expand All @@ -138,7 +219,102 @@ class Em511:
return_type=int,
writable=True,
),
"alarm_status": RegisterSpec(address=0x0307, count=1, return_type=int),
"alarm_status": RegisterSpec(
address=0x0306,
count=1,
range=True,
min=0,
max=1,
return_type=int,
),
"alarm_mode": RegisterSpec(
address=0x1015,
count=1,
range=True,
min=1,
max=6,
return_type=int,
writable=True,
),
"alarm_delay": RegisterSpec(
address=0x101A,
count=1,
range=True,
min=0,
max=3600,
return_type=int,
writable=True,
),
"dmd_integration_time": RegisterSpec(
address=0x1010,
count=2,
range=True,
min=0,
max=6,
return_type=int,
writable=True,
),
"device_id": RegisterSpec(
address=0x2000,
count=1,
range=True,
min=1,
max=247,
return_type=int,
writable=True,
),
"baud_rate": RegisterSpec(
address=0x2001,
count=1,
range=True,
min=1,
max=5,
return_type=int,
writable=True,
),
"parity": RegisterSpec(
address=0x2002,
count=1,
range=True,
min=1,
max=2,
return_type=int,
writable=True,
),
"stop_bit": RegisterSpec(
address=0x2003,
count=1,
range=True,
min=0,
max=1,
return_type=int,
writable=True,
),
"reply_delay": RegisterSpec(
address=0x2004,
count=1,
range=True,
min=0,
max=1000,
return_type=int,
writable=True,
),
"identification_code": RegisterSpec(
address=0x000B,
count=1,
range=True,
min=1792,
max=1795,
return_type=int,
),
"measure_mode": RegisterSpec(
address=0x1103,
count=1,
range=True,
min=0,
max=1,
return_type=int,
),
}

def __init__(self, device_address: int, client: ModbusSerialClient) -> None:
Expand Down Expand Up @@ -211,6 +387,10 @@ def _write_register(self, address: int, value: int) -> None:
)
raise ModbusException(msg)

def write_register(self, address: int, value: int) -> None:
"""Public wrapper for `_write_register`."""
self._write_register(address, value)

def _unpack(self, regs: list[int], address: int) -> int:
"""Unpack raw Modbus register data into an integer value.

Expand Down Expand Up @@ -244,3 +424,62 @@ def _unpack(self, regs: list[int], address: int) -> int:

msg = f"Unexpected register count: {len(regs)} for address={address}"
raise ValueError(msg)

@property
def firmware_and_revision_code(self) -> str:
"""Read firmware version and revision code.

This property reads a 16-bit Modbus register that holds firmware version
and revision information.

Returns:
str: A string formatted as "<major>.<minor>,<revision>", for example "4.3,67".

Raises:
ValueError: If the raw register value is outside the valid 16-bit range
or cannot be decoded properly.
"""
regs = self._read_input_registers(self.EM511_REGISTER_FIRMWARE_AND_REVISION, self.INT16_REG_COUNT)
value = regs[0]

msb = (value >> 8) & 0xFF
revision = value & 0xFF
minor = msb & 0x0F
major = (msb >> 4) & 0x0F

return f"{major}.{minor},{revision}"

def reset_tot_energy_and_run_hour_counter(self) -> None:
"""Reset total energy + total run hour counters (excluding lifetime).

Raises:
ModbusException: If failed to write to single register.
"""
self.write_register(self.EM511_REGISTER_RESET_TOT_ENERGY_AND_RUN_HOUR_COUNTER, 1)

def reset_partial_energy_and_hour_counter(self) -> None:
"""Reset partial energy + partial run hour counters.

Raises:
ModbusException: If failed to write to single register.
"""
self.write_register(self.EM511_REGISTER_RESET_PARTIAL_ENERGY_AND_HOUR_COUNTER, 1)

def reset_dmd_and_dmd_max(self) -> None:
"""Reset DMD and DMD max values.

Raises:
ModbusException: If failed to write to single register.
"""
self.write_register(self.EM511_REGISTER_RESET_DMD_AND_DMD_MAX, 1)

def reset_to_factory_settings(self) -> None:
"""Factory Restore (Default settings).

Write 0x0A0A=2570, then within 1s write 0xC1A0=49568 to trigger reset.

Raises:
ModbusException: If failed to write to single register.
"""
self.write_register(self.EM511_REGISTER_RESET_TO_FACTORY_SETTINGS, 0x0A0A)
self.write_register(self.EM511_REGISTER_RESET_TO_FACTORY_SETTINGS, 0xC1A0)
33 changes: 33 additions & 0 deletions src/em511/em511.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,45 @@ from decimal import Decimal
from pymodbus.client import ModbusSerialClient

class Em511:
EM511_REGISTER_RESET_TOT_ENERGY_AND_RUN_HOUR_COUNTER: int
EM511_REGISTER_RESET_PARTIAL_ENERGY_AND_HOUR_COUNTER: int
EM511_REGISTER_RESET_DMD_AND_DMD_MAX: int
EM511_REGISTER_RESET_TO_FACTORY_SETTINGS: int

def __init__(self, device_address: int, client: ModbusSerialClient) -> None: ...
V: Decimal
A: Decimal
W: Decimal
W_dmd: Decimal
W_dmd_peak: Decimal
A_dmd: Decimal
A_dmd_peak: Decimal
Hz: Decimal
kwh_tot: Decimal
kwh_partial: Decimal
hour_counter: Decimal
lifetime_counter: Decimal
hour_counter_part: Decimal
password: int
alarm_status: int
alarm_mode: int
alarm_delay: int
dmd_integration_time: int
device_id: int
baud_rate: int
parity: int
stop_bit: int
reply_delay: int
identification_code: int
measure_mode: int

def _unpack(self, registers: list[int], address: int) -> int: ...
def _write_register(self, address: int, value: int) -> None: ...
def _read_register(self, register_name: str) -> Decimal | int: ...
def _read_input_registers(self, address: int, count: int) -> list[int]: ...
def write_register(self, address: int, value: int) -> None: ...
def reset_tot_energy_and_run_hour_counter(self) -> None: ...
def reset_partial_energy_and_hour_counter(self) -> None: ...
def reset_dmd_and_dmd_max(self) -> None: ...
def reset_to_factory_settings(self) -> None: ...
def firmware_and_revision_code(self) -> str: ...
Loading