Merge pull request #235 from ottowayi/develop

1.2.10

Author: ottowayi

docs/getting_started.rst

@@ -39,7 +39,12 @@ There are three possible forms:
     .. note::
 
         Both the IP Address and IP Address/Slot options are shortcuts, they will be replaced with the
-        CIP path automatically in the LogixDriver and SLCDriver.
+        CIP path automatically in the LogixDriver and SLCDriver, the CIPDriver will not modify the path.
+
+    .. note::
+
+        Path segments may be delimited by forward or back slashes or commas, e.g. ``10.10.30.100,bp,0``.
+        To use a custom port, provide it following a colon with the IP address, e.g. ``10.20.30.100:4444``.
 
 >>> from pycomm3 import CIPDriver
 >>> with CIPDriver('10.20.30.100') as drive:

docs/releases.rst

@@ -2,6 +2,14 @@
 Release History
 ===============
 
+1.2.10
+======
+
+CIPDriver
+---------
+- |:sparkles:| support port customization in the connection path
+- |:sparkles:| support comma delimiters in the connection path
+
 1.2.9
 =====
 

pycomm3/_version.py

@@ -22,5 +22,5 @@
 # SOFTWARE.
 #
 
-__version_info__ = (1, 2, 9)
+__version_info__ = (1, 2, 10)
 __version__ = ".".join(f"{x}" for x in __version_info__)

pycomm3/cip_driver.py

@@ -112,23 +112,6 @@ class CIPDriver:
     _auto_slot_cip_path = False
 
     def __init__(self, path: str, *args, **kwargs):
-        """
-        :param path: CIP path to intended target
-
-            The path may contain 3 forms:
-
-            - IP Address Only (``10.20.30.100``) - Use for a ControlLogix PLC is in slot 0 or if connecting to a CompactLogix or Micro800 PLC.
-            - IP Address/Slot (``10.20.30.100/1``) - (ControlLogix) if PLC is not in slot 0
-            - CIP Routing Path (``1.2.3.4/backplane/2/enet/6.7.8.9/backplane/0``) - Use for more complex routing.
-
-            .. note::
-
-                Both the IP Address and IP Address/Slot options are shortcuts, they will be replaced with the
-                CIP path automatically.  The ``enet`` / ``backplane`` (or ``bp``) segments are symbols for the CIP routing
-                port numbers and will be replaced with the correct value.
-
-        """
-
         self._sequence: cycle = cycle(65535, start=1)
         self._sock: Optional[Socket] = None
         self._session: int = 0
@@ -137,13 +120,13 @@ def __init__(self, path: str, *args, **kwargs):
         self._target_is_connected: bool = False
         self._info: Dict[str, Any] = {}
         self._cip_path = path
-        ip, _path = parse_connection_path(path, self._auto_slot_cip_path)
+        ip, port, _path = parse_connection_path(path, self._auto_slot_cip_path)
 
         self._cfg = {
             "context": b"_pycomm_",
             "protocol version": b"\x01\x00",
             "rpi": 5000,
-            "port": 44818,
+            "port": port or 44818,
             "timeout": 10,
             "ip address": ip,
             "cip_path": _path,
@@ -605,15 +588,27 @@ def _receive(self):
             return reply
 
 
-def parse_connection_path(path: str, auto_slot: bool = False) -> Tuple[str, List[PortSegment]]:
+def parse_connection_path(path: str, auto_slot: bool = False) -> Tuple[str, Optional[int], List[PortSegment]]:
     """
     Parses and validates the CIP path into the destination IP and
     sequence of port/link segments.
     Returns the IP and a list of PortSegments
     """
     try:
-        path = path.replace("\\", "/")
+        path = path.replace("\\", "/").replace(",", "/")
         ip, *route = path.split("/")
+        if ':' in ip:
+            ip, port = ip.split(':')
+            try:
+                port = int(port)
+            except Exception as err:
+                raise RequestError(f'Invalid port: {port}')
+            else:
+                if 0 > port >= 65535:
+                    raise RequestError(f'Invalid port: {port}')
+
+        else:
+            port = None
 
         try:
             ipaddress.ip_address(ip)
@@ -627,7 +622,7 @@ def parse_connection_path(path: str, auto_slot: bool = False) -> Tuple[str, List
     except Exception as err:
         raise RequestError(f"Failed to parse connection path: {path}") from err
     else:
-        return ip, _path
+        return ip, port, _path
 
 
 def parse_cip_route(path: Union[str, List[str]], auto_slot: bool = False) -> List[PortSegment]:

tests/offline/test_cip_driver.py

@@ -30,21 +30,19 @@
 CONNECT_PATH = "192.168.1.100"
 
 
-_simple_path = (
-    "192.168.1.100",
-    [
-        PortSegment("bp", 0),
-    ],
-)
+_simple_path = ("192.168.1.100", None, [PortSegment("bp", 0)])
 _simple_paths = [
     "192.168.1.100/bp/0",
     "192.168.1.100/backplane/0",
     r"192.168.1.100\bp\0",
     r"192.168.1.100\backplane\0",
+    "192.168.1.100,bp,0",
+    "192.168.1.100,1,0",
 ]
 
 _route_path = (
     "192.168.1.100",
+    None,
     [
         PortSegment(port="backplane", link_address=1),
         PortSegment(port="enet", link_address="10.11.12.13"),
@@ -54,12 +52,15 @@
 _route_paths = [
     "192.168.1.100/backplane/1/enet/10.11.12.13/bp/0",
     r"192.168.1.100\backplane\1\enet\10.11.12.13\bp\0",
+    "192.168.1.100,backplane,1,enet,10.11.12.13,bp,0",
 ]
 
 path_tests = [
     *[(p, _simple_path) for p in _simple_paths],
     *[(p, _route_path) for p in _route_paths],
-    ("192.168.1.100", ("192.168.1.100", [])),
+    ("192.168.1.100", ("192.168.1.100", None, [])),
+    ('192.168.1.100:123', ('192.168.1.100', 123, [])),
+    ('192.168.1.100:123/bp,0', ('192.168.1.100', 123, _simple_path[-1]))
 ]
 
 
@@ -91,8 +92,16 @@ def test_plc_path_auto_slot(path, expected_output):
     "1.1.1.300",
     "192.168.1.100/Z",
     "bp/0",
+    "192.168.1.100/-1",
     "192.168.1.100/backplan/1",
     "192.168.1.100/backplane/1/10.11.12.13/bp/0",
+    "192.168.1.100//bp/1",
+    "192.168.1.100/",
+    "192.168.1.100,bp,1,0",
+    "192.168.1.100,",
+    "192.168.1.100:abc",
+    "192.168.1.100:/bp/0",
+    "192.168.100:-123",
 ]
 
 
@@ -101,7 +110,7 @@ def test_bad_plc_paths(path):
     with pytest.raises(
         (DataError, RequestError),
     ):
-        ip, segments = parse_connection_path(path)
+        ip, port, segments = parse_connection_path(path)
         PADDED_EPATH.encode(segments, length=True)