Counter Datatype

Counter Datatype

COUNTER is a composite datatype that models a PLC-style up/down counter with preset, accumulator, done, overflow, underflow, and reset bits. It can be driven manually via increment/decrement, or run autonomously as a background worker.

from ethernetip_emulator.server.device import actions

# Tag definition — pass the preset (count target) as the default
("O_Updates", actions.type.COUNTER(1000))

Expanded Subtags

Each COUNTER tag expands into eight primitive subtags:

Subtag	Type	Description
<name>.PRE	DINT	Preset. the count target. Set at registration time from the default value.
<name>.ACC	DINT	Accumulator. the current count. Starts at 0.
<name>.CU	BOOL	Count-up bit. True while a start worker is running.
<name>.CD	BOOL	Count-down bit.
<name>.DN	BOOL	Done bit. True when ACC >= PRE.
<name>.OV	BOOL	Overflow bit. True briefly when ACC wraps past DINT.MAX.
<name>.UN	BOOL	Underflow bit. True briefly when ACC wraps below DINT.MIN.
<name>.RES	BOOL	Reset bit. Writing True triggers a full reset via on_set_hook.

Manual Counting

increment(name_prefix)

Increments ACC by 1. Sets DN to True if ACC >= PRE. If ACC has already reached DINT.MAX, the overflow bit is set briefly and ACC wraps to 0.

actions.counter.increment("O_Updates")

decrement(name_prefix)

Decrements ACC by 1. Clears DN if ACC drops below PRE. If ACC has already reached DINT.MIN, the underflow bit is set briefly and ACC wraps to 0.

actions.counter.decrement("O_Updates")

reset(name_prefix)

Sets ACC, DN, OV, UN, and RES all back to 0/False.

actions.counter.reset("O_Updates")

Reset also triggers automatically via on_set_hook whenever a PLC client writes True to <name>.RES.

Reading Subtag Values

Method	Returns	Description
get_accumulated_value(name)	int	Current value of ACC.
get_preset_value(name)	int	Current value of PRE.
get_done_bit(name)	bool	Current value of DN.
get_overflow_bit(name)	bool	Current value of OV.
get_underflow_bit(name)	bool	Current value of UN.
get_count_up_bit(name)	bool	Current value of CU.
get_count_down_bit(name)	bool	Current value of CD.
get_reset_bit(name)	bool	Current value of RES.

acc = actions.counter.get_accumulated_value("O_Updates")   # → 42
pre = actions.counter.get_preset_value("O_Updates")        # → 1000
dn  = actions.counter.get_done_bit("O_Updates")            # → False

You can also update the preset at runtime:

actions.counter.set_preset_value("O_Updates", 500)

Reacting to Counter State: is_done and is_reset

is_done and is_reset return a CounterSignal object that can be used both as a boolean check and as a decorator for registering listeners.

As a boolean

if actions.counter.is_done("O_Updates"):
    print("Counter has reached its preset")

As a decorator

When used as a decorator, the callback fires whenever DN (or RES) is written. By default, is_done guards the callback so it only fires on a rising edge (True). Pass rising_edge=False to fire on every write.

@actions.counter.is_done("O_Updates")
def handle_done(attr, key, value) -> None:
    actions.string.set_val("O_Status", "Count Completed")
    actions.counter.reset("O_Updates")

# Fire on both rising and falling edge of DN
@actions.counter.is_done("O_Updates", rising_edge=False)
def handle_dn_any(attr, key, value) -> None:
    print(f"DN changed to {value[0]}")

is_reset works identically but listens on <name>.RES:

@actions.counter.is_reset("O_Updates")
def handle_reset(attr, key, value) -> None:
    print("Counter was reset")

Both is_done and is_reset accept a defer keyword argument, which has the same meaning as in on_change:

@actions.counter.is_done("O_Updates", defer=True)
def handle_done_deferred(attr, key, value) -> None:
    ...

Autonomous Counting: start

start launches a background worker thread that increments the counter at a fixed interval. The worker sets CU to True while it is running and to False when it stops.

Signature

actions.counter.start(
    tag_prefix,
    *,
    period=1.0,
    preset=None,
    initial_delay=1.0,
    enable=None,
)

Parameter	Type	Default	Description
tag_prefix	str	—	The counter tag name.
period	float	1.0	Seconds between each increment.
preset	int | None	None	Override the PRE value at worker start.
initial_delay	float	1.0	Seconds to wait before the first increment.
enable	str | Callable | None	None	Gate the counter. Pass a BOOL tag name or a zero-argument callable. The worker skips each tick while the gate is False.

# Count every 500ms, stop after reaching 200
actions.counter.start("O_Cycles", period=0.5, preset=200)

Worker behaviour

Each tick the worker:

1. Checks the enable gate (if set) and skips the tick if the gate is closed.
2. Checks for DINT.MAX overflow — if hit, briefly sets OV and wraps ACC to 0.
3. Increments ACC by 1.
4. If ACC >= PRE (and PRE > 0), sets DN to True, sleeps for one period, then resets ACC to 0 and clears DN.

Gating with a BOOL tag

# Only count while O_Running is True
actions.counter.start("O_Cycles", period=1.0, enable="O_Running")

Gating with a callable

import datetime

def is_work_hours() -> bool:
    return 8 <= datetime.datetime.now().hour < 18

actions.counter.start("O_WorkCycles", period=60.0, enable=is_work_hours)

Full Example

# tags.py
@tag_registry.register
def _():
    return [
        ("O_Parts",   actions.type.COUNTER(100)),
        ("O_Running", actions.type.BOOL(False)),
    ]


# actions.py
@actions.bool.on_change("O_Running")
def handle_running(attr, key, value) -> None:
    if value[0]:
        actions.counter.start("O_Parts", period=0.5, enable="O_Running")


@actions.counter.is_done("O_Parts")
def handle_batch_done(attr, key, value) -> None:
    actions.bool.set_val("O_Running", False)
    actions.string.set_val("O_Status", "Batch Complete")
    actions.counter.reset("O_Parts")

Method Reference

Method	Description
increment(name)	Increment ACC by 1; set DN if preset reached
decrement(name)	Decrement ACC by 1; clear DN if below preset
reset(name)	Reset ACC, DN, OV, UN, and RES to zero/false
start(name, *, period, preset, initial_delay, enable)	Start a background auto-increment worker
is_done(name, *, defer, rising_edge)	Boolean check or decorator for DN changes
is_reset(name, *, defer, rising_edge)	Boolean check or decorator for RES changes
get_accumulated_value(name)	Read ACC
get_preset_value(name)	Read PRE
set_preset_value(name, value)	Write PRE
get_done_bit(name)	Read DN
get_overflow_bit(name)	Read OV
get_underflow_bit(name)	Read UN
get_count_up_bit(name)	Read CU
get_count_down_bit(name)	Read CD
get_reset_bit(name)	Read RES