fix more typos

Author: gijzelaerr

snap7/common.py

@@ -24,8 +24,8 @@ def _raise_error() -> NoReturn:
     error = f"""can't find snap7 shared library.
 
 This probably means you are installing python-snap7 from source. When no binary wheel is found for you architecture, pip
-install falls back on a source install. For this to work, you need to manually install the snap7 library, which python-snap7
-uses under the hood.
+install falls back on a source install. For this to work, you need to manually install the snap7 library, which
+python-snap7 uses under the hood.
 
 The shortest path to success is to try to get a binary wheel working. Probably you are running on an unsupported
 platform or python version. You are running:

snap7/logo.py

@@ -15,6 +15,35 @@
 logger = logging.getLogger(__name__)
 
 
+def parse_address(vm_address: str) -> tuple[int, WordLen]:
+    logger.debug(f"read, vm_address:{vm_address}")
+    if re.match(r"V[0-9]{1,4}\.[0-7]", vm_address):
+        logger.info(f"read, Bit address: {vm_address}")
+        address = vm_address[1:].split(".")
+        # transform string to int
+        address_byte = int(address[0])
+        address_bit = int(address[1])
+        start = (address_byte * 8) + address_bit
+        return start, WordLen.Bit
+    elif re.match("V[0-9]+", vm_address):
+        # byte value
+        logger.info(f"Byte address: {vm_address}")
+        start = int(vm_address[1:])
+        return start, WordLen.Byte
+    elif re.match("VW[0-9]+", vm_address):
+        # byte value
+        logger.info(f"Word address: {vm_address}")
+        start = int(vm_address[2:])
+        return start, WordLen.Word
+    elif re.match("VD[0-9]+", vm_address):
+        # byte value
+        logger.info(f"DWord address: {vm_address}")
+        start = int(vm_address[2:])
+        return start, WordLen.DWord
+    else:
+        raise ValueError("Unknown address format")
+
+
 class Logo(Client):
     """
     A snap7 Siemens Logo client:
@@ -61,35 +90,8 @@ def read(self, vm_address: str) -> int:
         area = Area.DB
         db_number = 1
         size = 1
-        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
-            logger.info(f"read, Bit address: {vm_address}")
-            address = vm_address[1:].split(".")
-            # transform string to int
-            address_byte = int(address[0])
-            address_bit = int(address[1])
-            start = (address_byte * 8) + address_bit
-            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 = 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 = 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 = WordLen.DWord
-        else:
-            logger.info("Unknown address format")
-            return 0
+        start, wordlen = parse_address(vm_address)
 
         type_ = wordlen.ctype
         data = (type_ * size)()
@@ -121,53 +123,29 @@ def write(self, vm_address: str, value: int) -> int:
         """
         area = Area.DB
         db_number = 1
-        amount = 1
-        wordlen: WordLen
-        logger.debug(f"write, vm_address:{vm_address}, value:{value}")
-        if re.match(r"^V[0-9]{1,4}\.[0-7]$", vm_address):
-            # bit value
-            logger.info(f"read, Bit address: {vm_address}")
-            address = vm_address[1:].split(".")
-            # transform string to int
-            address_byte = int(address[0])
-            address_bit = int(address[1])
-            start = (address_byte * 8) + address_bit
-            wordlen = WordLen.Bit
+        size = 1
+        start, wordlen = parse_address(vm_address)
+        type_ = wordlen.ctype
+
+        if wordlen == WordLen.Bit:
+            type_ = WordLen.Byte.ctype
             if value > 0:
                 data = bytearray([1])
             else:
                 data = bytearray([0])
-        elif re.match("^V[0-9]+$", vm_address):
-            # byte value
-            logger.info(f"Byte address: {vm_address}")
-            start = int(vm_address[1:])
-            wordlen = WordLen.Byte
+        elif 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 = WordLen.Word
+        elif 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 = WordLen.DWord
+        elif wordlen == WordLen.DWord:
             data = bytearray(struct.pack(">l", value))
         else:
-            logger.info(f"write, Unknown address format: {vm_address}")
-            return 1
-
-        if wordlen == WordLen.Bit:
-            type_ = WordLen.Byte.ctype
-        else:
-            type_ = wordlen.ctype
+            raise ValueError(f"Unknown wordlen {wordlen}")
 
-        cdata = (type_ * amount).from_buffer_copy(data)
+        cdata = (type_ * size).from_buffer_copy(data)
 
         logger.debug(f"write, vm_address:{vm_address} value:{value}")
 
-        result = self._lib.Cli_WriteArea(self._s7_client, area, db_number, start, amount, wordlen, byref(cdata))
+        result = self._lib.Cli_WriteArea(self._s7_client, area, db_number, start, size, wordlen, byref(cdata))
         check_error(result, context="client")
         return result

snap7/server/__init__.py

@@ -44,7 +44,7 @@ def __init__(self, log: bool = True):
             event logging to python logging.
 
         Args:
-            log: `True` for enabling the event logging. Optinoal.
+            log: `True` for enabling the event logging.
         """
         self._lib: Snap7CliProtocol = load_library()
         self.create()
@@ -110,49 +110,49 @@ def set_events_callback(self, call_back: Callable[..., Any]) -> int:
         logger.info("setting event callback")
         callback_wrap: Callable[..., Any] = CFUNCTYPE(None, c_void_p, POINTER(SrvEvent), c_int)
 
-        def wrapper(_: Optional[c_void_p], pevent: SrvEvent, __: int) -> int:
+        def wrapper(_: Optional[c_void_p], event: SrvEvent, __: int) -> int:
             """Wraps python function into a ctypes function
 
             Args:
                 _: not used
-                pevent: pointer to snap7 event struct
+                event: pointer to snap7 event struct
                 __: not used
 
             Returns:
                 Should return an int
             """
-            logger.info(f"callback event: {self.event_text(pevent.contents)}")
-            call_back(pevent.contents)
+            logger.info(f"callback event: {self.event_text(event.contents)}")
+            call_back(event.contents)
             return 0
 
         self._callback = cast(type[CFuncPtr], callback_wrap(wrapper))
-        usrPtr = c_void_p()
-        return self._lib.Srv_SetEventsCallback(self._s7_server, self._callback, usrPtr)
+        data = c_void_p()
+        return self._lib.Srv_SetEventsCallback(self._s7_server, self._callback, data)
 
     @error_wrap(context="server")
     def set_read_events_callback(self, call_back: Callable[..., Any]) -> int:
         """Sets the user callback that the Server object has to call when a Read
             event is created.
 
         Args:
-            call_back: a callback function that accepts a pevent argument.
+            call_back: a callback function that accepts an event argument.
         """
         logger.info("setting read event callback")
         callback_wrapper: Callable[..., Any] = CFUNCTYPE(None, c_void_p, POINTER(SrvEvent), c_int)
 
-        def wrapper(usrptr: Optional[c_void_p], pevent: SrvEvent, size: int) -> int:
+        def wrapper(_: Optional[c_void_p], event: SrvEvent, __: int) -> int:
             """Wraps python function into a ctypes function
 
             Args:
-                usrptr: not used
-                pevent: pointer to snap7 event struct
-                size:
+                _: data, not used
+                event: pointer to snap7 event struct
+                __: size, not used
 
             Returns:
                 Should return an int
             """
-            logger.info(f"callback event: {self.event_text(pevent.contents)}")
-            call_back(pevent.contents)
+            logger.info(f"callback event: {self.event_text(event.contents)}")
+            call_back(event.contents)
             return 0
 
         self._read_callback = callback_wrapper(wrapper)
@@ -168,16 +168,16 @@ def log_callback(event: SrvEvent) -> None:
         self.set_events_callback(log_callback)
 
     @error_wrap(context="server")
-    def start(self, tcpport: int = 102) -> int:
+    def start(self, tcp_port: int = 102) -> int:
         """Starts the server.
 
         Args:
-            tcpport: port that the server will listen. Optional.
+            tcp_port: port that the server will listen. Optional.
         """
-        if tcpport != 102:
-            logger.info(f"setting server TCP port to {tcpport}")
-            self.set_param(Parameter.LocalPort, tcpport)
-        logger.info(f"starting server on 0.0.0.0:{tcpport}")
+        if tcp_port != 102:
+            logger.info(f"setting server TCP port to {tcp_port}")
+            self.set_param(Parameter.LocalPort, tcp_port)
+        logger.info(f"starting server on 0.0.0.0:{tcp_port}")
         return self._lib.Srv_Start(self._s7_server)
 
     @error_wrap(context="server")
@@ -208,11 +208,11 @@ def get_status(self) -> Tuple[str, str, int]:
         error = self._lib.Srv_GetStatus(self._s7_server, byref(server_status), byref(cpu_status), byref(clients_count))
         check_error(error)
         logger.debug(f"status server {server_status.value} cpu {cpu_status.value} clients {clients_count.value}")
-        return (server_statuses[server_status.value], cpu_statuses[cpu_status.value], clients_count.value)
+        return server_statuses[server_status.value], cpu_statuses[cpu_status.value], clients_count.value
 
     @error_wrap(context="server")
     def unregister_area(self, area: SrvArea, index: int) -> int:
-        """'Unshares' a memory area previously shared with Srv_RegisterArea().
+        """Unregisters a memory area previously registered with Srv_RegisterArea().
 
         Notes:
             That memory block will be no longer visible by the clients.
@@ -345,7 +345,7 @@ def get_param(self, number: int) -> int:
         Returns:
             Value of the parameter.
         """
-        logger.debug(f"retreiving param number {number}")
+        logger.debug(f"retrieving param number {number}")
         value = c_int()
         code = self._lib.Srv_GetParam(self._s7_server, number, byref(value))
         check_error(code)
@@ -377,32 +377,32 @@ def clear_events(self) -> int:
         return self._lib.Srv_ClearEvents(self._s7_server)
 
 
-def mainloop(tcpport: int = 1102, init_standard_values: bool = False) -> None:
+def mainloop(tcp_port: int = 1102, init_standard_values: bool = False) -> None:
     """Init a fake Snap7 server with some default values.
 
     Args:
-        tcpport: port that the server will listen.
+        tcp_port: port that the server will listen.
         init_standard_values: if `True` will init some defaults values to be read on DB0.
     """
 
     server = Server()
     size = 100
-    DBdata: CDataArrayType = (WordLen.Byte.ctype * size)()
-    PAdata: CDataArrayType = (WordLen.Byte.ctype * size)()
-    TMdata: CDataArrayType = (WordLen.Byte.ctype * size)()
-    CTdata: CDataArrayType = (WordLen.Byte.ctype * size)()
-    server.register_area(SrvArea.DB, 1, DBdata)
-    server.register_area(SrvArea.PA, 1, PAdata)
-    server.register_area(SrvArea.TM, 1, TMdata)
-    server.register_area(SrvArea.CT, 1, CTdata)
+    db_data: CDataArrayType = (WordLen.Byte.ctype * size)()
+    pa_data: CDataArrayType = (WordLen.Byte.ctype * size)()
+    tm_data: CDataArrayType = (WordLen.Byte.ctype * size)()
+    ct_data: CDataArrayType = (WordLen.Byte.ctype * size)()
+    server.register_area(SrvArea.DB, 1, db_data)
+    server.register_area(SrvArea.PA, 1, pa_data)
+    server.register_area(SrvArea.TM, 1, tm_data)
+    server.register_area(SrvArea.CT, 1, ct_data)
 
     if init_standard_values:
         logger.info("initialising with standard values")
         ba = _init_standard_values()
         userdata = WordLen.Byte.ctype * len(ba)
         server.register_area(SrvArea.DB, 0, userdata.from_buffer(ba))
 
-    server.start(tcpport=tcpport)
+    server.start(tcp_port=tcp_port)
     while True:
         while True:
             event = server.pick_event()

snap7/util/getters.py

@@ -62,8 +62,7 @@ def get_word(bytearray_: bytearray, byte_index: int) -> bytearray:
         Word value.
 
     Examples:
-        >>> data = bytearray([0, 100])  # two bytes for a word
-        >>> get_word(data, 0)
+        >>> get_word(bytearray([0, 100]), 0)
             100
     """
     data = bytearray_[byte_index : byte_index + 2]
@@ -88,8 +87,7 @@ def get_int(bytearray_: bytearray, byte_index: int) -> int:
         Value read.
 
     Examples:
-        >>> data = bytearray([0, 255])
-        >>> get_int(data, 0)
+        >>> get_int(bytearray([0, 255]), 0)
             255
     """
     data = bytearray_[byte_index : byte_index + 2]
@@ -120,12 +118,7 @@ def get_uint(bytearray_: bytearray, byte_index: int) -> int:
         >>> get_uint(data, 0)
             65535
     """
-    data = bytearray_[byte_index : byte_index + 2]
-    data[1] = data[1] & 0xFF
-    data[0] = data[0] & 0xFF
-    packed = struct.pack("2B", *data)
-    value: int = struct.unpack(">H", packed)[0]
-    return value
+    return int(get_word(bytearray_, byte_index))
 
 
 def get_real(bytearray_: bytearray, byte_index: int) -> float:
@@ -467,7 +460,7 @@ def get_sint(bytearray_: bytearray, byte_index: int) -> int:
     return value
 
 
-def get_lint(bytearray_: bytearray, byte_index: int) -> NoReturn:
+def get_lint(bytearray_: bytearray, byte_index: int) -> int:
     """Get the long int
 
     THIS VALUE IS NEITHER TESTED NOR VERIFIED BY A REAL PLC AT THE MOMENT
@@ -492,10 +485,9 @@ def get_lint(bytearray_: bytearray, byte_index: int) -> NoReturn:
             12345
     """
 
-    # raw_lint = bytearray_[byte_index:byte_index + 8]
-    # lint = struct.unpack('>q', struct.pack('8B', *raw_lint))[0]
-    # return lint
-    raise NotImplementedError
+    raw_lint = bytearray_[byte_index : byte_index + 8]
+    lint = struct.unpack(">q", struct.pack("8B", *raw_lint))[0]
+    return int(lint)
 
 
 def get_lreal(bytearray_: bytearray, byte_index: int) -> float:
@@ -548,10 +540,9 @@ def get_lword(bytearray_: bytearray, byte_index: int) -> bytearray:
         >>> get_lword(data, 0)
             bytearray(b"\\x00\\x00\\x00\\x00\\x00\\x00\\xAB\\xCD")
     """
-    #  data = bytearray_[byte_index:byte_index + 4]
-    #  dword = struct.unpack('>Q', struct.pack('8B', *data))[0]
-    #  return bytearray(dword)
-    raise NotImplementedError
+    data = bytearray_[byte_index : byte_index + 4]
+    dword = struct.unpack(">Q", struct.pack("8B", *data))[0]
+    return bytearray(dword)
 
 
 def get_ulint(bytearray_: bytearray, byte_index: int) -> int: