Decoding Scaled Integer Registers for Flow and pH

Not every field instrument speaks in IEEE 754 floats. A large share of the flow meters and pH analyzers on a water utility’s control network report their measurements as plain integers — a raw count that only becomes a real engineering value once you multiply by a device-specific scale and add an offset. This page is the integer-focused companion inside the parent Modbus TCP Parsing Workflows section, and it deliberately covers different ground than its sibling on parsing Modbus registers for turbidity sensors, which decodes 32-bit floats. Here the whole game is the scaling relationship $y = m \cdot \text{raw} + b$, where a signed or unsigned integer raw is converted into liters per second, million gallons per day, or pH units. Get the signedness, the register width, or the multiplier wrong and you will silently publish a plausible-but-false reading into the wider SCADA Data Ingestion & Time-Series Sync architecture — no protocol error, no exception, just a bad number feeding compliance math.

Prerequisites & Environment Setup

The code below targets Python 3.10+ and the asynchronous pymodbus client, which is the practical baseline for polling PLCs and RTUs at the cadence a compliance historian expects. Pin the library version: the BinaryPayloadDecoder used here ships on pymodbus 3.x and is removed on 4.x, where the same integers are decoded through client.convert_from_registers(...). Optional pydantic typing pays for itself the moment these records leave the parser and enter validation.

You also need two artifacts that never come from a package index. The first is the instrument’s communication register map — the manufacturer document that states, per point, the starting address, the register width, whether the value is signed, and the scale and offset to apply. The second is a read-only network path to the device, provisioned through the egress described in Security Boundary Design; an ingestion parser is never a writable Modbus master. Without the register map, every integer you read is an unlabeled count with no units.

python3 -m venv .venv && source .venv/bin/activate
pip install "pymodbus[serial]==3.6.9" "pydantic==2.7.*"
# Optional: pandas for downstream historian batching
pip install "pandas==2.2.*"

Step-by-Step Implementation

Step 1 — Classify each point by width and signedness

An integer register carries no self-describing type. The exact same 16 bits mean two completely different things depending on whether the map declares the point signed or unsigned, and that choice is not cosmetic. A magmeter that supports reverse flow encodes direction in the sign bit: a raw register of 0xFFFF is -1 when read as a signed int16 (a small reverse flow after scaling) but 65535 when read as an unsigned uint16 (a nonsense forward spike). pH is always physically non-negative, so its measurement register is typically unsigned, but a temperature-compensation or millivolt-offset register alongside it is frequently signed. The diagram below shows the fork that a single mislabeled point creates.

Step 2 — Decode 16-bit and 32-bit integers by declared type

A single ingestion service typically polls flow, pH, and their companion registers together, so the decode step must dispatch on a declared data type rather than hard-coding one width. Flow rate and pH usually fit a 16-bit register, while a resettable flow totalizer needs a 32-bit accumulator spread across two consecutive registers. The helper below decodes all four integer shapes with BinaryPayloadDecoder, keeping byte order fixed to network (big-endian) order while allowing the word order of 32-bit values to follow the device profile, since Modicon-lineage PLCs frequently swap the two 16-bit words.

import logging
from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

logger = logging.getLogger("scaled_integer_parser")

# Register width in 16-bit words, keyed by declared point type.
DATATYPE_WORDS = {"int16": 1, "uint16": 1, "int32": 2, "uint32": 2}


def decode_integer(registers, data_type: str, word_order=Endian.BIG) -> int:
    """Decode raw holding registers into a Python int by declared type.

    Byte order stays big-endian (Modbus network order); word_order lets
    a 32-bit point follow a word-swapped device profile.
    """
    decoder = BinaryPayloadDecoder.fromRegisters(
        registers, byteorder=Endian.BIG, wordorder=word_order
    )
    if data_type == "int16":
        return decoder.decode_16bit_int()
    if data_type == "uint16":
        return decoder.decode_16bit_uint()
    if data_type == "int32":
        return decoder.decode_32bit_int()
    if data_type == "uint32":
        return decoder.decode_32bit_uint()
    raise ValueError(f"Unsupported data_type: {data_type}. Use {list(DATATYPE_WORDS)}")

The word count that read_holding_registers must request comes straight from DATATYPE_WORDS, which prevents the common mistake of reading one register for a 32-bit totalizer and truncating the high word.

Step 3 — Apply engineering-unit scaling and range validation

Decoding yields a raw count; the reading is only meaningful after the linear transform $y = m \cdot \text{raw} + b$. For a magmeter reporting hundredths of a liter per second, $m = 0.01$ and $b = 0$; for a pH analyzer that transmits the value multiplied by one hundred, $m = 0.01$ so a raw 725 becomes 7.25 pH; an instrument that reports a biased 4–20 mA equivalent may carry a non-zero offset $b$. Scaling is where an out-of-range count must be caught, because a stuck or saturated register often produces a number that is arithmetically valid but physically impossible — a pH of 14.7 or a flow larger than the pipe can carry. The function below reads, decodes, scales, bounds-checks, and emits a record with an explicit quality flag, mirroring the flag vocabulary the sibling turbidity parser uses so both feed a uniform contract.

from datetime import datetime, timezone
from typing import Any, Dict, Optional
from pymodbus.client import AsyncModbusTcpClient


async def parse_scaled_point(
    client: AsyncModbusTcpClient,
    address: int,
    unit_id: int,
    data_type: str,
    scale: float,
    offset: float,
    lo: float,
    hi: float,
    word_order=Endian.BIG,
) -> Dict[str, Any]:
    """Read one scaled-integer point and return a compliance-ready record."""
    now = datetime.now(timezone.utc)
    count = DATATYPE_WORDS[data_type]
    try:
        response = await client.read_holding_registers(
            address=address, count=count, slave=unit_id
        )
        if response.isError():
            logger.error("Modbus read error at %s: %s", address, response)
            return {"value": None, "quality": "BAD", "timestamp": now}

        raw = decode_integer(response.registers, data_type, word_order)
        engineering = (raw * scale) + offset

        if not (lo <= engineering <= hi):
            logger.warning("Out-of-range value %.4f at %s", engineering, address)
            return {"value": engineering, "quality": "SUSPECT", "timestamp": now}

        return {
            "value": round(engineering, 4),
            "raw": raw,
            "quality": "GOOD",
            "timestamp": now,
        }
    except Exception as exc:  # noqa: BLE001 - boundary logging
        logger.exception("Unhandled parse error at %s: %s", address, exc)
        return {"value": None, "quality": "BAD", "timestamp": now}

Keeping the raw count alongside the scaled value in the record is deliberate: when an auditor or a field technician disputes a reading, the untransformed integer plus the versioned scale is the evidence that reproduces it exactly. That raw-plus-transform provenance is what makes the value defensible when it reaches the Historian Integration Patterns for Compliance Pipelines layer.

Step 4 — Emit a typed record and hand off to alignment

Once each point is decoded and flagged, the parser should not pass around loose dictionaries. A small pydantic model rejects malformed records at the boundary and gives every downstream consumer a stable schema. Because the same poll may return a pH at one instant and a flow totalizer at another, the model carries the point identity and units so the time-series alignment strategies module can reconcile them onto a common axis before any averaging window is computed.

from pydantic import BaseModel, field_validator


class ScaledReading(BaseModel):
    point_id: str
    units: str
    value: Optional[float]
    raw: Optional[int] = None
    quality: str
    timestamp: datetime

    @field_validator("quality")
    @classmethod
    def known_flag(cls, v: str) -> str:
        if v not in {"GOOD", "SUSPECT", "BAD"}:
            raise ValueError(f"unknown quality flag: {v}")
        return v

Only GOOD readings should ever contribute to a running average; SUSPECT and BAD are retained for audit but excluded from compliance statistics, exactly as the MCL Exceedance Logic Implementation expects when it evaluates thresholds against a clean series.

Configuration Reference

Treat every value in these tables as versioned configuration pulled from the device communication map, never as an inline constant. The first table is a representative register block for a magmeter with pH on the same RTU; the second gives the scaling and bounds for each point.

Register map (representative flow + pH RTU)

Scaling and validation parameters

Quality-flag codes

Verification & Testing

The decisive test is the signed/unsigned boundary, because that is where a wrong declaration produces a silent, plausible number rather than a crash. Feed known register words through decode_integer for both interpretations and assert the two’s-complement result, then confirm the scaled value lands where the map says it should.

from pymodbus.constants import Endian


def test_signed_vs_unsigned_diverge():
    # 0xFFFF is -1 signed, 65535 unsigned.
    assert decode_integer([0xFFFF], "int16") == -1
    assert decode_integer([0xFFFF], "uint16") == 65535


def test_ph_scale_matches_map():
    # Raw 725 at m=0.01, b=0 -> pH 7.25.
    raw = decode_integer([0x02D5], "uint16")  # 0x02D5 == 725
    assert round(raw * 0.01, 2) == 7.25


def test_word_swapped_totalizer():
    # A uint32 of 100000 stored word-swapped decodes back correctly.
    hi, lo = 0x0001, 0x86A0  # 0x000186A0 == 100000
    assert decode_integer([lo, hi], "uint32", Endian.LITTLE) == 100000

Acceptance criteria before promoting the parser to production:

• Every point’s signedness, width, scale, offset, and bounds are pinned from the device communication map.
• A 0xFFFF register decodes to -1 under int16 and 65535 under uint16 in a committed test.
• 32-bit totalizers request two registers and decode identically under the device’s declared word order.
• A known raw count reproduces the documented engineering value through $y = m\cdot\text{raw}+b$.
• Out-of-range counts resolve to SUSPECT and are excluded from averages, never clamped.
• Modbus errors and short reads resolve to BAD with a None value, never a numeric one.
• Each emitted record carries the raw count, scaled value, units, quality flag, and a UTC timestamp.

Troubleshooting & Gotchas

• Flow flips to an enormous positive number during reverse flow. The point is declared unsigned but the meter is signed; -1 is being read as 65535. Re-declare the point as int16 and re-run the signed/unsigned test above before trusting any dashboard.
• A totalizer resets or jumps by exactly 65,536. The 32-bit accumulator is being read one register at a time, so only the low word is captured. Request two registers (count=2) and decode as int32/uint32; confirm the word order, since a Modicon-lineage PLC will need Endian.LITTLE word order.
• pH sits at a constant, physically impossible value like 655.35. A saturated or stuck raw register (0xFFFF at m=0.01) passed straight through scaling. The lo–hi guard should already catch it as SUSPECT; if it does not, your bounds are wider than the instrument’s real range.
• ImportError: BinaryPayloadDecoder on pymodbus 4.x. The helper is gone in 4.x. Decode with client.convert_from_registers(response.registers, data_type=client.DATATYPE.INT16) (or UINT16/INT32/UINT32), or pin pymodbus==3.6.9 until you migrate.
• Every read returns a Modbus error. Almost always a wrong unit_id/slave or a 1-based map address handed to the 0-based client. Subtract the documented 40001 offset when translating addresses, and confirm the RTU’s slave ID.

Related

• Modbus TCP Parsing Workflows — parent section and the shared parsing contract these records satisfy
• Parsing Modbus Registers for Turbidity Sensors — the float-decoding counterpart to this integer workflow
• Time-Series Alignment Strategies — reconciling the timestamps this parser emits onto one axis
• Historian Integration Patterns for Compliance Pipelines — where raw-plus-scale provenance is persisted
• MCL Exceedance Logic Implementation — evaluates thresholds against the clean series this parser produces