Merge pull request #602 from gijzelaerr/feat/missing-data-type-setters

Add missing data type setters and fix get_lword/get_wchar bugs

Author: gijzelaerr

snap7/util/__init__.py

@@ -2,6 +2,7 @@
     set_bool,
     set_fstring,
     set_string,
+    set_wstring,
     set_real,
     set_dword,
     set_udint,
@@ -11,11 +12,16 @@
     set_word,
     set_byte,
     set_char,
+    set_wchar,
     set_usint,
     set_sint,
     set_time,
     set_lreal,
+    set_lword,
     set_date,
+    set_tod,
+    set_dtl,
+    set_dt,
 )
 
 from .getters import (
@@ -39,6 +45,10 @@
     get_date,
     get_tod,
     get_lreal,
+    get_lword,
+    get_ltime,
+    get_ltod,
+    get_ldt,
     get_char,
     get_wchar,
     get_dtl,
@@ -60,6 +70,10 @@
     "get_date",
     "get_tod",
     "get_lreal",
+    "get_lword",
+    "get_ltime",
+    "get_ltod",
+    "get_ldt",
     "get_char",
     "get_wchar",
     "get_dtl",
@@ -72,17 +86,23 @@
     "set_dword",
     "set_date",
     "set_lreal",
+    "set_lword",
     "set_udint",
     "set_dint",
     "set_uint",
     "set_int",
     "set_word",
     "set_byte",
     "set_char",
+    "set_wchar",
     "set_usint",
     "set_sint",
     "set_time",
     "set_bool",
     "set_fstring",
     "set_string",
+    "set_wstring",
+    "set_tod",
+    "set_dtl",
+    "set_dt",
 ]

snap7/util/db.py

@@ -86,7 +86,7 @@
 
 import re
 from logging import getLogger
-from datetime import datetime, date
+from datetime import datetime, date, timedelta
 from typing import Any, Optional, Union, Iterator, Tuple, Dict, Callable
 
 from snap7 import Client
@@ -96,6 +96,7 @@
     set_bool,
     set_fstring,
     set_string,
+    set_wstring,
     set_real,
     set_dword,
     set_udint,
@@ -105,11 +106,15 @@
     set_word,
     set_byte,
     set_char,
+    set_wchar,
     set_usint,
     set_sint,
     set_time,
     set_lreal,
     set_date,
+    set_tod,
+    set_dtl,
+    set_dt,
     get_bool,
     get_fstring,
     get_string,
@@ -672,6 +677,14 @@ def set_value(self, byte_index: Union[str, int], type_: str, value: Union[bool,
             set_string(bytearray_, byte_index, value, max_size_int)
             return None
 
+        if type_.startswith("WSTRING") and isinstance(value, str):
+            max_size = re.search(r"\d+", type_)
+            if max_size is None:
+                raise ValueError("Max size could not be determinate. re.search() returned None")
+            max_size_int = int(max_size.group(0))
+            set_wstring(bytearray_, byte_index, value, max_size_int)
+            return None
+
         if type_ == "REAL":
             return set_real(bytearray_, byte_index, value)
 
@@ -681,6 +694,9 @@ def set_value(self, byte_index: Union[str, int], type_: str, value: Union[bool,
         if type_ == "CHAR" and isinstance(value, str):
             return set_char(bytearray_, byte_index, value)
 
+        if type_ == "WCHAR" and isinstance(value, str):
+            return set_wchar(bytearray_, byte_index, value)
+
         if isinstance(value, int):
             type_to_func = {
                 "DWORD": set_dword,
@@ -702,6 +718,15 @@ def set_value(self, byte_index: Union[str, int], type_: str, value: Union[bool,
         if type_ == "DATE" and isinstance(value, date):
             return set_date(bytearray_, byte_index, value)
 
+        if type_ in ("TIME_OF_DAY", "TOD") and isinstance(value, timedelta):
+            return set_tod(bytearray_, byte_index, value)
+
+        if type_ == "DTL" and isinstance(value, datetime):
+            return set_dtl(bytearray_, byte_index, value)
+
+        if type_ == "DATE_AND_TIME" and isinstance(value, datetime):
+            return set_dt(bytearray_, byte_index, value)
+
         raise ValueError
 
     def write(self, client: Client) -> None:

snap7/util/getters.py

@@ -515,15 +515,13 @@ def get_lreal(bytearray_: bytearray, byte_index: int) -> float:
     return float(struct.unpack_from(">d", bytearray_, offset=byte_index)[0])
 
 
-def get_lword(bytearray_: bytearray, byte_index: int) -> bytearray:
+def get_lword(bytearray_: bytearray, byte_index: int) -> int:
     """Get the long word
 
-    THIS VALUE IS NEITHER TESTED NOR VERIFIED BY A REAL PLC AT THE MOMENT
-
     Notes:
         Datatype `lword` (long word) consists in 8 bytes in the PLC.
-        Maximum value posible is bytearray(b"\\xFF\\xFF\\xFF\\xFF\\xFF\\xFF\\xFF\\xFF")
-        Lowest value posible is bytearray(b"\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00")
+        Maximum value is 18446744073709551615 (0xFFFFFFFFFFFFFFFF).
+        Minimum value is 0.
 
     Args:
         bytearray_: buffer to read from.
@@ -533,15 +531,13 @@ def get_lword(bytearray_: bytearray, byte_index: int) -> bytearray:
         Value read.
 
     Examples:
-        read lword value (here as example 0xAB\0xCD) from DB1.10 of a PLC
-        >>> from snap7 import Client
-        >>> data = Client().db_read(db_number=1, start=10, size=8)
+        >>> data = bytearray(b"\\x00\\x00\\x00\\x00\\x00\\x00\\xAB\\xCD")
         >>> get_lword(data, 0)
-            bytearray(b"\\x00\\x00\\x00\\x00\\x00\\x00\\xAB\\xCD")
+        43981
     """
-    data = bytearray_[byte_index : byte_index + 4]
-    dword = struct.unpack(">Q", struct.pack("8B", *data))[0]
-    return bytearray(dword)
+    data = bytearray_[byte_index : byte_index + 8]
+    lword: int = struct.unpack(">Q", struct.pack("8B", *data))[0]
+    return lword
 
 
 def get_ulint(bytearray_: bytearray, byte_index: int) -> int:
@@ -591,16 +587,72 @@ def get_date(bytearray_: bytearray, byte_index: int = 0) -> date:
     return date_val
 
 
-def get_ltime(bytearray_: bytearray, byte_index: int) -> str:
-    raise NotImplementedError
+def get_ltime(bytearray_: bytearray, byte_index: int) -> timedelta:
+    """Get LTIME value from bytearray.
 
+    Notes:
+        Datatype `LTIME` consists of 8 bytes (64-bit signed integer) representing
+        nanoseconds. Used in S7-1500 PLCs.
 
-def get_ltod(bytearray_: bytearray, byte_index: int) -> str:
-    raise NotImplementedError
+    Args:
+        bytearray_: buffer to read from.
+        byte_index: byte index from where to start reading.
 
+    Returns:
+        timedelta value.
 
-def get_ldt(bytearray_: bytearray, byte_index: int) -> str:
-    raise NotImplementedError
+    Examples:
+        >>> data = bytearray(8)
+        >>> data[:] = b'\\x00\\x00\\x00\\x00\\x3b\\x9a\\xca\\x00'  # 1 second in nanoseconds
+        >>> get_ltime(data, 0)
+        datetime.timedelta(seconds=1)
+    """
+    raw = bytearray_[byte_index : byte_index + 8]
+    nanoseconds: int = struct.unpack(">q", struct.pack("8B", *raw))[0]
+    return timedelta(microseconds=nanoseconds // 1000)
+
+
+def get_ltod(bytearray_: bytearray, byte_index: int) -> timedelta:
+    """Get LTOD (Long Time of Day) value from bytearray.
+
+    Notes:
+        Datatype `LTOD` consists of 8 bytes (64-bit unsigned integer) representing
+        nanoseconds since midnight. Used in S7-1500 PLCs.
+        Range: 0 to 86399999999999 ns.
+
+    Args:
+        bytearray_: buffer to read from.
+        byte_index: byte index from where to start reading.
+
+    Returns:
+        timedelta value representing time of day.
+    """
+    raw = bytearray_[byte_index : byte_index + 8]
+    nanoseconds: int = struct.unpack(">Q", struct.pack("8B", *raw))[0]
+    result = timedelta(microseconds=nanoseconds // 1000)
+    if result.days >= 1:
+        raise ValueError("LTOD value exceeds 24 hours")
+    return result
+
+
+def get_ldt(bytearray_: bytearray, byte_index: int) -> datetime:
+    """Get LDT (Long Date and Time) value from bytearray.
+
+    Notes:
+        Datatype `LDT` consists of 8 bytes (64-bit unsigned integer) representing
+        nanoseconds since 1970-01-01 00:00:00 UTC. Used in S7-1500 PLCs.
+
+    Args:
+        bytearray_: buffer to read from.
+        byte_index: byte index from where to start reading.
+
+    Returns:
+        datetime value.
+    """
+    raw = bytearray_[byte_index : byte_index + 8]
+    nanoseconds: int = struct.unpack(">Q", struct.pack("8B", *raw))[0]
+    epoch = datetime(1970, 1, 1)
+    return epoch + timedelta(microseconds=nanoseconds // 1000)
 
 
 def get_dtl(bytearray_: bytearray, byte_index: int) -> datetime:
@@ -662,7 +714,7 @@ def get_wchar(bytearray_: bytearray, byte_index: int) -> str:
         'C'
     """
     if bytearray_[byte_index] == 0:
-        return chr(bytearray_[1])
+        return chr(bytearray_[byte_index + 1])
     return bytearray_[byte_index : byte_index + 2].decode("utf-16-be")