data-collector-gateway/app/services/modbus_codec.py

from __future__ import annotations

import struct
from typing import Any


def register_quantity_for_type(point_type: str) -> int:
    if point_type in {"bool", "int16", "uint16"}:
        return 1
    if point_type in {"int64", "uint64", "float64"}:
        return 4
    return 2


def ordered_bytes(registers: list[int], word_byte_order: str) -> bytes:
    raw = b"".join(register.to_bytes(2, byteorder="big", signed=False) for register in registers)
    words = [raw[index : index + 2] for index in range(0, len(raw), 2)]
    if word_byte_order in {"BADC", "DCBA"}:
        words = [word[::-1] for word in words]
    if word_byte_order in {"CDAB", "DCBA"}:
        words = list(reversed(words))
    return b"".join(words)


def convert_register_value(registers: list[int], point_type: str, word_byte_order: str, bit: int | None) -> Any:
    if point_type == "bool":
        value = registers[0]
        if bit is None:
            return value != 0
        return ((value >> bit) & 1) == 1

    payload = ordered_bytes(registers, word_byte_order)

    if point_type == "int16":
        return int.from_bytes(payload, byteorder="big", signed=True)
    if point_type == "uint16":
        return int.from_bytes(payload, byteorder="big", signed=False)
    if point_type == "int32":
        return int.from_bytes(payload, byteorder="big", signed=True)
    if point_type == "uint32":
        return int.from_bytes(payload, byteorder="big", signed=False)
    if point_type == "int64":
        return int.from_bytes(payload, byteorder="big", signed=True)
    if point_type == "uint64":
        return int.from_bytes(payload, byteorder="big", signed=False)
    if point_type == "float32":
        return struct.unpack(">f", payload)[0]
    if point_type == "float64":
        return struct.unpack(">d", payload)[0]

    raise ValueError(f"unsupported point type: {point_type}")