### InstanceIDAttributes Usage Example Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Demonstrates iterating through instance list results and extracting tag names using InstanceIDAttributes. ```python reply = await conn._get_instance_list_subset(user_defined=True) for instance in reply: attr = InstanceIDAttributes(instance.data) print(f"Variable: {attr.tag_name()}") ``` -------------------------------- ### Example Usage of MonitoredVariable Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Demonstrates setting up a MonitoredVariable, binding a callback function to its value changes, and how the variable is refreshed automatically. ```python def on_value_change(): print(f"Value changed to {monitored_var.value}") dispatcher = NSeriesThreadDispatcher('192.168.250.9') monitored = MonitoredVariable(dispatcher, 'TestInt', refresh_time=0.1) monitored.bind_to_value(on_value_change) # Value will be refreshed every 0.1 seconds # on_value_change will be called when it changes ``` -------------------------------- ### Type Compatibility Examples Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/types.md Demonstrates writing and reading various Python data types to and from a CIP controller using the NSeries connection. ```python from aphyt.omron import NSeries import datetime with NSeries('192.168.250.9') as conn: # Numeric types conn.write_variable('MyBool', True) conn.write_variable('MyInt', 42) conn.write_variable('MyLongInt', 1000000) conn.write_variable('MyFloat', 3.14159) # String conn.write_variable('MyString', 'Test Message') # Array conn.write_variable('MyArray', [1, 2, 3, 4, 5]) # Structure (if UDT defined) struct_data = { 'X': 100.0, 'Y': 200.0, 'Z': 300.0 } conn.write_variable('Position', struct_data) # DateTime now = datetime.datetime.now(datetime.timezone.utc) conn.write_variable('CurrentTime', now) # Reading back bool_val = conn.read_variable('MyBool') int_val = conn.read_variable('MyInt') array_val = conn.read_variable('MyArray') struct_val = conn.read_variable('Position') # Access structure members x = conn.read_variable('Position.X') ``` -------------------------------- ### Example Usage with AsyncNSeries and Load Dictionary Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Demonstrates using the AsyncNSeries context manager to connect, load a variable dictionary if present, and read a variable. ```python async with AsyncNSeries('192.168.250.9') as conn: # Load cached dictionary, or fetch and cache if not present await conn.load_dictionary_file_if_present('variables.pkl') value = await conn.read_variable('TestInt') ``` -------------------------------- ### NSeries Constructor and Connection Examples Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Demonstrates how to instantiate the NSeries class using a context manager for automatic connection handling or by manually managing the connection lifecycle. ```python import time from aphyt import omron # Context manager (recommended) with omron.NSeries('192.168.250.9') as eip_conn: value = eip_conn.read_variable('TestInt') print(value) # Manual connection management conn = omron.NSeries() conn.connect_explicit('192.168.250.9') try: value = conn.read_variable('TestInt') finally: conn.close_explicit() ``` -------------------------------- ### Extend CIP Reply for Get Instance List Ex2 Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt Extends the CIP Reply from the Get Instance List Ex2 service to the Tag Name Server 0x6A, adding descriptive properties for easier data interpretation. ```python class GetInstanceListEx2Reply: """ Extend the CIP Reply from the Get Instance List Ex2 service to Tag Name Server 0x6A to add descriptive properties """ def __init__(self, reply_bytes: bytes): super().__init__(reply_bytes=reply_bytes) @property def data_length(self): return struct.unpack(" str: return str(self.reply_data[9:9+self.tag_name_length]) @staticmethod def command_specific_data_from_eip_message_bytes(eip_message_bytes: bytes): """ :param eip_message_bytes: :return: """ eip_message = EIPMessage() eip_message.from_bytes(eip_message_bytes) return EIP.command_specific_data_from_eip_message(eip_message) ``` -------------------------------- ### Low-Level Communication Example Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-cip.md Demonstrates how to perform low-level CIP communication using `AsyncEIPConnectedCIPDispatcher`. This includes creating a CIP request, executing it, and handling potential exceptions. ```APIDOC ## Low-Level Communication Example ```python from aphyt.cip.cip import CIPRequest, CIPReply, CIPException from aphyt.eip import AsyncEIPConnectedCIPDispatcher async def low_level_example(): dispatcher = AsyncEIPConnectedCIPDispatcher() await dispatcher.connect_explicit('192.168.1.100') # Create a CIP read request read_request = CIPRequest( b'\x4c', # Read Tag Service b'\x20\x02\x24\x01', # Simple path ) try: reply = await dispatcher.execute_cip_command(read_request) print(f"Status: {reply.general_status}") print(f"Data: {reply.reply_data}") except CIPException as e: print(f"Error: {e.get_message()}") finally: await dispatcher.close_explicit() ``` ``` -------------------------------- ### Get Instance List Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Retrieves a list of instances, specifically for Omron-specific service 0x5f. Use this to discover available instances. ```python async def get_instance_list(self, starting_instance: int, quantity: int, user_defined: bool) -> CIPReply ``` -------------------------------- ### AsyncNSeries High-Level and EIPMessage Low-Level Example Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Demonstrates connecting to an Omron device using the high-level AsyncNSeries interface for reading and writing variables, and shows the creation of a low-level EIPMessage. ```python import asyncio from aphyt.eip import EIPMessage from aphyt.omron import AsyncNSeries async def example(): # Using the high-level NSeries interface async with AsyncNSeries('192.168.1.100') as conn: # Connection is handled automatically value = await conn.read_variable('TestVar') await conn.write_variable('TestVar', 42) # Low-level EIP message example msg = EIPMessage(b'\x6f\x00') msg.set_context(1) # Send via dispatcher... asyncio.run(example()) ``` -------------------------------- ### Safe Structure Access Example Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/index.md Demonstrates how to safely access members of a structure read from a connection, including handling cases where the structure might not be found. ```python from aphyt.cip.cip import CIPException try: struct_val = conn.read_variable('MyStruct') x = struct_val['X'].value() y = struct_val.members['Y'].value() except CIPException as e: if e.status == b'\x04': print("Structure not found or not published") ``` -------------------------------- ### Low-Level CIP Communication Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-cip.md Demonstrates establishing an explicit connection, executing a low-level CIP read command, and handling potential exceptions. This example uses the AsyncEIPConnectedCIPDispatcher for asynchronous operations. ```python from aphyt.cip.cip import CIPRequest, CIPReply, CIPException from aphyt.eip import AsyncEIPConnectedCIPDispatcher async def low_level_example(): dispatcher = AsyncEIPConnectedCIPDispatcher() await dispatcher.connect_explicit('192.168.1.100') # Create a CIP read request read_request = CIPRequest( b'\x4c', # Read Tag Service b'\x20\x02\x24\x01', # Simple path ) try: reply = await dispatcher.execute_cip_command(read_request) print(f"Status: {reply.general_status}") print(f"Data: {reply.reply_data}") except CIPException as e: print(f"Error: {e.get_message()}") finally: await dispatcher.close_explicit() ``` -------------------------------- ### Persist Session and Start Keep-Alive Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Configure the NSeriesThreadDispatcher to persist the connection session and start a keep-alive mechanism. This helps prevent frequent connection drops. ```python dispatcher = NSeriesThreadDispatcher('192.168.250.9') dispatcher.connection_status.persist_session = True dispatcher.start_keep_alive(keep_alive_time=0.5) ``` -------------------------------- ### Instantiate and Use NSeriesThreadDispatcher Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md This example demonstrates how to create an instance of NSeriesThreadDispatcher with custom connection parameters and then read a variable. The dispatcher handles connection management and retries automatically. ```python with NSeriesThreadDispatcher('192.168.250.9', connection_timeout=5.0, retry_time=2.0, max_attempts=5) as dispatcher: value = dispatcher.read_variable('TestInt') ``` -------------------------------- ### Get List of System Variables Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Returns a list of system-defined variables on the controller. These are typically used for internal controller functions. ```python async def system_variable_list(self) -> list[str] ``` -------------------------------- ### Concurrent Reads with AsyncNSeries Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md This example shows how to perform concurrent read operations on multiple variables using the AsyncNSeries client. It utilizes asyncio.gather for parallel execution. ```python import asyncio from aphyt.omron import AsyncNSeries async def example(): async with AsyncNSeries('192.168.250.9') as conn: # Concurrent reads values = await asyncio.gather( conn.read_variable('Var1'), conn.read_variable('Var2'), conn.read_variable('Var3'), ) return values result = asyncio.run(example()) ``` -------------------------------- ### VariableObjectReply Usage Example Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Demonstrates how to use VariableObjectReply to parse a response and access variable properties like size and data type. ```python request_path = variable_request_path_segment('MyVariable') response = await dispatcher.get_attribute_all_service(request_path) var_obj = VariableObjectReply(response.bytes) print(f"Size: {var_obj.size}") print(f"Type: {var_obj.cip_data_type.hex()}") ``` -------------------------------- ### VariableTypeObjectReply Usage Example Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Shows how to retrieve a variable type object using its instance ID and print its name and member count. ```python # Get structure definition var_type_obj = await conn._get_variable_type_object(instance_id=5) print(f"Type Name: {var_type_obj.variable_type_name.decode('utf-8')}") print(f"Members: {var_type_obj.number_of_members}") ``` -------------------------------- ### CIP Dispatcher System Variables Property Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Provides access to the dictionary of system variables (names starting with '_'). Use this to inspect internal system states. ```python @property def system_variables(self) -> dict ``` -------------------------------- ### Get Attribute All Service Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Retrieves all attributes of a CIP object using service 0x01. Useful for inspecting an object's complete state. ```python async def get_attribute_all_service(self, request_path: bytes) -> CIPReply ``` -------------------------------- ### Error Handling for CIP Exceptions Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/index.md Provides an example of how to catch and handle specific CIP exceptions, such as 'Variable not found' or 'PLC resources exhausted', when interacting with the PLC. It also includes handling generic network socket errors. ```python from aphyt.cip.cip import CIPException import socket try: with NSeries('192.168.250.9') as conn: value = conn.read_variable('SomeVar') except CIPException as e: if e.status == b'\x04': print("Variable not found") elif e.status == b'\x02': print("PLC resources exhausted - retry later") else: print(f"Error: {e.get_message()}") except socket.error as e: print(f"Network error: {e}") ``` -------------------------------- ### Update Dictionary and Check Variable Existence Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md This example shows how to update the variable dictionary to discover all available variables and then check if a specific variable exists before attempting to read it. This helps resolve 'Variable Not Found' errors. ```python # Update dictionary to discover all variables dispatcher.update_variable_dictionary() available_vars = dispatcher.variable_list() print("Available variables:", available_vars) # Check if variable exists if 'TestVar' in available_vars: value = dispatcher.read_variable('TestVar') ``` -------------------------------- ### Configure Monitored Variable Refresh Time Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Sets a longer refresh interval for monitored variables to reduce network traffic. This example sets the refresh time to 1.0 second. ```python from aphyt.omron import NSeriesThreadDispatcher, MonitoredVariable dispatcher = NSeriesThreadDispatcher('192.168.250.9') # Option 2: Use monitored variables with longer refresh intervals monitor = MonitoredVariable(dispatcher, 'TestInt', refresh_time=1.0) # Refreshes every 1 second instead of 50ms ``` -------------------------------- ### Encode Tag Path Segment Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-cip.md Illustrates the structure of a CIP request path segment for a logical tag. The example shows how to encode the tag name with appropriate length and padding. ```python # For tag "TestInt", the path would be: # 0x91 (logical tag segment) # 0x07 (length in words = 7 ASCII chars / 2 rounded up) # b'TestInt\0' (7 bytes padded to 8) # 0x00 (padding word) ``` -------------------------------- ### Get Variable Object Asynchronously Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Asynchronously retrieves the variable object for a specific instance ID. Use this to get detailed information about a variable. ```python async def _get_variable_object(self, instance_id: int) -> VariableObjectReply ``` -------------------------------- ### Set up Callbacks for Value Monitoring Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md This snippet demonstrates how to bind callback functions to monitor changes in device values. It shows how to access the current value and clean up resources. ```python for name, monitor in monitors.items(): monitor.bind_to_value(lambda: print(f"{name}: {monitor.value}")) # Monitoring happens automatically # To access values: temp = monitors['Temperature'].value pressure = monitors['Pressure'].value # Clean up for monitor in monitors.values(): monitor.cancel() ``` -------------------------------- ### NSeriesThreadDispatcher.system_variable_list Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Retrieves a list of system variable names (those starting with an underscore) from the controller. This method is part of the synchronous NSeriesThreadDispatcher. ```APIDOC ## system_variable_list Returns system variables. **Returns:** List of system variable names ``` -------------------------------- ### NSeriesThreadDispatcher.user_variable_list Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Retrieves a list of user-defined variable names (those not starting with an underscore) from the controller. This method is part of the synchronous NSeriesThreadDispatcher. ```APIDOC ## user_variable_list Returns user-defined variables. **Returns:** List of user variable names ``` -------------------------------- ### Get List of User-Defined Variables Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Returns a list of variables that are defined by the user on the controller. This helps in filtering for custom variables. ```python async def user_variable_list(self) -> list[str] ``` -------------------------------- ### Get List of Discovered Variables Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Returns a list of all variables that have been discovered on the controller. This typically reflects the current variable dictionary. ```python async def variable_list(self) -> list[str] ``` -------------------------------- ### NSeries Configuration with Host Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Instantiates NSeries and automatically connects to the specified host. ```python # With host (auto-connect) conn = NSeries('192.168.250.9') ``` -------------------------------- ### VariableNameAttributeAllReply Class Definition Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Defines the VariableNameAttributeAllReply class for parsing CIP replies from Get Attribute All on the Tag Name Server. ```python class VariableNameAttributeAllReply(CIPReply): """CIP Reply from Tag Name Server Get Attribute All""" ``` -------------------------------- ### EIP Class Initialization and Connection Management Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt Initializes the EIP class, setting up default values for connection parameters and session handles. Includes methods for explicit connection and disconnection. ```python class EIP: """ EIP is an encapsulation protocol for CIP (common industrial protocol) messages """ explicit_message_port = 44818 null_address_item = b' oo oo' cip_handle = b' oo oo oo oo' def __init__(self): self.explicit_message_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.session_handle_id = b' oo oo oo oo oo oo oo oo' self.is_connected_explicit = False self.has_session_handle = False self.variables = {} self.user_variables = {} self.system_variables = {} self.BUFFER_SIZE = 4096 def __del__(self): self.close_explicit() def connect_explicit(self, host): """ :param host: """ try: self.explicit_message_socket.connect((host, self.explicit_message_port)) self.is_connected_explicit = True except socket.error as err: if self.explicit_message_socket: self.explicit_message_socket.close() self.is_connected_explicit = False def close_explicit(self): """ :return: """ self.is_connected_explicit = False self.has_session_handle = False if self.explicit_message_socket: self.explicit_message_socket.close() ``` -------------------------------- ### VariableObjectReply Class Definition Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Defines the VariableObjectReply class for parsing CIP replies from the Get Attribute All service on Variable Object. ```python class VariableObjectReply(CIPReply): """CIP Reply adding descriptive properties to Variable Object""" ``` -------------------------------- ### Minimal NSeries Configuration Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Instantiates NSeries with default settings and explicitly connects to a controller. ```python import socket from aphyt.omron import NSeries # Minimal configuration conn = NSeries() conn.connect_explicit('192.168.250.9') ``` -------------------------------- ### Basic Read/Write with NSeries Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/index.md Demonstrates synchronous reading and writing of PLC variables using the NSeries context manager. Ensure the PLC IP address is correct and the variable names exist on the controller. ```python from aphyt.omron import NSeries with NSeries('192.168.250.9') as conn: # Read integer value = conn.read_variable('TestInt') # Write integer conn.write_variable('TestInt', 42) # Read with verification verified_value = conn.read_variable('TestInt') ``` -------------------------------- ### VariableTypeObjectReply Class Definition Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Defines the VariableTypeObjectReply class for parsing CIP replies from the Get Attribute All service on Variable Type Object. ```python class VariableTypeObjectReply(CIPReply): """CIP Reply for Variable Type Object (structure/UDT definitions)""" ``` -------------------------------- ### Basic PLC Read/Write Operations Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/README.md Demonstrates basic synchronous read and write operations to PLC variables using the NSeries class. Ensure the PLC is accessible at the specified IP address. ```python from aphyt.omron import NSeries with NSeries('192.168.250.9') as conn: # Read a variable value = conn.read_variable('TestInt') # Write a variable conn.write_variable('TestInt', 42) # Read with verification conn.verified_write_variable('TestInt', 42) ``` -------------------------------- ### connect_explicit Method Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Establishes a connection to the target and registers an EIP session. Supports a connection timeout for establishment. ```python async def connect_explicit(self, host: str, connection_timeout: float = None) -> None ``` -------------------------------- ### list_services Method Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Sends the ListServices command to determine which services the target supports. The host parameter specifies the target host. ```python async def list_services(self, host: str = '') -> bytes ``` -------------------------------- ### VariableTypeObjectReply Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Parses CIP replies from the Get Attribute All service on Variable Type Object (Class Code 0x6C). Used for structure and UDT definitions. ```APIDOC ## Class VariableTypeObjectReply ### Description Parses CIP replies from the Get Attribute All service on Variable Type Object (Class Code 0x6C). Used for structure and UDT definitions. ### Properties | Property | Type | Description | |----------|------|-------------| | size_in_memory | int | Size of type in memory | | cip_data_type | bytes | CIP data type code | | cip_data_type_of_array | bytes | Array element type code | | array_dimension | int | Array dimensions | | number_of_elements | list[int] | Array size per dimension | | number_of_members | int | Number of structure members | | crc_code | bytes | CRC-16 of structure definition | | variable_type_name | bytes | Name of the type | | variable_type_name_length | int | Length of type name | | next_instance_id | bytes | Instance ID of next member | | nesting_variable_type_instance_id | bytes | Instance ID for nested definitions | | start_array_elements | list[int] | Array starting indices | ### Usage ```python # Get structure definition var_type_obj = await conn._get_variable_type_object(instance_id=5) print(f"Type Name: {var_type_obj.variable_type_name.decode('utf-8')}") print(f"Members: {var_type_obj.number_of_members}") ``` ``` -------------------------------- ### Production Configuration Pattern Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md A reliable configuration pattern for production environments, emphasizing long connection timeouts, retry intervals, and session persistence for stability. ```python from aphyt.omron import NSeriesThreadDispatcher import logging logging.basicConfig(level=logging.WARNING) dispatcher = NSeriesThreadDispatcher( host='192.168.250.9', connection_timeout=10.0, # Long timeout for reliability retry_time=5.0, # Wait 5 seconds between retries max_attempts=None # Never give up ) # Enable session persistence dispatcher.connection_status.persist_session = True try: dispatcher.update_variable_dictionary() value = dispatcher.read_variable('TestInt') finally: dispatcher.close_explicit() ``` -------------------------------- ### MonitoredVariable Configuration with Observers Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Sets up a MonitoredVariable with an initial list of observer callbacks to be notified on value changes. ```python # With callbacks def on_value_change(): print(f"Value changed to {monitor.value}") monitor = MonitoredVariable( dispatcher, 'TestInt', refresh_time=0.05, observers=[on_value_change] ) ``` -------------------------------- ### VariableObjectReply Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Parses CIP replies from the Get Attribute All service on Variable Object (Class Code 0x6B). It provides descriptive properties for variable attributes. ```APIDOC ## Class VariableObjectReply ### Description Parses CIP replies from the Get Attribute All service on Variable Object (Class Code 0x6B). ### Properties | Property | Type | Description | |----------|------|-------------| | size | int | Size of variable in bytes | | cip_data_type | bytes | CIP data type code of the variable | | cip_data_type_of_array | bytes | Data type code for array elements (if array) | | array_dimension | int | Number of array dimensions (0 if not array) | | number_of_elements | list[int] | Size of each array dimension | | bit_number | int | Bit number (for boolean arrays) | | variable_type_instance_id | bytes | Instance ID of variable type definition | | start_array_elements | list[int] | Starting index for each dimension | ### Usage ```python request_path = variable_request_path_segment('MyVariable') response = await dispatcher.get_attribute_all_service(request_path) var_obj = VariableObjectReply(response.bytes) print(f"Size: {var_obj.size}") print(f"Type: {var_obj.cip_data_type.hex()}") ``` ``` -------------------------------- ### connect_explicit Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Establishes an explicit Ethernet/IP connection to the controller and registers a session. ```APIDOC ## connect_explicit ### Description Establishes an explicit Ethernet/IP connection to the controller and registers a session. ### Method connect_explicit ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **host** (str) - Required - IP address of the controller - **connection_timeout** (float) - Optional - Timeout in seconds for establishing connection ### Raises socket.error if connection fails ### Example ```python conn = omron.NSeries() conn.connect_explicit('192.168.250.9', connection_timeout=5.0) ``` ``` -------------------------------- ### Development Configuration Pattern Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md A configuration pattern for development, prioritizing fast feedback with shorter timeouts and automatic connection/disconnection handling. ```python from aphyt.omron import NSeries with NSeries('192.168.250.9', timeout=5.0) as conn: # Auto-connect on entry, auto-close on exit value = conn.read_variable('TestInt') conn.write_variable('TestInt', 42) ``` -------------------------------- ### Using Context Managers for Connection Handling Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/errors.md Always use context managers (with statement) for managing connections to ensure they are properly closed, even if errors occur. Avoid manual connection opening and closing. ```python # Good with NSeries('192.168.250.9') as conn: value = conn.read_variable('Var') # Avoid conn = NSeries('192.168.250.9') value = conn.read_variable('Var') # May not close properly if error occurs ``` -------------------------------- ### MonitoredVariable.value Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Provides access to the monitored variable's current value. It can be used to get the value or to set a new value, which will trigger a verified write operation. ```APIDOC ## value (property) ### Description Get or set the monitored variable value. Setting the value performs a `verified_write_variable`. ### Method `@property def value(self) -> Any` `@value.setter def value(self, value: Any) -> None` ### Parameters #### Path Parameters (for setter) - **value** (Any) - Required - The new value to set for the monitored variable. ``` -------------------------------- ### Get Member Instance Asynchronously Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Asynchronously retrieves a CIP datatype instance for a structure member, given its instance ID. Useful for navigating complex structures. ```python async def _get_member_instance(self, member_instance_id: int, variable_type_object_reply: VariableTypeObjectReply = None ) -> CIPDataType ``` -------------------------------- ### Low-Level CIP Read Request Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Demonstrates constructing and executing a raw CIP read request at a low level. This involves manually building the request path and command, then parsing the reply. ```python async def low_level_read(): from aphyt.cip.cip import CIPRequest from aphyt.omron.n_series import variable_request_path_segment async with AsyncNSeries('192.168.250.9') as conn: # Build path for 'TestInt' path = variable_request_path_segment('TestInt') # Create read request request = CIPRequest(b'\x4c', path, b'') # Execute reply = await conn.connected_cip_dispatcher.execute_cip_command(request) # Parse reply print(f"Status: {reply.general_status.hex()}") print(f"Data: {reply.reply_data.hex()}") ``` -------------------------------- ### Get Instance from Variable Name Asynchronously Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Asynchronously discovers and returns the correct CIP datatype instance for a given variable name. This simplifies variable access. ```python async def _get_instance_from_variable_name(self, variable_name: str) -> CIPDataType ``` -------------------------------- ### Recommended Import Pattern Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/index.md Import high-level classes for Omron devices, including the main synchronous class, its asynchronous counterpart, a thread-safe dispatcher, and monitored variable functionality. ```python from aphyt.omron import NSeries, NSeriesThreadDispatcher, MonitoredVariable ``` -------------------------------- ### VariableNameAttributeAllReply Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Parses CIP replies from Get Attribute All on Tag Name Server. Extracts CIP data type, instance ID, and variable type ID. ```APIDOC ## Class VariableNameAttributeAllReply ### Description Parses CIP replies from Get Attribute All on Tag Name Server. Extracts CIP data type, instance ID, and variable type ID. ### Properties | Property | Type | Description | |----------|------|-------------| | cip_data_type | bytes | CIP data type code | | instance_id | bytes | Instance ID of the variable | | variable_type_id | bytes | Instance ID of variable type | ``` -------------------------------- ### Create and Serialize EIPMessage Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Instantiate an EIPMessage with a command code and serialize it into bytes for transmission. ```python from aphyt.eip import EIPMessage msg = EIPMessage(b'\x6f\x00') packet = msg.bytes() ``` -------------------------------- ### NSeries Configuration with Timeout Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Instantiates NSeries with a specified connection timeout. ```python # With timeout conn = NSeries('192.168.250.9', timeout=10.0) ``` -------------------------------- ### Get or Set Monitored Variable Value Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Provides property access to the monitored variable's value. Setting the value triggers a verified write operation to the controller. ```python @property def value(self) -> Any @value.setter def value(self, value: Any) -> None ``` -------------------------------- ### list_identity Method Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Sends the ListIdentity command to discover EIP devices on the network. The host parameter specifies the target host. ```python async def list_identity(self, host: str = '') -> bytes ``` -------------------------------- ### Abstract send_command Method Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Abstract method to send an EIP command and receive a response. Must be implemented by subclasses. ```python async def send_command(self, eip_command: EIPMessage, host: str) -> EIPMessage ``` -------------------------------- ### List Supported Services Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt Discovers the services supported by an Ethernet/IP target. Sends a List Services (0x0400) command and returns the raw command data from the response. ```python def list_services(self): """ Find which services a target supports :return: """ eip_message = EIPMessage(b'\x04\x00') return self.send_command(eip_message).command_data ``` -------------------------------- ### Get Variable Type Object Asynchronously Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Asynchronously retrieves the variable type object, which defines the structure, for a given instance ID. Essential for understanding complex data types. ```python async def _get_variable_type_object(self, instance_id: int) -> VariableTypeObjectReply ``` -------------------------------- ### Low-Level CIP Dispatcher Methods Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Methods for direct interaction with the CIP protocol, enabling sending commands and receiving replies. ```APIDOC ## Low-Level CIP Dispatcher Methods ### `execute_cip_command` Sends a CIP command and receives the reply. #### Signature ```python async def execute_cip_command(self, cip_request: CIPRequest) -> CIPReply ``` #### Example ```python request = CIPRequest(b'\x4c', path_bytes, b'') reply = await dispatcher.execute_cip_command(request) ``` ### `read_tag_service` Reads a tag using the read tag service (0x4c). #### Signature ```python async def read_tag_service(self, request_path: bytes) -> CIPReply ``` ### `write_tag_service` Writes a tag using the write tag service (0x4d). #### Signature ```python async def write_tag_service(self, request_path: bytes, data: CIPCommonFormat) -> CIPReply ``` ### `get_attribute_all_service` Gets all attributes from an object using service 0x01. #### Signature ```python async def get_attribute_all_service(self, request_path: bytes) -> CIPReply ``` ### `get_instance_list` Gets a list of instances (Omron-specific service 0x5f). #### Signature ```python async def get_instance_list(self, starting_instance: int, quantity: int, user_defined: bool) -> CIPReply ``` ``` -------------------------------- ### Service Discovery Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt APIs for discovering services and device information. ```APIDOC ## GET /list/services ### Description Find which services a target supports. ### Method GET ### Endpoint /list/services ### Response #### Success Response (200) - **command_data** (bytes) - Data containing the list of supported services. ### Response Example ```json { "command_data": "..." } ``` ## GET /list/identity ### Description Used by an originator to locate possible targets. ### Method GET ### Endpoint /list/identity ### Response #### Success Response (200) - **command_data** (bytes) - Data containing the identity information of the target. ### Response Example ```json { "command_data": "..." } ``` ## GET /list/interfaces ### Description Used by an originator to identify possible non-CIP interfaces on the target. ### Method GET ### Endpoint /list/interfaces ### Response #### Success Response (200) - **command_data** (bytes) - Data containing the list of non-CIP interfaces. ### Response Example ```json { "command_data": "..." } ``` ``` -------------------------------- ### Load Dictionary from File or Fetch from Controller Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Loads the variable dictionary from a file if it exists. Otherwise, it fetches the dictionary from the controller and saves it to the specified file. ```python async def load_dictionary_file_if_present(self, file_name: str) -> None ``` -------------------------------- ### EIPConnectionStatus Session Persistence and Keep-Alive Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Shows how to enable automatic session persistence and the keep-alive mechanism for EIP connections. ```python # Enable automatic session persistence dispatcher.connection_status.persist_session = True # Enable keep-alive mechanism dispatcher.connection_status.keep_alive = True ``` -------------------------------- ### Get Number of Variables from Tag Name Server Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt Determines the total number of variables available on the Tag Name Server. It sends a specific request to the server and parses the reply to extract the count. ```python def _get_number_of_variables(self) -> int: """ Find number of variables from Tag Name Server :return: """ route_path = address_request_path_segment(b'\x6a', b'\x00\x00') reply = self._get_request_route_path(route_path) return int.from_bytes(reply[10:12], 'little') ``` -------------------------------- ### list_interfaces Method Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Sends the ListInterfaces command to identify non-CIP interfaces. The host parameter specifies the target host. ```python async def list_interfaces(self, host: str = '') -> bytes ``` -------------------------------- ### Get Variable List from Tag Name Server Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt Retrieves a list of all available tags (variables) from the Tag Name Server. It iterates through the number of variables, constructs route paths, and decodes tag names from responses. ```python def _get_variable_list(self): "" :return: """ tag_list = [] for tag_index in range(self._get_number_of_variables()): offset = tag_index + 1 route_path = address_request_path_segment(b'\x6a', offset.to_bytes(2, 'little')) response = self._get_request_route_path(route_path) tag = str(response[13:13 + int.from_bytes(response[12:13], 'utf-8')], 'utf-8') tag_list.append(tag) return tag_list ``` -------------------------------- ### list_services Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Sends the ListServices command to determine which services the target device supports. This can be called synchronously or asynchronously. ```APIDOC ## list_services ### Description Sends the ListServices command to determine which services the target device supports. This can be called synchronously or asynchronously. ### Method `async def` or `def` ### Parameters #### Path Parameters None #### Query Parameters - **host** (str) - Optional - The target host address. Defaults to an empty string. ### Returns - **bytes** - Response payload bytes. ``` -------------------------------- ### CommonPacketFormat Constructor Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Encapsulates data and address items in the format required by EIP. ```APIDOC ## CommonPacketFormat Constructor ### Description Encapsulates data and address items in the format required by EIP. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body * **packets** (List[DataAndAddressItem]) - Required - List of data/address items (min 0, max 2) ### Request Example ```python from aphyt.eip import CommonPacketFormat, DataAndAddressItem items = [DataAndAddressItem(type_id=b'\x00\x00', data=b'')] cpf = CommonPacketFormat(packets=items) ``` ### Response #### Success Response (200) None #### Response Example None ``` -------------------------------- ### NSeries Socket Options Configuration Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Sets socket options before establishing a connection with NSeries. ```python # Set socket options before connecting conn.message_timeout = 0.5 ``` -------------------------------- ### Establish Connection and Exchange Variables with Omron Controller Source: https://github.com/aphyt/aphytcomm/blob/master/README.md This snippet demonstrates how to establish an Ethernet/IP connection to an Omron N-Series controller and perform read/write operations on a variable named 'TestInt'. It requires the 'aphyt' library and a running Omron controller accessible at the specified IP address. ```python import time from aphyt import omron if __name__ == '__main__': with omron.NSeries('192.168.250.9') as eip_conn: for i in range(10): print(eip_conn.read_variable('TestInt')) eip_conn.write_variable('TestInt', 24) print(eip_conn.read_variable('TestInt')) eip_conn.write_variable('TestInt', 7) time.sleep(.5) ``` -------------------------------- ### Get Variable with CIP Data Type Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt Internal method to retrieve a variable's CIP data type from an Ethernet/IP device. It constructs a CIP READ_TAG_SERVICE request, sends it, and parses the response to extract and return the CIP data type object. ```python def _get_variable_with_cip_data_type(self, variable_name: str): """ ToDo Currently just reading symbolic. Add Logical Segment (Class Instance Attribute) :param variable_name: :return: """ request_path = variable_request_path_segment(variable_name) cip_message = CIPRequest(CIPService.READ_TAG_SERVICE, request_path, b'\x01\x00') data_address_item = DataAndAddressItem(DataAndAddressItem.UNCONNECTED_MESSAGE, cip_message.bytes) packets = [data_address_item] common_packet_format = CommonPacketFormat(packets) command_specific_data = CommandSpecificData(encapsulated_packet=common_packet_format.bytes()) response = self.send_rr_data(command_specific_data.bytes()).packets[1].bytes() reply_data_and_address_item = DataAndAddressItem('', b'') reply_data_and_address_item.from_bytes(response) cip_reply = CIPReply(reply_data_and_address_item.data) cip_data_type = CIPDataTypes() cip_data_type.from_bytes(cip_reply.reply_data) return cip_data_type ``` -------------------------------- ### list_interfaces Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Sends the ListInterfaces command to identify non-CIP interfaces on the target device. This can be called synchronously or asynchronously. ```APIDOC ## list_interfaces ### Description Sends the ListInterfaces command to identify non-CIP interfaces on the target device. This can be called synchronously or asynchronously. ### Method `async def` or `def` ### Parameters #### Path Parameters None #### Query Parameters - **host** (str) - Optional - The target host address. Defaults to an empty string. ### Returns - **bytes** - Response payload bytes. ``` -------------------------------- ### SimpleDataSegmentRequest bytes() Method Usage Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Shows how to create a SimpleDataSegmentRequest and append its encoded bytes to a variable request path for large data operations. ```python # Read 100 bytes starting at offset 200 segment = SimpleDataSegmentRequest(offset=200, size=100) request_path = variable_request_path_segment('LargeArray') + segment.bytes() ``` -------------------------------- ### Discover All Variables Programmatically Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Shows how to programmatically discover all variables (user-defined and system) within the controller. It updates the controller's variable dictionary and then lists variable counts and details. ```python async def list_all_variables(): async with AsyncNSeries('192.168.250.9') as conn: await conn.update_variable_dictionary() all_vars = await conn.variable_list() user_vars = await conn.user_variable_list() system_vars = await conn.system_variable_list() print(f"Total: {len(all_vars)}") print(f"User-defined: {len(user_vars)}") print(f"System: {len(system_vars)}") # Get details for each variable for var_name in all_vars[:10]: # First 10 var_obj = conn.connected_cip_dispatcher.variables[var_name] print(f"{var_name}: {var_obj.size} bytes, type {var_obj.data_type_code().hex()}") ``` -------------------------------- ### list_services Method (Synchronous) Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Synchronous wrapper for the ListServices command. ```python def list_services(self, host: str = '') -> bytes ``` -------------------------------- ### Load Variable Dictionary from File Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Loads a previously saved variable dictionary from a specified pickle file. This can speed up initialization by avoiding fetching from the controller. ```python async def load_dictionary_file(self, file_name: str) -> None ``` -------------------------------- ### Basic Try-Catch for CIP and Network Errors Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/errors.md This pattern demonstrates how to handle potential CIP protocol errors and network socket errors using try-except blocks. It includes establishing a connection, reading a variable, and printing specific error messages based on the exception type. ```python from aphyt.cip.cip import CIPException from aphyt.omron import NSeries import socket try: with NSeries('192.168.250.9') as conn: value = conn.read_variable('SomeVariable') except CIPException as e: print(f"CIP Error: {e.get_message()}") print(f"Status Code: {e.status.hex()}") except socket.error as e: print(f"Network Error: {e}") ``` -------------------------------- ### AsyncNSeries Constructor Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Initializes the asynchronous class for communicating with Omron N-Series controllers. ```APIDOC ## AsyncNSeries Constructor Initializes the asynchronous class for communicating with Omron N-Series controllers. ### Parameters #### Parameters - **host** (str) - Optional - IP address of the controller - **timeout** (float) - Optional - Connection timeout in seconds ``` -------------------------------- ### Basic NSeriesThreadDispatcher Configuration Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/configuration.md Configures NSeriesThreadDispatcher with host, connection timeout, retry time, and max attempts for auto-reconnect. ```python from aphyt.omron import NSeriesThreadDispatcher # Basic configuration with auto-reconnect dispatcher = NSeriesThreadDispatcher( host='192.168.250.9', connection_timeout=5.0, retry_time=2.0, max_attempts=5 ) ``` -------------------------------- ### Read Structure Definition Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-advanced.md Demonstrates how to retrieve and inspect the definition of a structure variable, including its members and their data types. ```python from aphyt.omron import AsyncNSeries async def inspect_structure(): async with AsyncNSeries('192.168.250.9') as conn: # Get the structure definition struct_obj = await conn._get_instance_from_variable_name('MyStruct') print(f"Structure Type: {struct_obj.variable_type_name}") print(f"Size: {struct_obj.size} bytes") print(f"Members:") for member_name, member_type in struct_obj.members.items(): print(f" {member_name}: {member_type.data_type_code().hex()}") ``` -------------------------------- ### Write and Verify Variable Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Asynchronously writes a variable to the controller and then verifies the write operation. Includes a retry mechanism for robustness. ```python async def verified_write_variable(self, variable_name: str, data: Any, retry_count: int = 2) -> None ``` -------------------------------- ### Establish Explicit Connection with NSeries Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Establishes an explicit Ethernet/IP connection to the Omron controller. Specify a connection timeout for the operation. ```python conn = omron.NSeries() conn.connect_explicit('192.168.250.9', connection_timeout=5.0) ``` -------------------------------- ### connect_explicit Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-eip.md Establishes an explicit connection to the target EIP device and registers an EIP session. This method is part of the AsyncEIPConnectedCIPDispatcher. ```APIDOC ## connect_explicit ### Description Establishes an explicit connection to the target EIP device and registers an EIP session. This method is part of the AsyncEIPConnectedCIPDispatcher. ### Method `async def` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **host** (str) - Required - The IP address or hostname of the target EIP device. - **connection_timeout** (float) - Optional - The timeout in seconds for establishing the connection. ``` -------------------------------- ### connect_explicit Source: https://github.com/aphyt/aphytcomm/blob/master/_autodocs/api-reference-nseries.md Asynchronously establishes an explicit connection to the controller. This method is used to initiate communication with the NSeries device. ```APIDOC ## connect_explicit ### Description Asynchronously establishes an explicit connection to the controller. ### Method `async def connect_explicit(self, host: str, connection_timeout: float = None) -> None` ### Parameters #### Path Parameters - **host** (str) - Required - The hostname or IP address of the controller. - **connection_timeout** (float) - Optional - The timeout in seconds for the connection attempt. ### Response No specific response body is defined for success. The method returns `None` upon successful connection. ``` -------------------------------- ### Session Management Source: https://github.com/aphyt/aphytcomm/blob/master/src/aphyt/eip/notes.txt APIs for establishing and managing sessions with the target device. ```APIDOC ## POST /register/session ### Description Used by an originator to establish a session. It is required before sending CIP messages. ### Method POST ### Endpoint /register/session ### Parameters #### Request Body - **command_data** (bytes) - Optional - Command data for session registration, defaults to b'\x01\x00\x00\x00'. ### Request Example ```json { "command_data": "01000000" } ``` ### Response #### Success Response (200) - **session_handle_id** (int) - The handle ID for the established session. #### Response Example ```json { "session_handle_id": 12345 } ``` ```