feat(plc4py): Added Generic for PlcConfiguration in PlcConnection

sandbox/plc4py/plc4py/api/PlcConnection.py

@@ -17,8 +17,8 @@
 # under the License.
 #
 import asyncio
-from abc import abstractmethod
-from typing import Awaitable
+from abc import abstractmethod, ABC
+from typing import Awaitable, Generic, TypeVar, Type
 
 from plc4py.api.messages.PlcResponse import PlcResponse, PlcReadResponse
 from plc4py.api.messages.PlcRequest import ReadRequestBuilder, PlcRequest
@@ -27,12 +27,14 @@
 from plc4py.spi.configuration.PlcConfiguration import PlcConfiguration
 from plc4py.utils.GenericTypes import GenericGenerator
 
+T = TypeVar("T", bound=PlcConfiguration)
 
-class PlcConnection(GenericGenerator):
 
-    def __init__(self, url: str):
+class PlcConnection(GenericGenerator, Generic[T], ABC):
+
+    def __init__(self, url: str, config_class: Type[T]):
         self.url = url
-        self._configuration = PlcConfiguration(url)
+        self._configuration = config_class(url)
 
     @abstractmethod
     def connect(self) -> None:

sandbox/plc4py/plc4py/drivers/mock/MockConnection.py

@@ -37,6 +37,7 @@
 from plc4py.api.messages.PlcResponse import PlcReadResponse, PlcResponse
 from plc4py.api.value.PlcValue import PlcResponseCode, PlcValue
 from plc4py.drivers.PlcDriverLoader import PlcDriverLoader
+from plc4py.spi.configuration.PlcConfiguration import PlcConfiguration
 from plc4py.spi.messages.PlcReader import PlcReader
 from plc4py.spi.messages.utils.ResponseItem import ResponseItem
 from plc4py.spi.values.PlcBOOL import PlcBOOL
@@ -92,7 +93,7 @@ def read(self, field: str) -> list[ResponseItem[PlcValue]]:
 
 
 @dataclass
-class MockConnection(PlcConnection, PlcReader):
+class MockConnection(PlcConnection[PlcConfiguration], PlcReader):
     _is_connected: bool = False
     device: MockDevice = field(default_factory=lambda: MockDevice())
 

sandbox/plc4py/plc4py/drivers/modbus/ModbusConfiguration.py

@@ -0,0 +1,31 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+from plc4py.spi.configuration.PlcConfiguration import PlcConfiguration
+
+
+class ModbusConfiguration(PlcConfiguration):
+
+    def __init__(self, url):
+        super().__init__(url)
+
+        if self.transport is None:
+            self.transport = "tcp"
+
+        if self.port is None:
+            self.port = 502

sandbox/plc4py/plc4py/drivers/modbus/ModbusConnection.py

@@ -26,16 +26,17 @@
 from plc4py.api.messages.PlcResponse import PlcResponse
 from plc4py.api.messages.PlcRequest import ReadRequestBuilder
 from plc4py.drivers.PlcDriverLoader import PlcDriverLoader
+from plc4py.drivers.modbus.ModbusConfiguration import ModbusConfiguration
 from plc4py.drivers.modbus.ModbusProtocol import ModbusProtocol
 from plc4py.spi.transport.TCPTransport import TCPTransport
 
 
-class ModbusConnection(PlcConnection):
+class ModbusConnection(PlcConnection[ModbusConfiguration]):
     """A hook implementation namespace."""
     _transport: TCPTransport = None
 
     def __init__(self, url: str):
-        super().__init__(url)
+        super().__init__(url, ModbusConfiguration)
 
     async def connect(self):
         """

sandbox/plc4py/tests/unit/plc4py/drivers/modbus/test_modbus_connection.py

@@ -22,6 +22,6 @@
 
 async def test_plc_driver_modbus_connect():
     driver_manager = PlcDriverManager()
-    with driver_manager.connection("modbus:tcp://127.0.0.1:502") as connection:
+    with driver_manager.connection("modbus://127.0.0.1:502") as connection:
         await connection.connect()
         assert connection.is_connected()