Plc EthernetIP Workflow Block

inference/core/workflows/core_steps/loader.py

@@ -370,6 +370,9 @@
 from inference.core.workflows.core_steps.visualizations.triangle.v1 import (
     TriangleVisualizationBlockV1,
 )
+from inference.core.workflows.core_steps.sinks.PLCethernetIP.v1 import (
+    PLCBlockV1,
+)
 from inference.core.workflows.execution_engine.entities.types import (
     BAR_CODE_DETECTION_KIND,
     BOOLEAN_KIND,
@@ -582,6 +585,7 @@ def load_blocks() -> List[Type[WorkflowBlock]]:
         EnvironmentSecretsStoreBlockV1,
         SlackNotificationBlockV1,
         TwilioSMSNotificationBlockV1,
+        PLCBlockV1,
     ]
 
 

inference/core/workflows/core_steps/sinks/PLCethernetIP/v1.py

@@ -0,0 +1,176 @@
+from typing import Dict, List, Optional, Type, Union
+from pydantic import ConfigDict, Field
+from typing_extensions import Literal
+
+import pylogix
+
+from inference.core.workflows.execution_engine.entities.base import (
+    OutputDefinition,
+    WorkflowImageData,
+    VideoMetadata,
+)
+from inference.core.workflows.execution_engine.entities.types import (
+    LIST_OF_VALUES_KIND,
+    STRING_KIND,
+    WorkflowParameterSelector,
+    Selector,
+)
+from inference.core.workflows.prototypes.block import (
+    WorkflowBlock,
+    WorkflowBlockManifest,
+)
+
+LONG_DESCRIPTION = """
+This **PLC Communication** block integrates a Roboflow Workflow with a PLC using Ethernet/IP communication.
+It can:
+- Read tags from a PLC if `mode='read'`.
+- Write tags to a PLC if `mode='write'`.
+- Perform both read and write in a single run if `mode='read_and_write'`.
+
+**Parameters depending on mode:**
+- If `mode='read'` or `mode='read_and_write'`, `tags_to_read` must be provided.
+- If `mode='write'` or `mode='read_and_write'`, `tags_to_write` must be provided.
+
+If a read or write operation fails, an error message is printed to the terminal, 
+and the corresponding entry in the output dictionary is set to a generic "ReadFailure" or "WriteFailure" message.
+"""
+
+
+class PLCBlockManifest(WorkflowBlockManifest):
+    """Manifest for a PLC communication block using Ethernet/IP.
+
+    The block can be used in one of three modes:
+    - 'read': Only reads specified tags.
+    - 'write': Only writes specified tags.
+    - 'read_and_write': Performs both reading and writing in one execution.
+
+    `tags_to_read` and `tags_to_write` are applicable depending on the mode chosen.
+    """
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "name": "PLC EthernetIP",
+            "version": "v1",
+            "short_description": "Generic PLC read/write block using pylogix over Ethernet/IP.",
+            "long_description": LONG_DESCRIPTION,
+            "license": "Apache-2.0",
+            "block_type": "sinks",
+        }
+    )
+
+    type: Literal["roboflow_core/sinks@v1"]
+
+    plc_ip: Union[str, WorkflowParameterSelector(kind=[STRING_KIND])] = Field(
+        description="IP address of the target PLC.",
+        examples=["192.168.1.10"]
+    )
+
+    mode: Literal["read", "write", "read_and_write"] = Field(
+        description="Mode of operation: 'read', 'write', or 'read_and_write'.",
+        examples=["read", "write", "read_and_write"]
+    )
+
+    tags_to_read: Union[List[str], WorkflowParameterSelector(kind=[LIST_OF_VALUES_KIND])] = Field(
+        default=[],
+        description="List of PLC tag names to read. Applicable if mode='read' or mode='read_and_write'.",
+        examples=[["camera_msg", "sku_number"]]
+    )
+
+    tags_to_write: Union[Dict[str, Union[int, float, str]], WorkflowParameterSelector(kind=[LIST_OF_VALUES_KIND])] = Field(
+        default={},
+        description="Dictionary of tags and the values to write. Applicable if mode='write' or mode='read_and_write'.",
+        examples=[{"camera_fault": True, "defect_count": 5}]
+    )
+
+    depends_on: Selector() = Field(
+        description="Reference to the step output this block depends on.",
+        examples=["$steps.some_previous_step"]
+    )
+
+    @classmethod
+    def describe_outputs(cls) -> List[OutputDefinition]:
+        return [
+            OutputDefinition(
+                name="plc_results",
+                kind=[LIST_OF_VALUES_KIND],
+            ),
+        ]
+
+    @classmethod
+    def get_execution_engine_compatibility(cls) -> Optional[str]:
+        return ">=1.0.0,<2.0.0"
+
+
+class PLCBlockV1(WorkflowBlock):
+    """A PLC communication workflow block using Ethernet/IP and pylogix.
+
+    Depending on the selected mode:
+    - 'read': Reads specified tags.
+    - 'write': Writes provided values to specified tags.
+    - 'read_and_write': Reads and writes in one go.
+
+    In case of failures, errors are printed to terminal and the corresponding tag entry in the output is set to "ReadFailure" or "WriteFailure".
+    """
+
+    @classmethod
+    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
+        return PLCBlockManifest
+
+    def run(
+        self,
+        plc_ip: str,
+        mode: str,
+        tags_to_read: List[str],
+        tags_to_write: Dict[str, Union[int, float, str]],
+        depends_on: any,
+        image: Optional[WorkflowImageData] = None,
+        metadata: Optional[VideoMetadata] = None,
+    ) -> dict:
+        """Run PLC read/write operations using pylogix over Ethernet/IP.
+
+        Args:
+            plc_ip (str): PLC IP address.
+            mode (str): 'read', 'write', or 'read_and_write'.
+            tags_to_read (List[str]): Tags to read if applicable.
+            tags_to_write (Dict[str, Union[int, float, str]]): Tags to write if applicable.
+            depends_on (any): The step output this block depends on.
+            image (Optional[WorkflowImageData]): Not required for this block.
+            metadata (Optional[VideoMetadata]): Not required for this block.
+
+        Returns:
+            dict: A dictionary with `plc_results` as a list containing one dictionary. That dictionary has 'read' and/or 'write' keys.
+        """
+        read_results = {}
+        write_results = {}
+
+        with pylogix.PLC() as comm:
+            comm.IPAddress = plc_ip
+
+            # If mode involves reading
+            if mode in ["read", "read_and_write"]:
+                for tag in tags_to_read:
+                    response = comm.Read(tag)
+                    if response.Status == "Success":
+                        read_results[tag] = response.Value
+                    else:
+                        print(f"Error reading tag '{tag}': {response.Status}")
+                        read_results[tag] = "ReadFailure"
+
+            # If mode involves writing
+            if mode in ["write", "read_and_write"]:
+                for tag, value in tags_to_write.items():
+                    response = comm.Write(tag, value)
+                    if response.Status == "Success":
+                        write_results[tag] = "WriteSuccess"
+                    else:
+                        print(f"Error writing tag '{tag}' with value '{value}': {response.Status}")
+                        write_results[tag] = "WriteFailure"
+
+        plc_output = {}
+        if read_results:
+            plc_output["read"] = read_results
+        if write_results:
+            plc_output["write"] = write_results
+
+        # Return as a list of dicts for the 'plc_results' key
+        return {"plc_results": [plc_output]}

requirements/_requirements.txt

@@ -37,3 +37,4 @@ tokenizers>=0.19.0,<=0.20.3
 slack-sdk~=3.33.4
 twilio~=9.3.7
 httpx>=0.25.1,<0.28.0  # must be pinned as bc in 0.28.0 is causing Anthropics to fail
+pylogix==1.0.5