Types and lib location feature (#255)

* added small int support

* change order

* spaces between functions

* changed functions small_int to usint (unsigned small int)

* change in db_row get value y set value changed comparation from sint to usint (0-255)

* added to db_row get_value and set_value comparation to sint (-128 to 127)

* added support for SMALL INT (-127 to 128)

* changes in test_util in bytearray, added numbers to ocuppe all bytes of the data structure in DTL

* test passed | changed _bytearray in test to match the lenght for the test

* added doc

* change space

* arreglos de espacios en blanco

* added s7areaDB  for default

* added default to db and db_row -> area=S7AreaDB

* same

* added read area | write area --- TODO tests

* correction to the db1['test'].read() -> added client

* read area debe corregirse

* correction to client.read_area of function write in class db_row

* corrections to def write in class db_row

* Small changes

- Types.py:
 * Added enum class to handle areas
- Utils:
 * Added some missing types
 * Added area parameter to the DB class, this way you could handle the inputs, outputs, or marks as a DB for reading it TODO: add tests
- Common.py:
   * Missing types
- Client.py:
 * Added an optional init parameter for the lib location, this way people who can't instal properly the library, they can specify the path to the .dll file.

* Types.py readd S7Areas as alone values

* Test fixes 1

* Typing errors fix 2

* White spaces fixes

* White spaces

* Remove enum Areas, so we stick to the ADict class

* Remove it TODOs in util.py

* Utils type export fix

* _type parameter switch to type_

* Fix typing for areas Adict

* Migration to Enum for Areas and WordLen

* switch assert date

* whitespaces

* Fix typing errors in util file

* ignore typing

* fix test

* whitespaces

* Added find locally snap7.dll
Fix logo errors

* fix assert error in date

* types.py:
* remove it string multiline methods added-

common.py
* Remove it type ignore from ctypes import

* Added common test and change common find_locally from using os to pathlib

* Pathlib

* Path lib fix to find in the current working directory

Author: lautarodapin

.gitignore

@@ -53,3 +53,9 @@ doc/_build/
 venv/
 
 .DS_Store
+
+# VSCode
+.vscode/
+
+# DLL
+snap7.dll

snap7/client.py

@@ -1,7 +1,10 @@
 import logging
+import os
 import platform
 from ctypes import c_char
 from ctypes.util import find_library
+from typing import Optional, Union
+import pathlib
 
 from snap7.exceptions import Snap7Exception
 
@@ -21,7 +24,7 @@ class ADict(dict):
     Accessing dict keys like an attribute.
     """
     __getattr__ = dict.__getitem__
-    __setattr__ = dict.__setitem__
+    __setattr__ = dict.__setitem__  # type: ignore
 
 
 class Snap7Library:
@@ -30,6 +33,7 @@ class Snap7Library:
     sure the library is loaded only once.
     """
     _instance = None
+    lib_location: Optional[str]
 
     def __new__(cls, *args, **kwargs):
         if not cls._instance:
@@ -38,17 +42,17 @@ def __new__(cls, *args, **kwargs):
             cls._instance.cdll = None
         return cls._instance
 
-    def __init__(self, lib_location=None):
-        if self.cdll:
+    def __init__(self, lib_location: Optional[str] = None):
+        if self.cdll:  # type: ignore
             return
-        self.lib_location = lib_location or self.lib_location or find_library('snap7')
+        self.lib_location = lib_location or self.lib_location or find_library('snap7') or find_locally('snap7')
         if not self.lib_location:
             msg = "can't find snap7 library. If installed, try running ldconfig"
             raise Snap7Exception(msg)
         self.cdll = cdll.LoadLibrary(self.lib_location)
 
 
-def load_library(lib_location: str = None):
+def load_library(lib_location: Optional[str] = None):
     """
     :returns: a ctypes cdll object with the snap7 shared library loaded.
     """
@@ -87,3 +91,10 @@ def error_text(error, context: str = "client") -> bytes:
     elif context == "partner":
         library.Par_ErrorText(error, text, len_)
     return text.value
+
+
+def find_locally(fname):
+    file = pathlib.Path.cwd() / f"{fname}.dll"
+    if file.exists():
+        return str(file)
+    return None

snap7/common.py

@@ -11,7 +11,7 @@
 from snap7 import types
 from snap7.common import check_error, load_library, ipv4
 from snap7.exceptions import Snap7Exception
-from snap7.types import S7Object
+from snap7.types import S7Object, WordLen
 from snap7.types import param_types
 
 logger = logging.getLogger(__name__)
@@ -88,11 +88,11 @@ def read(self, vm_address: str):
         :param vm_address: of Logo memory (e.g. V30.1, VW32, V24)
         :returns: integer
         """
-        area = types.S7AreaDB
+        area = types.Areas.DB
         db_number = 1
         size = 1
         start = 0
-        wordlen = 0
+        wordlen: WordLen
         logger.debug(f"read, vm_address:{vm_address}")
         if re.match(r"V[0-9]{1,4}\.[0-7]", vm_address):
             # bit value
@@ -102,42 +102,42 @@ def read(self, vm_address: str):
             address_byte = int(address[0])
             address_bit = int(address[1])
             start = (address_byte * 8) + address_bit
-            wordlen = types.S7WLBit
+            wordlen = WordLen.Bit
         elif re.match("V[0-9]+", vm_address):
             # byte value
             logger.info(f"Byte address: {vm_address}")
             start = int(vm_address[1:])
-            wordlen = types.S7WLByte
+            wordlen = WordLen.Byte
         elif re.match("VW[0-9]+", vm_address):
             # byte value
             logger.info(f"Word address: {vm_address}")
             start = int(vm_address[2:])
-            wordlen = types.S7WLWord
+            wordlen = WordLen.Word
         elif re.match("VD[0-9]+", vm_address):
             # byte value
             logger.info(f"DWord address: {vm_address}")
             start = int(vm_address[2:])
-            wordlen = types.S7WLDWord
+            wordlen = WordLen.DWord
         else:
             logger.info("Unknown address format")
             return 0
 
-        type_ = snap7.types.wordlen_to_ctypes[wordlen]
+        type_ = snap7.types.wordlen_to_ctypes[wordlen.value]
         data = (type_ * size)()
 
-        logger.debug(f"start:{start}, wordlen:{wordlen}, data-length:{len(data)}")
+        logger.debug(f"start:{start}, wordlen:{wordlen.name}={wordlen.value}, data-length:{len(data)}")
 
-        result = self.library.Cli_ReadArea(self.pointer, area, db_number, start,
-                                           size, wordlen, byref(data))
+        result = self.library.Cli_ReadArea(self.pointer, area.value, db_number, start,
+                                           size, wordlen.value, byref(data))
         check_error(result, context="client")
         # transform result to int value
-        if wordlen == types.S7WLBit:
+        if wordlen == WordLen.Bit:
             return data[0]
-        if wordlen == types.S7WLByte:
+        if wordlen == WordLen.Byte:
             return struct.unpack_from(">B", data)[0]
-        if wordlen == types.S7WLWord:
+        if wordlen == WordLen.Word:
             return struct.unpack_from(">h", data)[0]
-        if wordlen == types.S7WLDWord:
+        if wordlen == WordLen.DWord:
             return struct.unpack_from(">l", data)[0]
 
     def write(self, vm_address: str, value: int) -> int:
@@ -148,11 +148,11 @@ def write(self, vm_address: str, value: int) -> int:
         :param vm_address: write offset
         :param value: integer
         """
-        area = types.S7AreaDB
+        area = types.Areas.DB
         db_number = 1
         start = 0
         amount = 1
-        wordlen = 0
+        wordlen: WordLen
         data = bytearray(0)
         logger.debug(f"write, vm_address:{vm_address}, value:{value}")
         if re.match(r"^V[0-9]{1,4}\.[0-7]$", vm_address):
@@ -163,7 +163,7 @@ def write(self, vm_address: str, value: int) -> int:
             address_byte = int(address[0])
             address_bit = int(address[1])
             start = (address_byte * 8) + address_bit
-            wordlen = types.S7WLBit
+            wordlen = WordLen.Bit
             if value > 0:
                 data = bytearray([1])
             else:
@@ -172,34 +172,34 @@ def write(self, vm_address: str, value: int) -> int:
             # byte value
             logger.info(f"Byte address: {vm_address}")
             start = int(vm_address[1:])
-            wordlen = types.S7WLByte
+            wordlen = WordLen.Byte
             data = bytearray(struct.pack(">B", value))
         elif re.match("^VW[0-9]+$", vm_address):
             # byte value
             logger.info(f"Word address: {vm_address}")
             start = int(vm_address[2:])
-            wordlen = types.S7WLWord
+            wordlen = WordLen.Word
             data = bytearray(struct.pack(">h", value))
         elif re.match("^VD[0-9]+$", vm_address):
             # byte value
             logger.info(f"DWord address: {vm_address}")
             start = int(vm_address[2:])
-            wordlen = types.S7WLDWord
+            wordlen = WordLen.DWord
             data = bytearray(struct.pack(">l", value))
         else:
             logger.info(f"write, Unknown address format: {vm_address}")
             return 1
 
-        if wordlen == types.S7WLBit:
-            type_ = snap7.types.wordlen_to_ctypes[types.S7WLByte]
+        if wordlen == WordLen.Bit:
+            type_ = snap7.types.wordlen_to_ctypes[WordLen.Byte.value]
         else:
-            type_ = snap7.types.wordlen_to_ctypes[wordlen]
+            type_ = snap7.types.wordlen_to_ctypes[wordlen.value]
 
         cdata = (type_ * amount).from_buffer_copy(data)
 
         logger.debug(f"write, vm_address:{vm_address} value:{value}")
 
-        result = self.library.Cli_WriteArea(self.pointer, area, db_number, start, amount, wordlen, byref(cdata))
+        result = self.library.Cli_WriteArea(self.pointer, area.value, db_number, start, amount, wordlen.value, byref(cdata))
         check_error(result, context="client")
         return result
 
@@ -214,7 +214,7 @@ def db_read(self, db_number: int, start: int, size: int) -> bytearray:
         """
         logger.debug(f"db_read, db_number:{db_number}, start:{start}, size:{size}")
 
-        type_ = snap7.types.wordlen_to_ctypes[snap7.types.S7WLByte]
+        type_ = snap7.types.wordlen_to_ctypes[WordLen.Byte.value]
         data = (type_ * size)()
         result = (self.library.Cli_DBRead(
             self.pointer, db_number, start, size,
@@ -230,8 +230,8 @@ def db_write(self, db_number: int, start: int, data: bytearray) -> int:
         :param start: start address for Logo7 0..951 / Logo8 0..1469
         :param data: bytearray
         """
-        wordlen = snap7.types.S7WLByte
-        type_ = snap7.types.wordlen_to_ctypes[wordlen]
+        wordlen = WordLen.Byte
+        type_ = snap7.types.wordlen_to_ctypes[wordlen.value]
         size = len(data)
         cdata = (type_ * size).from_buffer_copy(data)
         logger.debug(f"db_write db_number:{db_number} start:{start} size:{size} data:{data}")

snap7/logo.py

@@ -293,10 +293,10 @@ def clear_events(self) -> int:
 def mainloop(tcpport: int = 1102):
     server = snap7.server.Server()
     size = 100
-    DBdata = (snap7.types.wordlen_to_ctypes[snap7.types.S7WLByte] * size)()
-    PAdata = (snap7.types.wordlen_to_ctypes[snap7.types.S7WLByte] * size)()
-    TMdata = (snap7.types.wordlen_to_ctypes[snap7.types.S7WLByte] * size)()
-    CTdata = (snap7.types.wordlen_to_ctypes[snap7.types.S7WLByte] * size)()
+    DBdata = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
+    PAdata = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
+    TMdata = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
+    CTdata = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
     server.register_area(snap7.types.srvAreaDB, 1, DBdata)
     server.register_area(snap7.types.srvAreaPA, 1, PAdata)
     server.register_area(snap7.types.srvAreaTM, 1, TMdata)

snap7/server.py

@@ -4,6 +4,7 @@
 import ctypes
 
 from snap7.common import ADict
+from enum import Enum
 
 S7Object = ctypes.c_void_p
 buffer_size = 65536
@@ -51,14 +52,26 @@
 mkEvent = 0
 mkLog = 1
 
+
 # Area ID
+class Areas(Enum):
+    PE = 0x81
+    PA = 0x82
+    MK = 0x83
+    DB = 0x84
+    CT = 0x1C
+    TM = 0x1D
+
+
+# Leave it for now
 S7AreaPE = 0x81
 S7AreaPA = 0x82
 S7AreaMK = 0x83
 S7AreaDB = 0x84
 S7AreaCT = 0x1C
 S7AreaTM = 0x1D
 
+
 areas = ADict({
     'PE': 0x81,
     'PA': 0x82,
@@ -68,7 +81,19 @@
     'TM': 0x1D,
 })
 
+
 # Word Length
+class WordLen(Enum):
+    Bit = 0x01
+    Byte = 0x02
+    Word = 0x04
+    DWord = 0x06
+    Real = 0x08
+    Counter = 0x1C
+    Timer = 0x1D
+
+
+# Leave it for now
 S7WLBit = 0x01
 S7WLByte = 0x02
 S7WLWord = 0x04
@@ -212,6 +237,10 @@ class S7DataItem(ctypes.Structure):
         ('pData', ctypes.POINTER(ctypes.c_uint8))
     ]
 
+    def __str__(self) -> str:
+        return f"<S7DataItem Area: {self.Area} WordLen: {self.WordLen} Result: {self.Result} "\
+               f"DBNumber: {self.DBNumber} Start: {self.Start} Amount: {self.Amount} pData: {self.pData}>"
+
 
 class S7CpuInfo(ctypes.Structure):
     _fields_ = [
@@ -222,6 +251,10 @@ class S7CpuInfo(ctypes.Structure):
         ('ModuleName', ctypes.c_char * 25)
     ]
 
+    def __str__(self):
+        return f"<S7CpuInfo ModuleTypeName: {self.ModuleTypeName} SerialNumber: {self.SerialNumber} "\
+               f"ASName: {self.ASName} Copyright: {self.Copyright} ModuleName: {self.ModuleName}>"
+
 
 class S7SZLHeader(ctypes.Structure):
     """
@@ -233,6 +266,9 @@ class S7SZLHeader(ctypes.Structure):
         ('NDR', ctypes.c_uint16)
     ]
 
+    def __str__(self) -> str:
+        return f"<S7SZLHeader LengthDR: {self.LengthDR}, NDR: {self.NDR}>"
+
 
 class S7SZL(ctypes.Structure):
     """See §33.1 of System Software for S7-300/400 System and Standard Functions"""
@@ -241,6 +277,9 @@ class S7SZL(ctypes.Structure):
         ('Data', ctypes.c_byte * (0x4000 - 4))
     ]
 
+    def __str__(self) -> str:
+        return f"<S7SZL Header: {self.S7SZHeader}, Data: {self.Data}>"
+
 
 class S7SZLList(ctypes.Structure):
     _fields_ = [
@@ -266,6 +305,10 @@ class S7CpInfo(ctypes.Structure):
         ('MaxBusRate', ctypes.c_uint16)
     ]
 
+    def __str__(self) -> str:
+        return f"<S7CpInfo MaxPduLength: {self.MaxPduLength} MaxConnections: {self.MaxConnections} "\
+               f"MaxMpiRate: {self.MaxMpiRate} MaxBusRate: {self.MaxBusRate}>"
+
 
 class S7Protection(ctypes.Structure):
     """See §33.19 of System Software for S7-300/400 System and Standard Functions"""