Skip to content

Domain

Pure domain logic — no I/O, no framework dependencies. These modules are independently testable and reusable.

Schmitt Trigger

gas2mqtt.domain.schmitt

Schmitt trigger for gas meter rotation detection.

A Schmitt trigger is a comparator with hysteresis. It converts the continuous Bz magnetic field value into a clean binary signal by using two thresholds: level + hysteresis (upper) and level - hysteresis (lower). This prevents rapid toggling when Bz is near the threshold.

State transitions

LOW → HIGH: when bz > level + hysteresis (magnet present) HIGH → LOW: when bz < level - hysteresis (magnet absent) Rising edge (LOW → HIGH): counts as one gas meter tick

TriggerState

Bases: Enum

Binary state of the Schmitt trigger.

TriggerEvent dataclass 

TriggerEvent(
    previous: TriggerState,
    current: TriggerState,
    is_rising_edge: bool,
)

Record of a state transition in the Schmitt trigger.

Attributes:

Name Type Description
previous TriggerState

State before the transition.
current TriggerState

State after the transition.
is_rising_edge bool

True when transitioning LOW → HIGH (one gas tick).

SchmittTrigger 

SchmittTrigger(level: int, hysteresis: int)

Schmitt trigger with configurable level and hysteresis.

Parameters:

Name Type Description Default
level int

Centre threshold for the Bz magnetic field value.

 required
hysteresis int

Half-width of the dead band around level.

 required
Source code in packages/src/gas2mqtt/domain/schmitt.py 
50
51
52
53
def __init__(self, level: int, hysteresis: int) -> None:
    self._level = level
    self._hysteresis = hysteresis
    self._state = TriggerState.LOW

state property 

state: TriggerState

Current trigger state.

update 

update(bz: int) -> TriggerEvent | None

Feed a Bz magnetometer value and return any state change.

Returns:

Type Description
TriggerEvent | None

A TriggerEvent when the state transitions, otherwise None.
Source code in packages/src/gas2mqtt/domain/schmitt.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def update(self, bz: int) -> TriggerEvent | None:
    """Feed a Bz magnetometer value and return any state change.

    Returns:
        A ``TriggerEvent`` when the state transitions, otherwise ``None``.
    """
    upper = self._level + self._hysteresis
    lower = self._level - self._hysteresis

    if bz > upper:
        new_state = TriggerState.HIGH
    elif bz < lower:
        new_state = TriggerState.LOW
    else:
        # Inside hysteresis band — no change.
        return None

    if new_state == self._state:
        return None

    previous = self._state
    self._state = new_state
    is_rising = previous is TriggerState.LOW and new_state is TriggerState.HIGH
    return TriggerEvent(
        previous=previous,
        current=new_state,
        is_rising_edge=is_rising,
    )

reset 

reset() -> None

Reset the trigger to the initial LOW state.

Source code in packages/src/gas2mqtt/domain/schmitt.py 
89
90
91
def reset(self) -> None:
    """Reset the trigger to the initial ``LOW`` state."""
    self._state = TriggerState.LOW

Consumption Tracker

gas2mqtt.domain.consumption

Gas consumption tracker.

Tracks cumulative gas consumption in cubic meters. Each gas meter tick represents a configurable amount of gas (default: 10 liters = 0.01 m³).

The consumption value can be: - Incremented by ticks (each tick adds liters_per_tick / 1000 m³) - Set to an absolute value via MQTT command - Reset to zero

This is an optional feature (enabled via settings).

ConsumptionTracker 

ConsumptionTracker(
    liters_per_tick: float, initial_m3: float = 0.0
)

Cumulative gas consumption counter.

Parameters:

Name Type Description Default
liters_per_tick float

Liters of gas per meter tick.

 required
initial_m3 float

Starting consumption in cubic meters.

 0.0
Source code in packages/src/gas2mqtt/domain/consumption.py 
25
26
27
def __init__(self, liters_per_tick: float, initial_m3: float = 0.0) -> None:
    self._liters_per_tick = liters_per_tick
    self._consumption_m3 = initial_m3

consumption_m3 property 

consumption_m3: float

Current cumulative consumption in cubic meters.

tick 

tick() -> float

Record one gas meter tick and return the new consumption.

Source code in packages/src/gas2mqtt/domain/consumption.py 
34
35
36
37
def tick(self) -> float:
    """Record one gas meter tick and return the new consumption."""
    self._consumption_m3 += self._liters_per_tick / 1000.0
    return self._consumption_m3

set_consumption 

set_consumption(m3: float) -> None

Set the consumption to an absolute value (e.g. from MQTT).

Source code in packages/src/gas2mqtt/domain/consumption.py 
39
40
41
def set_consumption(self, m3: float) -> None:
    """Set the consumption to an absolute value (e.g. from MQTT)."""
    self._consumption_m3 = m3

reset 

reset() -> None

Reset consumption to zero.

Source code in packages/src/gas2mqtt/domain/consumption.py 
43
44
45
def reset(self) -> None:
    """Reset consumption to zero."""
    self._consumption_m3 = 0.0