Merge pull request #5 from ottowayi/develop

new get_module_info method

pycomm3/__init__.py

@@ -24,7 +24,7 @@
 # SOFTWARE.
 #
 
-__version_info__ = (0, 1, 1)
+__version_info__ = (0, 2, 0)
 __version__ = '.'.join(f'{x}' for x in __version_info__)
 
 

pycomm3/base.py

@@ -29,12 +29,12 @@
 from autologging import logged
 
 from . import DataError, CommError
-from .bytes_ import (pack_usint, pack_udint, pack_uint, pack_dint, unpack_dint, unpack_uint, unpack_usint,
-                     print_bytes_line, print_bytes_msg, DATA_FUNCTION_SIZE, UNPACK_DATA_FUNCTION)
+from .bytes_ import (pack_usint, pack_udint, pack_uint, pack_dint, unpack_dint, unpack_uint, unpack_usint, pack_ulong,
+                     unpack_udint, print_bytes_line, print_bytes_msg, DATA_FUNCTION_SIZE, UNPACK_DATA_FUNCTION)
 from .const import (DATA_ITEM, DATA_TYPE, TAG_SERVICES_REQUEST, EXTEND_CODES, ENCAPSULATION_COMMAND, EXTENDED_SYMBOL,
                     ELEMENT_ID, CLASS_CODE, PADDING_BYTE, CONNECTION_SIZE, CLASS_ID, INSTANCE_ID, FORWARD_CLOSE,
                     FORWARD_OPEN, LARGE_FORWARD_OPEN, CONNECTION_MANAGER_INSTANCE, PRIORITY, TIMEOUT_MULTIPLIER,
-                    TIMEOUT_TICKS, TRANSPORT_CLASS, ADDRESS_ITEM)
+                    TIMEOUT_TICKS, TRANSPORT_CLASS, ADDRESS_ITEM, UNCONNECTED_SEND, PRODUCT_TYPES, VENDORS, STATES)
 from .socket_ import Socket
 
 
@@ -234,6 +234,14 @@ def register_session(self):
         self.__log.warning(self._status)
         return None
 
+    def un_register_session(self):
+        """ Un-register a connection
+
+        """
+        message = self.build_header(ENCAPSULATION_COMMAND['unregister_session'], 0)
+        self._send(message)
+        self._session = None
+
     def forward_open(self):
         """ CIP implementation of the forward open message
 
@@ -353,13 +361,78 @@ def forward_close(self):
         self.__log.warning(self._status)
         return False
 
-    def un_register_session(self):
-        """ Un-register a connection
+    def get_module_info(self, slot):
+        try:
+            if not self._target_is_connected:
+                if not self.forward_open():
+                    self._status = (10, "Target did not connected. get_plc_name will not be executed.")
+                    self.__log.warning(self._status)
+                    raise DataError(self._status[1])
+
+            msg = [
+                # unnconnected send portion
+                UNCONNECTED_SEND,
+                b'\x02',
+                CLASS_ID['8-bit'],
+                b'\x06',  # class
+                INSTANCE_ID["8-bit"],
+                b'\x01',
+                b'\x0A',  # priority
+                b'\x0e\x06\x00',
+
+                # Identity request portion
+                b'\x01',  # Service
+                b'\x02',
+                CLASS_ID['8-bit'],
+                CLASS_CODE['Identity Object'],
+                INSTANCE_ID["8-bit"],
+                b'\x01',  # Instance 1
+                b'\x01\x00',
+                b'\x01',  # backplane
+                pack_usint(slot),
+            ]
+            request = self.build_common_packet_format(DATA_ITEM['Unconnected'],
+                                                      b''.join(msg),
+                                                      ADDRESS_ITEM['UCMM'], )
+            reply = self.send_rr_data(request)
+
+            if reply:
+                info = self._parse_identity_object(reply)
+                # self._info = {**self._info, **info}
+                return info
+            else:
+                raise DataError('send_rr_data did not return valid data')
 
-        """
-        message = self.build_header(ENCAPSULATION_COMMAND['unregister_session'], 0)
-        self._send(message)
-        self._session = None
+        except Exception as err:
+            raise DataError(err)
+
+    @staticmethod
+    def _parse_identity_object(reply):
+        vendor = unpack_uint(reply[44:46])
+        product_type = unpack_uint(reply[46:48])
+        product_code = unpack_uint(reply[48:50])
+        major_fw = int(reply[50])
+        minor_fw = int(reply[51])
+        status = f'{unpack_uint(reply[52:54]):0{16}b}'
+        serial_number = f'{unpack_udint(reply[54:58]):0{8}x}'
+        product_name_len = int(reply[58])
+        tmp = 59 + product_name_len
+        device_type = reply[59:tmp].decode()
+
+        state = unpack_uint(reply[tmp:tmp+4]) if reply[tmp:] else -1  # some modules don't return a state
+
+        return {
+            'vendor': VENDORS.get(vendor, 'UNKNOWN'),
+            'product_type': PRODUCT_TYPES.get(product_type, 'UNKNOWN'),
+            'product_code': product_code,
+            'version_major': major_fw,
+            'version_minor': minor_fw,
+            'revision': f'{major_fw}.{minor_fw}',
+            'serial': serial_number,
+            'device_type': device_type,
+            'status': status,
+            'state': STATES.get(state, 'UNKNOWN'),
+        }
 
     def _send(self, message):
         """

pycomm3/bytes_.py

@@ -65,6 +65,14 @@ def pack_lint(l):
     return struct.pack('<q', l)
 
 
+def pack_long(l):
+    return struct.pack('<l', l)
+
+
+def pack_ulong(l):
+    return struct.pack('<L', l)
+
+
 def unpack_bool(st):
     return 1 if not st[0] == 0 else 0
 
@@ -107,6 +115,19 @@ def unpack_lint(st):
     return int(struct.unpack('<q', st[0:8])[0])
 
 
+def unpack_ulint(st):
+    """unpack 4 bytes little endian to int"""
+    return int(struct.unpack('<Q', st[0:8])[0])
+
+
+def unpack_long(st):
+    return int(struct.unpack('<l', st[0:4])[0])
+
+
+def unpack_ulong(st):
+    return int(struct.unpack('<L', st[0:4])[0])
+
+
 def print_bytes_line(msg):
     out = ''
     for ch in msg:

pycomm3/clx.py

@@ -36,7 +36,7 @@
                     REPLAY_INFO, TAG_SERVICES_REQUEST, PADDING_BYTE, ELEMENT_ID, DATA_ITEM, ADDRESS_ITEM,
                     CLASS_ID, CLASS_CODE, INSTANCE_ID, INSUFFICIENT_PACKETS, REPLY_START,
                     MULTISERVICE_READ_OVERHEAD, MULTISERVICE_WRITE_OVERHEAD, MIN_VER_INSTANCE_IDS, REQUEST_PATH_SIZE,
-                    VENDORS, PRODUCT_TYPES)
+                    VENDORS, PRODUCT_TYPES, KEYSWITCH)
 
 
 @logged
@@ -682,7 +682,7 @@ def get_plc_info(self):
 
             msg = [
                 pack_uint(self._get_sequence()),
-                b'\x01',
+                b'\x01',  # Service
                 REQUEST_PATH_SIZE,
                 CLASS_ID['8-bit'],
                 CLASS_CODE['Identity Object'],
@@ -697,7 +697,7 @@ def get_plc_info(self):
             reply = self.send_unit_data(request)
 
             if reply:
-                info = self._parse_identity_object(reply)
+                info = self._parse_plc_info(reply)
                 self._info = {**self._info, **info}
                 return info
             else:
@@ -720,28 +720,29 @@ def _parse_plc_name(reply):
             raise DataError(err)
 
     @staticmethod
-    def _parse_identity_object(reply):
+    def _parse_plc_info(reply):
 
         data = reply[REPLY_START:]
         vendor = unpack_uint(data[0:2])
         product_type = unpack_uint(data[2:4])
         product_code = unpack_uint(data[4:6])
         major_fw = int(data[6])
         minor_fw = int(data[7])
-        keyswitch = data[8:10]
+        keyswitch = KEYSWITCH.get(int(data[8]), {}).get(int(data[9]), 'UNKNOWN')
         serial_number = f'{unpack_udint(data[10:14]):0{8}x}'
         device_type_len = int(data[14])
         device_type = data[15:15 + device_type_len].decode()
 
         return {
-            'vendor': VENDORS[vendor],
-            'product_type': PRODUCT_TYPES[product_type],
+            'vendor': VENDORS.get(vendor, 'UNKNOWN'),
+            'product_type': PRODUCT_TYPES.get(product_type, 'UNKNOWN'),
             'product_code': product_code,
             'version_major': major_fw,
             'version_minor': minor_fw,
             'revision': f'{major_fw}.{minor_fw}',
             'serial': serial_number,
-            'device_type': device_type
+            'device_type': device_type,
+            'keyswitch': keyswitch
         }
 
     def get_tag_list(self, program=None, cache=True):

pycomm3/const.py

@@ -324,7 +324,6 @@
     'Direct Network': b'\x02'
 }
 
-
 CONNECTION_PARAMETER = {
     'PLC5': 0x4302,
     'SLC500': 0x4302,
@@ -475,6 +474,35 @@
     240: "Error code in EXT STS Byte"
 }
 
+# States defined in CIP Spec Vol 1, chapter 5, Identity Object
+STATES = {
+    0: 'Nonexistent',
+    1: 'Device Self Testing',
+    2: 'Standby',
+    3: 'Operational',
+    4: 'Major Recoverable Fault',
+    5: 'Major Unrecoverable Fault',
+    **{i: 'Reserved' for i in range(6, 255)},
+    255: 'Default for Get_Attributes_All service'
+
+}
+
+# From Rockwell KB Article #28917
+KEYSWITCH = {
+    96: {
+        16: 'RUN',
+        17: 'RUN',
+        48: 'REMOTE RUN',
+        49: 'REMOTE RUN'
+    },
+    112: {
+        32: 'PROG',
+        33: 'PROG',
+        48: 'REMOTE PROG',
+        49: 'REMOTE PROG'
+    }
+}
+
 #  Taken from PyLogix
 # List originally came from Wireshark /epan/dissectors/packet-cip.c
 PRODUCT_TYPES = {
@@ -1735,4 +1763,4 @@
     1238: 'Global Engineering Solutions Co., Ltd.',
     1239: 'ALTE Transportation, S.L.',
     1240: 'Penko Engineering B.V.'
-}
+}