ekmmeter

EKM Meter Serial API.

VALID_CT_RATIOS = (100, 200, 400, 600, 800, 1000, 1200, 1500, 2000, 3000, 4000, 5000): Valid CT ratios.

VALID_PULSE_OUTPUT_RATIOS: Dict[int, Tuple[int, ...]] = {100: (1, 2, 4, 5, 8, 10, 16, 20, 25, 32, 40, 50, 64, 80, 100, 160, 200, 320, 400, 800, 1600), 200: (1, 2, 4, 5, 8, 10, 16, 20, 25, 32, 40, 50, 80, 100, 160, 200, 400, 800), 400: (1, 2, 4, 5, 8, 10, 16, 20, 25, 40, 50, 80, 100, 200, 400), 600: (1, 2, 4, 5, 8, 10, 16, 20, 25, 32, 40, 50, 80, 100, 160, 200, 400, 800), 800: (1, 2, 4, 5, 8, 10, 20, 25, 40, 50, 100, 200), 1000: (1, 2, 4, 5, 8, 10, 16, 20, 32, 40, 80, 160), 1200: (1, 2, 4, 5, 8, 10, 16, 20, 25, 40, 50, 80, 100, 200, 400), 1500: (1, 2, 4, 5, 8, 10, 16, 20, 32, 40, 64, 80, 160, 320), 2000: (1, 2, 4, 5, 8, 10, 16, 20, 40, 80), 3000: (1, 2, 4, 5, 8, 10, 16, 20, 32, 40, 80, 160), 4000: (1, 2, 4, 5, 8, 10, 20, 40), 5000: (1, 2, 4, 8, 16, 32)}: Valid pulse output ratios mapped to corresponding valid CT ratios.

VALID_MAX_DEMAND_PERIODS = (1, 2, 4)

Max demand period values.

1 15 minutes
2 30 minutes
4 1 hour

VALID_MAX_DEMAND_RESET_INTERVALS = (0, 1, 2, 3, 4)

Valid max demand reset intervals.

0 OFF
1 Monthly
2 Weekly
3 Daily
4 Hourly

MAX_SEASONS = 4: Maximum number of TOU seasons.

MAX_SCHEDULES = 6: Maximum number of TOU schedules.

MAX_HOLIDAYS = 20: Maximum number of TOU holidays.

MAX_PERIODS = 4: Maximum number of TOU periods.

MAX_TARIFFS = 4: Maximum number of TOU tariffs.

MAX_DISPLAY_ITEMS = 40: Maximum number of LCD display items.

MAX_PULSE_INPUTS = 3: Number of pulse inputs.

MAX_PULSE_INPUT_RATIO = 9999: Maximum pulse input ratio.

Read meter data and return it as a dict.

If a scale value is present (kWh_Scale) in the data it will be applied to all floating point energy values.

Parameters:

serial_port – Serial port connected to meter(s)
address – The meter address/number
version – The meter protocol version. Default is v4.
include_b – Include B data block (ignored for v3 meters). An extra READ_B request will be performed and the data will be merged with the READ_A data.
end_session – Send the end session sequence to the meter.

Returns:

Unpacked meter data as a dict, or None if read failed.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.
EKMParamError – If meter version is invalid

read_v3(serial_port: Serial, address: str | SupportsInt | bytes, end_session: bool = False, use_float: bool = True) → dict | None[source]

Read V3 meter data and return it as a dict.

Parameters:

serial_port – Serial port connected to meter(s)
address – The meter address/number
end_session – Send the end session sequence to the meter.
use_float – Use float for scaled values (ie kWh). Default is True.

Returns:

Unpacked meter data as a dict, or None if no data.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

read_v4(serial_port: Serial, address: str | SupportsInt | bytes, readtype: str = 'a', kwh_scale: int | None = None, end_session: bool = False, use_float: bool = True) → dict | None[source]

Read V4 meter data and return it as a dict.

Parameters:

serial_port – Serial port connected to meter(s)
address – The meter address/number
readtype – The data block type to read (‘A’, ‘B’, ‘C’) Default is ‘A’.
kwh_scale – The kWh decimal precision (usually the value of kWh_Scale from an A block). Only required for ‘B’ and ‘C’ data reads. Default is 1.
end_session – Send the end session sequence to the meter. Default is False.
use_float – Use float for scaled values (ie kWh). Default is True.

Returns:

Data, or None if no data.

Raises:

EKMDataError – If the data is invalid or missing or if the serial port read or write fails.

read_a(serial_port: Serial, address: str | SupportsInt | bytes, end_session: bool = False) → dict | None[source]

Read V4 meter A-block data and return it as a dict.

This is the same as calling read_v4() with readtype=’A’.

If a scale value is present (kWh_Scale) in the data it will be applied to all floating point energy values.

Parameters:

serial_port – Serial port connected to meter(s)
address – The meter address/number
end_session – Send the end session sequence to the meter.

Returns:

Block A data, or None if no data.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

read_b(serial_port: Serial, address: str | SupportsInt | bytes, kwh_scale: int = 1, end_session: bool = False) → dict | None[source]

Read V4 meter B-block data and return it as a dict.

This is the same as calling read_v4() with readtype=’B’.

Parameters:

serial_port – Serial port connected to meter(s)
address – The meter address/number
kwh_scale – The kWh decimal precision (usually the value of kWh_Scale from READ_A data)
end_session – Send the end session sequence to the meter.

Returns:

Block B data.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

read_c(serial_port: Serial, address: str | SupportsInt | bytes, kwh_scale: int = 1, end_session: bool = False) → dict | None[source]

Read V4 meter C-block data and return it as a dict.

This is the same as calling read_v4() with readtype=’C’.

Parameters:

serial_port – Serial port connected to meter(s)
address – The meter address/number
kwh_scale – The kWh decimal precision (usually the value of kWh_Scale from an A block)
end_session – Send the end session sequence to the meter.

Returns:

Block C data, or None if no data.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_password(serial_port: Serial, new_password: str | SupportsInt | bytes, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set meter password.

Parameters:

serial_port – Serial port connected to meter(s)
new_password – Max 8 digit numeric password.
password – Previous password.

Returns:

True if successful.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_relay(serial_port: Serial, relay: int, state: int, duration: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Open/Close a meter relay.

Only supported by v4 and v5 meters.

Parameters:

serial_port – Serial port connected to meter(s)
relay – The relay number (1,2)
state – 0 = Open, 1 = Close
duration – Duration in seconds (0-9999). 0 = stay open/close.
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

reset_pulse_count(serial_port: Serial, pulse_input: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Reset pulse input count.

Only supported by v4 and v5 meters.

Parameters:

serial_port – Serial port connected to meter(s)
pulse_input – Pulse input 0-based (0-2)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_ct_ratio(serial_port: Serial, ct_ratio: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set meter CT ratio.

Parameters:

serial_port – Serial port connected to meter(s)
ct_ratio – CT ratio
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_datetime(serial_port: Serial, dt: datetime, wd: int | None = None, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set meter time. See set_time().

Parameters:

serial_port – Serial port connected to meter(s)
dt – Date and time
wd – Optional weekday (Sunday=1, Monday=2, …, Saturday=7)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_time(serial_port: Serial, yy: int, mo: int, dd: int, hh: int, mm: int, ss: int, tz: tzinfo | None = None, wd: int | None = None, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set meter time.

If weekday is not specified it will be calculated based on the other date/time values and the timezone.

Parameters:

serial_port – Serial port connected to meter(s)
yy – Year. Only the last two digits of the year are used by the meter internally. Historical reasons…
mo – Month 1-12
dd – Day 1-31
hh – Hour 0-23
mm – Minute 0-59
ss – Second 0-59
tz – Optional timezone
wd – Optional weekday (Sunday=1, Monday=2, …, Saturday=7)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

reset_max_demand(serial_port: Serial, password: str | SupportsInt | bytes = '00000000') → bool[source]

Reset Max_Demand to zero.

Parameters:

serial_port – Serial port connected to meter(s)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_max_demand_period(serial_port: Serial, period: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set the max demand period.

Parameters:

serial_port – Serial port connected to meter(s)
period – Can be 1, 2, or 4, where 1 = 15m, 2 = 30m, 4 = 1hr
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_max_demand_reset_interval(serial_port: Serial, interval: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set Max Demand reset interval.

Parameters:

serial_port – Serial port connected to meter(s)
interval – reset interval (0 = off, 1 = monthly, 2 = weekly, 3 = daily, 4 = hourly)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_schedule_tariffs(serial_port: Serial, schedule: int, tariff_periods: Sequence, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set tariff periods for schedule.

Parameters:

serial_port – Serial port connected to meter(s)
schedule – The tariff schedule number (0-5)
tariff_periods – A list of tariff periods as tuples (or lists) of the form (hour, minute, tariff). Valid tariff values are 1 thru 4.
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

get_schedule_tariffs(serial_port: Serial) → dict[source]

Get the tariff periods for schedules, and the season definitions.

Parameters:: serial_port – Serial port connected to meter(s)
Returns:: A dictionary containing a list of tariff periods (hh, mm, tariff) for each schedule, and a list of seasons.
Raises:: EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_season_schedules(serial_port: Serial, season_schedules: Sequence, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set season schedules.

Parameters:

serial_port – Serial port connected to meter(s)
season_schedules – A sequence of up to four tuples or lists containing (month, day, schedule). Schedule values are 1-based in this case.
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

get_holidays(serial_port: Serial) → dict[source]

Get holiday dates.

Raises:: EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_holidays(serial_port: Serial, holidays: Sequence, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set holiday dates.

Parameters:

serial_port – Serial port connected to meter(s)
holidays – A sequence of tuples or lists containing (month, day).
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_weekend_holiday_schedules(serial_port: Serial, weekend_schedule: int, holiday_schedule: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Assign schedule to weekend or holiday.

Parameters:

serial_port – Serial port connected to meter(s)
weekend_schedule – Weekend schedule id (1-6)
holiday_schedule – Holiday schedule id (1-6)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

get_month_tariffs(serial_port: Serial, forward: bool = True, kwh_scale: int = 1) → tuple[source]

Get the tariff values for the last six months.

Parameters:

serial_port – Serial port connected to meter(s)
forward – True for forward current tariffs, False for reverse (kWhReverse).
kwh_scale – The kwh scaling factor. This would be the value of kWh_Scale from a previous data read.
password – Optional meter password

Returns:

A tuple of six tuples containing (total_kwh, kwh_tariff_1, kwh_tariff_2, kwh_tariff_3, kwh_tariff_4)

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_pulse_input_ratio(serial_port: Serial, pulse_input: int, ratio: int, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set pulse input ratio for the specified pulse input.

Parameters:

serial_port – Serial port connected to meter(s)
pulse_input – The pulse input number, 0-based (0-2).
ratio – The new pulse input ratio (0-9999).
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_pulse_output_ratio(serial_port: Serial, ratio: int, ct_ratio: int | None = None, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set pulse output ratio (aka Impulse/kWh constant, CF_Ratio).

See:: VALID_PULSE_OUTPUT_RATIOS

Parameters:

serial_port – Serial port connected to meter(s)
ratio – The new pulse output ratio. The set of valid values for this is based on the current value of CT_Ratio.
ct_ratio – The current value of CT_Ratio. This is optional and just used to check for valid pulse output ratios.
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

reset_resettable_kwh(serial_port: Serial, password: str | SupportsInt | bytes = '00000000') → bool[source]

Reset the resettable kWh.

Parameters:

serial_port – Serial port connected to meter(s)
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_lcd(serial_port: Serial, display_items: Collection, password: str | SupportsInt | bytes = '00000000') → bool[source]

Set LCD display items.

Parameters:

serial_port – Serial port connected to meter(s)
display_items – A list of to 20 display codes.
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

set_display(serial_port: Serial, display_items: Collection, password: str | SupportsInt | bytes = '00000000') → bool

Set LCD display items.

Parameters:

serial_port – Serial port connected to meter(s)
display_items – A list of to 20 display codes.
password – Optional meter password

Returns:

True if the command succeeded, otherwise False.

Raises:

EKMDataError – If the data fails a CRC check, invalid, or missing, or if the serial port read or write fails.

get_meter_io_metrics(serial_port: Serial, address: str | SupportsInt | bytes, version: str | SupportsInt | bytes | None = None, timeout: float = 20, write_timeout: float = 20, inter_byte_timeout: float = 10, number: int = 1, time_rest: float = 0.1) → tuple[source]

Get meter serial port communication metrics.

This performs one or more data read requests and gathers some timing information.

Parameters:

serial_port – Serial port
address – Meter address
version – The meter protocol version. Default is v4.
timeout – Serial port timeout
write_timeout – Serial port timeout
inter_byte_timeout – Serial port timeout
number – Number of times to repeat the meter request
time_rest – Time to sleep between requests

Returns:

A tuple with raw read data and communication metrics.

(
    last data block,
    (avg write time, min write time, max write time),
    (avg read time, min read time, max read time),
    (
        avg time to first byte,
        min time to first byte,
        max time to first byte,
    ),
    (
        avg byte read time,
        min byte read time,
        max byte read time,
    ),
    (
        avg total read/write time,
        min total read/write time,
        max total read/write time,
    )
)

If the number of repeats is > 1 then the metrics will be averages.

Raises:

EKMParamError – If invalid parameter.
EKMDataError – If serial port error or data is invalid or missing.

get_v4_kwh_scale(raw_data: bytes) → int[source]: Extract the kWh_Scale value from raw v4 meter data.

check_password(serial_port: Serial, password: str | SupportsInt | bytes) → bool[source]

Check meter password.

Parameters:

serial_port – Serial port
password – Meter password to check

Raises:

EKMParamError – if the password is invalid.

Start a command session with a meter.

Performs a meter data read which signals the meter to listen for settings commands. This does not unpack the data so it is marginally faster than read_data().

Use read_data() if unpacked data is required.

Parameters:

serial_port – Serial port
address – Meter address
version – The meter protocol version. Default is v4.

Returns:

True if successful, otherwise False.

end_cmd_session(serial_port: Serial) → bool[source]

End a command session with a meter.

After a data read the meter will be in a state where it is listening for commands. This can potentially cause unintended consequences if there is other traffic on the serial port. If no command is going to be sent after a read and there are other types of devices connected to the serial port then it is generally a good idea to send the end session message. Otherwise it doesn’t do much.

This puts it into a state where it will only listen for READ(_A) commands.

Parameters:: serial_port – Serial port
Returns:: Always True. The meter does not send a response to this command.