feat: implement parsing of floats

soti/pyproject.toml

@@ -8,6 +8,7 @@ dependencies = [
     "pyserial>=3.5",
     "serial>=0.0.97",
     "urwid>=3.0.3",
+    "numpy>=2.1.1"
 ]
 
 [build-system]

soti/umsats_soti/parser/parser.py

@@ -6,67 +6,17 @@
 
 from .message import Message, MAX_BODY_SIZE
 from .constants import NodeID, CmdID
+from .value import Value
 
 
-DATA_ARG_RE = re.compile(r'(?:\((\w+)\))?(\S+)')
+# Regular Expressions
+DATA_ARG_RE = re.compile(r'(?:(\w+)\:)?(\S+)')
 
 
 class ArgumentException(Exception):
     pass
 
 
-def parse_int(s: str) -> int:
-    """
-    Casts numbers and enum members to int.
-    Raises ValueError for invalid values.
-    """
-    is_negative = False
-    if s[0] == '-':
-        is_negative = True
-        s = s[1:]
-
-    i: int = 0
-    if s.isnumeric():
-        i = int(s)
-    elif s[:2] == "0x":
-        i = int(s, 16)
-    elif s[:2] == "0b":
-        i = int(s, 2)
-    elif s in NodeID.__members__:
-        i = NodeID[s].value
-    elif s in CmdID.__members__:
-        i = CmdID[s].value
-    else:
-        raise ValueError(f"'{s}' cannot be parsed as an int")
-
-    if is_negative:
-        i = -i
-
-    return i
-
-
-def get_implied_type(value: int) -> str:
-    """Uses the range of the integer to infer an appropriate type to store the value."""
-    if value < 0:
-        if -(2**7) <= value < 2**7:
-            return "i8"
-        elif -(2**15) <= value < 2**15:
-            return "i16"
-        elif -(2**31) <= value < 2**31:
-            return "i32"
-        else:
-            raise ValueError(f"{value} is out of range for signed 32-bit integer.")
-    else:
-        if 0 <= value < 2**8:
-            return "u8"
-        elif 2**8 <= value < 2**16:
-            return "u16"
-        elif 2**16 <= value < 2**32:
-            return "u32"
-        else:
-            raise ValueError(f"{value} is out of range for unsigned 32-bit integer.")
-
-
 def get_implied_recipient(cmd: CmdID) -> NodeID | None:
     cmd_name_prefix = cmd.name.split('_')[0]
 
@@ -98,9 +48,9 @@ def parse_send(args: str, default_sender: NodeID) -> Message:
     if(len(parts) == 0):
         raise ValueError("No arguments provided")
     try:
-        cmd_id = CmdID(parse_int(parts[0]))
+        cmd_id = CmdID(Value.from_str(parts[0], 'u8').value)
     except ValueError as e:
-        raise ArgumentException(f"Invalid command ID '{parts[0]}'") from e
+        raise ArgumentException(f"Invalid command ID '{parts[0]}': {e}") from e
 
     # Assign default values for the command options.
     priority: int = 255
@@ -118,64 +68,39 @@ def parse_send(args: str, default_sender: NodeID) -> Message:
             if '=' in arg:
                 key, value = arg.split('=')
                 if key == "priority":
-                    priority = parse_int(value)
+                    priority = Value.from_str(value, 'u8').value
                     if not 0 <= priority <= 32:
                         raise ArgumentException(f"Invalid priority '{key}'. Expected range is [0, 32]")
                 elif key == "from":
                     try:
-                        sender_id = NodeID(parse_int(value))
+                        sender_id = NodeID(Value.from_str(value, 'u8').value)
                     except ValueError as exc:
                         raise ArgumentException(f"Invalid node ID '{value}'") from exc
                 elif key == "to":
                     try:
-                        recipient_id = NodeID(parse_int(value))
+                        recipient_id = NodeID(Value.from_str(value, 'u8').value)
                     except ValueError as exc:
                         raise ArgumentException(f"Invalid node ID '{value}'") from exc
                 elif key == "ack":
-                    if value.lower() == "true":
-                        is_ack = True
-                    elif value.lower() == "false":
-                        is_ack = False
-                    else:
-                        raise ArgumentException(f"Invalid value for ack '{value}'. Expected true or false")
+                    try:
+                        is_ack = bool(Value.from_str(value, 'u8').value)
+                    except ValueError as exc:
+                        raise ArgumentException(f"Invalid value for ack '{value}'. Expected true or false") from exc
                 else:
                     raise ArgumentException(f"Unknown option '{key}'")
 
             # treat as data argument
             elif data_index < MAX_BODY_SIZE:
-                data_type = None
-                data_size = 0
-                is_signed = False
-
                 re_match = DATA_ARG_RE.match(arg)
                 if not re_match:
                     raise ArgumentException(f"Invalid syntax for data argument '{arg}'")
 
                 data_type = re_match.group(1)
-                value = parse_int(re_match.group(2))
-
                 if not data_type:
-                    data_type = get_implied_type(value)
-
-                if data_type in ["u8", "u16", "u32"]:
-                    is_signed = False
-                elif data_type in ["i8", "i16", "i32"]:
-                    is_signed = True
-
-                if data_type in ["u8", "i8"]:
-                    data_size = 1
-                elif data_type in ["u16", "i16"]:
-                    data_size = 2
-                elif data_type in ["u32", "i32"]:
-                    data_size = 4
-                else:
-                    raise ArgumentException(f"Invalid type '{data_type}'")
+                    raise ArgumentException(f"You must specify a data type for argument '{arg}'")
 
-                # Convert value to a bytes object.
-                try:
-                    raw_bytes = value.to_bytes(data_size, byteorder="little", signed=is_signed)
-                except OverflowError as exc:
-                    raise ArgumentException(f"Integer overflow. '{re_match.group(2)}' cannot be represented as {data_type}") from exc
+                value = Value.from_str(re_match.group(2), data_type)
+                raw_bytes = value.to_bytes()
 
                 # Add the bytes of data.
                 data.extend(raw_bytes)

soti/umsats_soti/parser/value.py

@@ -0,0 +1,189 @@
+import re
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+import numpy as np
+
+from .constants import NodeID, CmdID
+
+
+DECIMAL_NUMBER_RE = re.compile(r'^-?[0-9\.]+$')
+
+
+class ValueParser(ABC):
+    @abstractmethod
+    def validate(self, s: str) -> bool:
+        """Checks if the parser can parse the provided string"""
+
+    @abstractmethod
+    def parse(self, s: str) -> any:
+        """Parses the string into a value"""
+
+
+class FloatParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        return DECIMAL_NUMBER_RE.match(s) is not None
+
+    def parse(self, s: str) -> any:
+        return float(s)
+
+
+class IntParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        return s.isnumeric()
+
+    def parse(self, s: str) -> any:
+        return int(s)
+
+
+class HexParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        if s[0] == '-':
+            s = s[1:]
+        return s[:2] == "0x"
+
+    def parse(self, s: str) -> any:
+        if s[0] == '-':
+            return -int(s, 16)
+        else:
+            return int(s, 16)
+
+
+class BinaryParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        if s[0] == '-':
+            s = s[1:]
+        return s[:2] == "0b"
+
+    def parse(self, s: str) -> any:
+        if s[0] == '-':
+            return -int(s, 2)
+        else:
+            return int(s, 2)
+
+
+class NodeIDParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        return s in NodeID.__members__
+
+    def parse(self, s: str) -> any:
+        return NodeID[s]
+
+
+class CmdIDParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        return s in CmdID.__members__
+
+    def parse(self, s: str) -> any:
+        return CmdID[s]
+
+
+class BooleanParser(ValueParser):
+    def validate(self, s: str) -> bool:
+        return s.lower() in ['true', 'false', '0', '1']
+
+    def parse(self, s: str) -> any:
+        if s.lower() in ['true', '1']:
+            return True
+        elif s.lower() in ['false', '0']:
+            return False
+        else:
+            raise ValueError(f"Value '{s}' cannot be converted to boolean")
+
+
+_all_parsers = [
+    IntParser(),
+    HexParser(),
+    BinaryParser(),
+    NodeIDParser(),
+    CmdIDParser(),
+    BooleanParser(),
+    FloatParser()
+]
+
+_data_type_map = {
+    "u8": np.uint8,
+    "u16": np.uint16,
+    "u32": np.uint32,
+    "i8": np.int8,
+    "i16": np.int16,
+    "i32": np.int32,
+    "f32": np.float32
+}
+
+_data_validators = {
+    "u8": lambda v: is_integer_like(v) and 0 <= get_integer_value(v) < 2**8,
+    "u16": lambda v: is_integer_like(v) and 0 <= get_integer_value(v) < 2**16,
+    "u32": lambda v: is_integer_like(v) and 0 <= get_integer_value(v) < 2**32,
+    "i8": lambda v: is_integer_like(v) and -(2**7) <= get_integer_value(v) < 2**7,
+    "i16": lambda v: is_integer_like(v) and -(2**15) <= get_integer_value(v) < 2**15,
+    "i32": lambda v: is_integer_like(v) and -(2**31) <= get_integer_value(v) < 2**31,
+    "f32": lambda v: isinstance(v, float)
+}
+
+def is_integer_like(v: any) -> bool:
+    """True if the value is integer-like"""
+    return isinstance(v, (int, bool, CmdID, NodeID))
+
+def get_integer_value(v: any) -> int:
+    """Returns the value of an integer-like value"""
+    if isinstance(v, int):
+        return v
+    elif isinstance(v, bool):
+        return 1 if v else 0
+    elif isinstance(v, (CmdID, NodeID)):
+        return v.value
+    else:
+        raise ValueError(f"Value '{v}' has no integer value")
+
+
+@dataclass
+class Value():
+    value: any
+    data_type: str
+
+    def to_bytes(self) -> bytes:
+        try:
+            dtype = _data_type_map[self.data_type]
+        except KeyError:
+            raise ValueError(f"Unkown data type '{self.data_type}'")
+
+        raw_bytes = np.dtype(dtype).newbyteorder("<").type(self.value).tobytes()
+
+        return raw_bytes
+
+    @staticmethod
+    def from_str(s: str, data_type: str | None = None) -> 'Value':
+        """
+        Extracts the value from the provided string.
+        Raises ValueError for invalid values.
+        """
+        # Remove whitespace
+        s = s.strip()
+
+        try:
+            value = Value._parse(s, _all_parsers)
+        except Exception as e:
+            raise ValueError(f"Failed to parse value '{s}': {e}")
+
+        if data_type is not None:
+            Value._validate_data(value, data_type)
+
+        return Value(value, data_type)
+
+    @staticmethod
+    def _parse(s: str, parsers: list[ValueParser]) -> any:
+        for p in parsers:
+            if p.validate(s):
+                return p.parse(s)
+        raise ValueError(f"Could not find matching parser for value '{s}'")
+
+    @staticmethod
+    def _validate_data(value: any, data_type: str) -> bool:
+        """Determines if the value object is valid"""
+        try:
+            validator = _data_validators[data_type]
+        except KeyError:
+            raise ValueError(f"Unknown data type 2 '{data_type}'")
+
+        return validator(value)