2.12.4 - Add integer clamping for scaled values

.github/workflows/python-package.yml

@@ -67,7 +67,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=runtimepy version=2.12.3
+                            repo=runtimepy version=2.12.4
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.3
-    hash=f8a5c23e6166199774f2dc5e19dc76d0
+    hash=2361470f35dc837ce96ac147be7e44ee
     =====================================
 -->
 
-# runtimepy ([2.12.3](https://pypi.org/project/runtimepy/))
+# runtimepy ([2.12.4](https://pypi.org/project/runtimepy/))
 
 [![python](https://img.shields.io/pypi/pyversions/runtimepy.svg)](https://pypi.org/project/runtimepy/)
 ![Build Status](https://github.com/vkottler/runtimepy/workflows/Python%20Package/badge.svg)

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 2
 minor: 12
-patch: 3
+patch: 4
 entry: runtimepy

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "2.12.3"
+version = "2.12.4"
 description = "A framework for implementing Python services."
 readme = "README.md"
 requires-python = ">=3.11"

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.3
-# hash=b58f93345cd96f4613356453688c5a3d
+# hash=8df5cf822b83a463b54443ea84e6e40e
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "2.12.3"
+VERSION = "2.12.4"
 
 # runtimepy-specific content.
 METRICS_NAME = "metrics"

runtimepy/primitives/base.py

@@ -131,10 +131,15 @@ def scaled(self) -> Numeric:
     def scaled(self, value: T) -> None:
         """Set this value but invert scaling information."""
 
-        self.value = invert(  # type: ignore
+        val = invert(
             value, scaling=self.scaling, should_round=self.kind.is_integer
         )
 
+        if self.kind.int_bounds is not None:
+            val = self.kind.int_bounds.clamp(val)  # type: ignore
+
+        self.value = val  # type: ignore
+
     def __call__(self, value: T = None) -> T:
         """
         A callable interface for setting and getting the underlying value.

runtimepy/primitives/type/base.py

@@ -9,6 +9,7 @@
 from struct import unpack as _unpack
 from typing import BinaryIO as _BinaryIO
 from typing import Generic as _Generic
+from typing import Optional as _Optional
 from typing import Type as _Type
 from typing import TypeVar as _TypeVar
 from typing import Union as _Union
@@ -19,6 +20,7 @@
     DEFAULT_BYTE_ORDER as _DEFAULT_BYTE_ORDER,
 )
 from runtimepy.primitives.byte_order import ByteOrder as _ByteOrder
+from runtimepy.primitives.type.bounds import IntegerBounds
 
 # Integer type aliases.
 Int8Ctype = _ctypes.c_byte
@@ -73,6 +75,7 @@ def __init__(self, struct_format: str, signed: bool = True) -> None:
 
         self.format = struct_format
         self.signed = signed
+        self.int_bounds: _Optional[IntegerBounds] = None
 
         # Make sure that the struct size and ctype size match. There's
         # unfortunately no obvious (or via public interfaces) way to just

runtimepy/primitives/type/bounds.py

@@ -0,0 +1,30 @@
+"""
+A module implementing an interface for keeping track of primitive-integer
+bounds (based on bit width).
+"""
+
+# built-in
+from typing import NamedTuple
+
+
+class IntegerBounds(NamedTuple):
+    """A container for integer bounds."""
+
+    min: int
+    max: int
+
+    def clamp(self, val: int) -> int:
+        """
+        Ensure that 'val' is between min and max, use the min or max value
+        instead of the provided value if it exceeds these bounds.
+        """
+
+        return max(self.min, min(val, self.max))
+
+    @staticmethod
+    def create(byte_count: int, signed: bool) -> "IntegerBounds":
+        """Compute maximum and minimum values given size and signedness."""
+
+        min_val = 0 if not signed else -1 * (2 ** (byte_count * 8 - 1))
+        width = 8 * byte_count if not signed else 8 * byte_count - 1
+        return IntegerBounds(min_val, (2**width) - 1)

runtimepy/primitives/type/int.py

@@ -12,6 +12,7 @@
 from runtimepy.primitives.type.base import Uint16Ctype as _Uint16Ctype
 from runtimepy.primitives.type.base import Uint32Ctype as _Uint32Ctype
 from runtimepy.primitives.type.base import Uint64Ctype as _Uint64Ctype
+from runtimepy.primitives.type.bounds import IntegerBounds
 
 
 class Int8Type(_PrimitiveType[_Int8Ctype]):
@@ -24,6 +25,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("b")
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(1, True)
 
 
 Int8 = Int8Type()
@@ -39,6 +41,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("h")
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(2, True)
 
 
 Int16 = Int16Type()
@@ -54,6 +57,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("i")
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(4, True)
 
 
 Int32 = Int32Type()
@@ -69,6 +73,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("q")
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(8, True)
 
 
 Int64 = Int64Type()
@@ -84,6 +89,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("B", signed=False)
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(1, False)
 
 
 Uint8 = Uint8Type()
@@ -99,6 +105,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("H", signed=False)
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(2, False)
 
 
 Uint16 = Uint16Type()
@@ -114,6 +121,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("I", signed=False)
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(4, False)
 
 
 Uint32 = Uint32Type()
@@ -129,6 +137,7 @@ def __init__(self) -> None:
         """Initialize this type."""
         super().__init__("Q", signed=False)
         assert self.is_integer
+        self.int_bounds = IntegerBounds.create(8, False)
 
 
 Uint64 = Uint64Type()

tests/channel/environment/test_command.py

@@ -30,7 +30,12 @@ def test_channel_command_scalings():
 
     env = ChannelEnvironment()
 
-    env.int_channel("4_20ma", commandable=True, scaling=[0.0, 3.81469727e-07])
+    env.int_channel(
+        "4_20ma",
+        commandable=True,
+        scaling=[0.0, 3.81469727e-07],
+        kind="uint16",
+    )
 
     processor = ChannelCommandProcessor(env, getLogger(__name__))
 
@@ -39,6 +44,10 @@ def test_channel_command_scalings():
         assert processor.command(f"set 4_20ma {val}")
         val += 0.01
 
+    # Ensure that clamping works.
+    assert processor.command("set 4_20ma 0.030")
+    assert isclose(env.value("4_20ma"), 0.025, rel_tol=0.001)  # type: ignore
+
 
 def test_channel_command_processor_basic():
     """Test basic interactions with the channel-command processor."""