### Install pyads from Source Source: https://pyads.readthedocs.io/en/latest/_sources/installation.md.txt Cloning the repository and installing via setup.py. ```bash git clone https://github.com/stlehmann/pyads.git --recursive cd pyads python setup.py install ``` -------------------------------- ### Install pyads via Package Managers Source: https://pyads.readthedocs.io/en/latest/_sources/installation.md.txt Standard installation commands for PyPi and conda-forge. ```bash pip install pyads ``` ```bash conda install pyads ``` -------------------------------- ### Manually Build adslib.so Source: https://pyads.readthedocs.io/en/latest/_sources/installation.md.txt Commands to compile and install the shared library if automatic installation fails. ```bash cd adslib make sudo make install ``` -------------------------------- ### Run basic testserver via command line Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/testserver.rst.txt Starts the testserver with the basic handler configuration. ```bash $ python -m pyads.testserver --handler basic ``` -------------------------------- ### Verify pyads Installation Source: https://pyads.readthedocs.io/en/latest/_sources/installation.md.txt Import the module in a Python console to check for successful installation. ```python >>> import pyads ``` -------------------------------- ### Install Linux Build Dependencies Source: https://pyads.readthedocs.io/en/latest/_sources/installation.md.txt Required packages for compiling adslib.so on Ubuntu-based systems. ```bash apt update apt install -y cmake g++ ``` -------------------------------- ### Handle missing symbols Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Example of an exception raised when a symbol is not found. ```python >>> plc.read_by_name('GVL.wrong_name', pyads.PLCTYPE_BOOL) ADSError: ADSError: symbol not found (1808) ``` -------------------------------- ### Example of SAdsSymbolEntry Data Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/structs.html Illustrates the structure and content of an SAdsSymbolEntry object. ```python value: 57172 # Current value info.entryLength: 88 # Total storage space for this symbol info.iGroup: 16448 # Group index info.iOffs: 385000 # Offset index inside group info.size: 2 # Number of bytes needed for the value info.dataType: 18 # Symbol type, in this case constants.ADST_UINT16 (18) info.flags: 8 # TwinCAT byte flags info.nameLength: 11 # Number of characters in the name info.typeLength: 4 # Number of characters in the PLC string representation of the type info.commentLength: 20 # Number of characters in the comment info.stringBuffer: # Concatenation of all string info bytes(info.stringBuffer): b'GVL.counter\x00UINT\x00 Counter (in ' 'pulses)\x00\x95\x19\x07\x18\x00\x00\x00\x00' bytes(info.stringBuffer).encode(): "GVL.counter UINT Counter (in pulses)" info.name: "GVL.counter" # The name section from the buffer info.symbol_type: "UINT" # The symbol_type section from the buffer info.comment: " Counter (in pulses)" # The comment (if any) ``` -------------------------------- ### Register a device notification Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Example usage of the notification decorator to handle incoming PLC data updates. ```python import pyads plc = pyads.Connection('172.18.3.25.1.1', 851) @plc.notification(pyads.PLCTYPE_STRING) def callback(handle, name, timestamp, value): print(handle, name, timestamp, value) with plc: attr = pyads.NotificationAttrib(20, pyads.ADSTRANS_SERVERCYCLE) handles = plc.add_device_notification('GVL.test', attr, callback) while True: pass ``` -------------------------------- ### Define structure_def for write_structure_by_name Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Example of the tuple structure required to define PLC types for structure writing. ```python structure_def = ( ('rVar', pyads.PLCTYPE_LREAL, 1), ('sVar', pyads.PLCTYPE_STRING, 2, 35), ('sVar', pyads.PLCTYPE_STRING, 1), ('rVar1', pyads.PLCTYPE_REAL, 1), ('iVar', pyads.PLCTYPE_DINT, 1), ) # i.e ('Variable Name', variable type, arr size (1 if not array), # length of string (if defined in PLC)) ``` -------------------------------- ### Define a PLC structure definition Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/ads.html Example structure definition tuple used for calculating sizes or parsing bytes from the PLC. ```python structure_def = ( ('rVar', pyads.PLCTYPE_LREAL, 1), ('sVar', pyads.PLCTYPE_STRING, 2, 35), ('sVar1', pyads.PLCTYPE_STRING, 1), ('rVar1', pyads.PLCTYPE_REAL, 1), ('iVar', pyads.PLCTYPE_DINT, 1), ('iVar1', pyads.PLCTYPE_INT, 3), ) # i.e ('Variable Name', variable type, arr size (1 if not array), # length of string (if defined in PLC)) ``` -------------------------------- ### Inspect SAdsSymbolEntry structure fields Source: https://pyads.readthedocs.io/en/latest/pyads.html Example showing the mapping of fields within an SAdsSymbolEntry structure, including how to decode the string buffer into name, type, and comment components. ```python value: 57172 # Current value info.entryLength: 88 # Total storage space for this symbol info.iGroup: 16448 # Group index info.iOffs: 385000 # Offset index inside group info.size: 2 # Number of bytes needed for the value info.dataType: 18 # Symbol type, in this case constants.ADST_UINT16 (18) info.flags: 8 # TwinCAT byte flags info.nameLength: 11 # Number of characters in the name info.typeLength: 4 # Number of characters in the PLC string representation of the type info.commentLength: 20 # Number of characters in the comment info.stringBuffer: # Concatenation of all string info bytes(info.stringBuffer): b'GVL.counterUINTCounter (in ' 'pulses)•' bytes(info.stringBuffer).encode(): "GVL.counter UINT Counter (in pulses)" info.name: "GVL.counter" # The name section from the buffer info.symbol_type: "UINT" # The symbol_type section from the buffer info.comment: " Counter (in pulses)" # The comment (if any) ``` -------------------------------- ### Use notification decorator Source: https://pyads.readthedocs.io/en/latest/pyads.html Example of using the notification decorator to handle incoming PLC data updates. ```python >>> import pyads >>> >>> plc = pyads.Connection('172.18.3.25.1.1', 851) >>> >>> >>> @plc.notification(pyads.PLCTYPE_STRING) >>> def callback(handle, name, timestamp, value): >>> print(handle, name, timestamp, value) >>> >>> >>> with plc: >>> attr = pyads.NotificationAttrib(20, >>> pyads.ADSTRANS_SERVERCYCLE) >>> handles = plc.add_device_notification('GVL.test', attr, >>> callback) >>> while True: >>> pass ``` -------------------------------- ### Read and Write Nested Structures Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Example of passing nested data using OrderedDict to a PLC structure. ```python >>> from collections import OrderedDict >>> vars_to_write = collections.OrderedDict([ ... ('rVar',0.1), ... ('structVar', ( ... OrderedDict([ ... ('rVar', 11.1), ... ('rVar2', 22.2), ... ('iVar', 3), ... ('iVar2', [4, 44, 444]), ... ('sVar', 'abc') ... ]), ... OrderedDict([ ... ('rVar', 55.5), ... ('rVar2', 66.6), ... ('iVar', 7), ... ('iVar2', [8, 88, 888]), ... ('sVar', 'xyz') ... ])) ... ) ... ]) >>> plc.write_structure_by_name('GVL.sample_structure', vars_to_write, structure_def) >>> plc.read_structure_by_name('GVL.sample_structure', structure_def) ... OrderedDict({'rVar': 0.1, 'structVar': [OrderedDict({'rVar': 11.1, 'rVar2': 22.200000762939453, 'iVar': 3, 'iVar2': ... [4, 44, 444], 'sVar': 'abc'}), OrderedDict({'rVar': 55.5, 'rVar2': 66.5999984741211, 'iVar': 7, 'iVar2': [8, 88, 888], ... 'sVar': 'xyz'})]}) ``` -------------------------------- ### Define a PLC structure definition Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/ads.html Example structure definition tuple used for mapping Python dictionaries to PLC data types. ```python structure_def = ( ('rVar', pyads.PLCTYPE_LREAL, 1), ('sVar', pyads.PLCTYPE_STRING, 2, 35), ('sVar2', pyads.PLCTYPE_STRING, 1), ('rVar1', pyads.PLCTYPE_REAL, 1), ('iVar', pyads.PLCTYPE_DINT, 1), ('iVar1', pyads.PLCTYPE_INT, 3) ) # i.e ('Variable Name', variable type, arr size (1 if not array), # length of string (if defined in PLC)) ``` -------------------------------- ### pyads.utils.get_num_of_chars Source: https://pyads.readthedocs.io/en/latest/pyads.html Gets the number of characters in a Beckhoff string based on the symbol type. ```APIDOC ## pyads.utils.get_num_of_chars(symbol_type_str) -> int ### Description Gets the number of characters in a Beckhoff string, wstring, or string array using the symbol type string. ### Parameters - **symbol_type_str** (SAdsSymbolEntry.symbol_type) - Required - Symbol type of the string. ``` -------------------------------- ### Connect to a target on Linux Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/routing.rst.txt Use the Connection class to establish a route to a target device on Linux. ```python >>> import pyads >>> remote_ip = '192.168.0.100' >>> remote_ads = '5.12.82.20.1.1' >>> with pyads.Connection(remote_ads, pyads.PORT_TC3PLC1, remote_ip) as plc: >>> plc.read_by_name('.TAG_NAME', pyads.PLCTYPE_INT) ``` -------------------------------- ### Connect to PLC and perform basic I/O operations Source: https://pyads.readthedocs.io/en/latest/_sources/quickstart.rst.txt Demonstrates adding a route, opening a connection, reading and writing variables, and using symbols. ```python >>> import pyads >>> # create some constants for connection >>> CLIENT_NETID = "192.168.1.10.1.1" >>> CLIENT_IP = "192.168.1.10" >>> TARGET_IP = "192.168.1.11" >>> TARGET_USERNAME = "Administrator" >>> TARGET_PASSWORD = "1" >>> ROUTE_NAME = "route-to-my-plc" >>> # add a new route to the target plc >>> pyads.add_route_to_plc( >>> CLIENT_NETID, CLIENT_IP, TARGET_IP, TARGET_USERNAME, TARGET_PASSWORD, >>> route_name=ROUTE_NAME >>> ) >>> # connect to plc and open connection using TwinCAT3. >>> # route is added automatically to client on Linux, on Windows use the TwinCAT router >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> plc.open() >>> # check the connection state >>> plc.read_state() (0, 5) >>> # read int value by name >>> i = plc.read_by_name("GVL.int_val") >>> # write int value by name >>> plc.write_by_name("GVL.real_val", 42.0) >>> # create a symbol that automatically updates to the plc value >>> real_val = plc.get_symbol("GVL.real_val", auto_update=True) >>> print(real_val.value) 42.0 >>> real_val.value = 5.0 >>> print(plc.read_by_name("GVL.real_val")) 5.0 >>> # close connection >>> plc.close() ``` -------------------------------- ### Create an AdsSymbol instance Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/symbols.rst.txt Demonstrates initializing a connection and retrieving a symbol by name, which triggers an automatic lookup of address and type. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> plc.open() >>> symbol = plc.get_symbol('global.bool_value') ``` -------------------------------- ### Write to target using buffer Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/symbols.rst.txt Demonstrates writing the current buffer to the target device. ```python >>> symbol.write(3.14) >>> # Is identical to: >>> symbol.value = 3.14 >>> symbol.write() ``` -------------------------------- ### Get PLC Symbol Information Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/pyads_ex.html Retrieves metadata for a PLC variable by name. ```python def adsGetSymbolInfo(port: int, address: AmsAddr, data_name: str) -> SAdsSymbolEntry: """Get the symbol information of the PLC-variable. :param int port: local AMS port as returned by adsPortOpenEx() :param pyads.structs.AmsAddr address: local or remote AmsAddr :param string data_name: data name :rtype: SAdsSymbolInfo :return: symbol_info: PLC Symbol info """ symbol_info = adsSyncReadWriteReqEx2( port, address, ADSIGRP_SYM_INFOBYNAMEEX, 0x0, SAdsSymbolEntry, data_name, PLCTYPE_STRING, ) return symbol_info ``` -------------------------------- ### Create an AdsSymbol with manual configuration Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/symbols.rst.txt Shows how to provide index group, index offset, and symbol type manually to skip the automatic lookup process. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> plc.open() # Remaining info will be looked up: >>> symbol = plc.get_symbol('global.bool_value') # Alternatively, specify all information and no lookup will be done: >>> symbol = plc.get_symbol('global.bool_value', index_group=123, index_offset=12345, symbol_type=pyads.PLCTYPE_BOOL) ``` -------------------------------- ### Registering and Handling Device Notifications Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Demonstrates how to connect to a local TwinCAT PLC, register a device notification with a callback, and clean up the notification. ```python >>> import pyads >>> from ctypes import sizeof >>> >>> # Connect to the local TwinCAT PLC >>> plc = pyads.Connection('127.0.0.1.1.1', 851) >>> tag = {"GVL.myvalue": pyads.PLCTYPE_INT} >>> >>> # Create callback function that prints the value >>> def mycallback(notification: SAdsNotificationHeader, data: str) -> None: >>> data_type = tag[data] >>> handle, timestamp, value = plc.parse_notification(notification, data_type) >>> print(value) >>> >>> with plc: >>> # Add notification with default settings >>> attr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_INT)) >>> >>> handles = plc.add_device_notification("GVL.myvalue", attr, mycallback) >>> >>> # Remove notification >>> plc.del_device_notification(handles) ``` -------------------------------- ### Get Local Address Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Retrieves the local AMS-address and port number associated with the current connection. ```python def get_local_address(self) -> Optional[AmsAddr]: """Return the local AMS-address and the port number. :rtype: AmsAddr """ if self._port is not None: return adsGetLocalAddressEx(self._port) return None ``` -------------------------------- ### Write and Read PLC Structures Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Demonstrates writing and reading structured data using OrderedDict and structure definitions. ```python >>> vars_to_write = OrderedDict([ ... ('rVar', 11.1), ... ('rar2', 22.2), ... ('iVar', 3), ... ('iVar2', [4, 44, 444]), ... ('sVar', 'abc')] ... ) >>> plc.write_structure_by_name('global.sample_structure', vars_to_write, structure_def) >>> plc.read_structure_by_name('global.sample_structure', structure_def) OrderedDict([('rVar', 11.1), ('rVar2', 22.2), ('iVar', 3), ('iVar2', [4, 44, 444]), ('sVar', 'abc')]) ``` -------------------------------- ### pyads.Connection.open() Source: https://pyads.readthedocs.io/en/latest/pyads.html Establishes a connection to the TwinCAT message router. ```APIDOC ## open() ### Description Connect to the TwinCAT message router. ### Returns - **None** ``` -------------------------------- ### Add and remove device notifications Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Demonstrates connecting to a PLC, registering a callback for notifications, and cleaning up handles. ```python >>> import pyads >>> from ctypes import sizeof >>> >>> # Connect to the local TwinCAT PLC >>> plc = pyads.Connection('127.0.0.1.1.1', 851) >>> >>> # Create callback function that prints the value >>> def mycallback(notification, data): >>> contents = notification.contents >>> value = next( >>> map(int, >>> bytearray(contents.data)[0:contents.cbSampleSize]) >>> ) >>> print(value) >>> >>> with plc: >>> # Add notification with default settings >>> atr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_INT)) >>> handles = plc.add_device_notification("GVL.myvalue", atr, mycallback) >>> >>> # Remove notification >>> plc.del_device_notification(handles) ``` -------------------------------- ### Read and write values by name Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Demonstrates reading and writing variables, including symbol caching options. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1): >>> plc.open() >>> >>> plc.read_by_name('GVL.bool_value') # datatype will be queried and cached True >>> plc.read_by_name('GVL.bool_value') # cached datatype will be used True >>> plc.read_by_name('GVL.bool_value', cache_symbol_info=False) # datatype will not be cached and queried on each call True >>> plc.read_by_name('GVL.int_value', pyads.PLCTYPE_INT) # datatype is provided and will not be queried 0 >>> plc.write_by_name('GVL.int_value', 10) # write to target >>> plc.read_by_name('GVL.int_value') 10 >>> plc.close() ``` -------------------------------- ### Accessing and Setting Symbol Value Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/symbol.html Gets or sets the current value of the symbol, with optional automatic write if auto-update is enabled. ```python @property def value(self) -> Any: """Return the current value of the symbol.""" return self._value @value.setter def value(self, val: Any) -> None: """Set the current value of the symbol. If auto_update is True then the the write command will be called automatically. """ self._value = val # write value to plc if auto_update is enabled if self.auto_update: self.write(val) ``` -------------------------------- ### AdsSymbol.__init__ Source: https://pyads.readthedocs.io/en/latest/pyads.html Initializes an AdsSymbol instance to point to a specific ADS variable on a PLC connection. ```APIDOC ## AdsSymbol(plc, name=None, index_group=None, index_offset=None, symbol_type=None, comment=None, auto_update=False, structure_def=None, array_size=1) ### Description Creates an AdsSymbol instance. Specify either the variable name or the index_group and index_offset to locate the symbol. ### Parameters - **plc** (Connection) - Required - Connection instance. - **name** (str) - Optional - Name of the variable. - **index_group** (int) - Optional - Index group of the symbol. - **index_offset** (int) - Optional - Index offset of the symbol. - **symbol_type** (str/Type) - Optional - PLC variable type (e.g., pyads.PLCTYPE_DINT). - **structure_def** (tuple) - Optional - Definition for complex structures. - **array_size** (int) - Optional - Size of array if reading an array of structures. ``` -------------------------------- ### Get Number of Characters from Symbol Type Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/utils.html Parses the symbol type string to determine the character capacity of a Beckhoff string. ```python def get_num_of_chars(symbol_type_str: SAdsSymbolEntry.symbol_type) -> int: """Gets the number of characters in a Beckhoff string using the symbol type str. TODO: Find this information some other way without string manipulation of the symbol type? Args: symbol_type_str (SAdsSymbolEntry.symbol_type): Symbol type of a string, wstring, or a string array Returns: num_characters (int): The number of characters im the string """ # Find "STRING" in type pos_of_string_in_name = symbol_type_str.upper().find(PLC_STRING_TYPE_NAME) # If not string return -1 if pos_of_string_in_name == -1: return(-1) # Generate start index based on the position of PLC_STRING_TYPE_NAME in the array start_index = pos_of_string_in_name + len(PLC_STRING_TYPE_NAME) + 1 # Return the value cast as int try: return(int(symbol_type_str[start_index:-1])) except ValueError: return(PLC_DEFAULT_STRING_SIZE) ``` -------------------------------- ### Constructing an AmsPacket from raw data Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/testserver.html Parses raw byte data to initialize an AmsHeader and returns a complete AmsPacket object. ```python ams_header = AmsHeader( # Extract target/source net ID's and ports data[6:12], data[12:14], data[14:20], data[20:22], # Extract command ID, state flags, and data length data[22:24], data[24:26], data[26:30], # Extract error code, invoke ID, and data data[30:34], data[34:38], data[38:], ) return AmsPacket(tcp_header, ams_header) ``` -------------------------------- ### AdvancedHandler Initialization and Reset Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/advanced_handler.html Initializes the variable storage database and provides a method to clear all saved variables. ```python def __init__(self) -> None: self._data: Dict[Tuple[int, int], PLCVariable] = {} # This will be our variables database # We won't both with indexing it by handle or name, speed is not # important. We store by group + offset index and will have to # search inefficiently for name or handle. (Unlike real ADS!) self.reset() ``` ```python def reset(self) -> None: """Clear saved variables in handler""" self._data = {} ``` -------------------------------- ### Get PLC Variable Handle Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/pyads_ex.html Retrieves the handle for a specific PLC variable by name using the ADS symbol handle group. ```python def adsGetHandle(port: int, address: AmsAddr, data_name: str) -> int: """Get the handle of the PLC-variable. :param int port: local AMS port as returned by adsPortOpenEx() :param pyads.structs.AmsAddr address: local or remote AmsAddr :param string data_name: data name :rtype: int :return: handle: PLC-variable handle """ handle = adsSyncReadWriteReqEx2( port, address, ADSIGRP_SYM_HNDBYNAME, 0x0, PLCTYPE_UDINT, data_name, PLCTYPE_STRING, ) return handle ``` -------------------------------- ### Initialize and Run AdsTestServer Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/testserver.html The AdsTestServer class inherits from threading.Thread and manages the TCP/IP socket server. It supports context manager usage via __enter__ and __exit__ methods. ```python class AdsTestServer(threading.Thread): """Simple ADS testing server. :ivar function handler: Request handler (see `default_handler` for example) :ivar str ip_address: Host address for server. Defaults to '127.0.0.1' :ivar int port: Host port to listen on, defaults to 48898 """ def __init__( self, handler: "AbstractHandler" = None, ip_address: str = "127.0.0.1", port: int = ADS_PORT, logging: bool = True, *args: Any, **kwargs: Any, ) -> None: self.handler = handler or BasicHandler() self.ip_address = ip_address self.port = port self._run = True global logger logger = logger if logging else null_logger # Keep track of all received AMS packets self.request_history: List[AmsPacket] = [] # Initialize TCP/IP socket server self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # Set option to allow instant socket reuse after shutdown self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server.bind((self.ip_address, self.port)) # Make sure we clean up on exit atexit.register(self.close) # Daemonize the server thread # Container for client connection threads self.clients: List[AdsClientConnection] = [] super(AdsTestServer, self).__init__(*args, **kwargs) self.daemon = True def __enter__(self) -> "AdsTestServer": """Enter context.""" self.start() return self def __exit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: """Exit context.""" self.close() ``` -------------------------------- ### Run Server Loop Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/testserver.html The run method listens for incoming TCP connections and spawns new AdsClientConnection threads for each client. ```python def run(self) -> None: """Listen for incoming connections from clients.""" self._run = True # Start server listening self.server.listen(5) logger.info( "Server listening on {0}:{1}".format( self.ip_address or "localhost", self.port ) ) # Server loop while self._run: # Check for new connections at server socket ready, _, _ = select.select([self.server], [], [], 0.1) if ready: # Accept connection from client try: client, address = self.server.accept() except: continue logger.info("New connection from {0}:{1}".format(*address)) # Delegate handling of connection to client thread client_thread = AdsClientConnection( handler=self.handler, client=client, address=address, server=self ) client_thread.daemon = True client_thread.start() self.clients.append(client_thread) ``` -------------------------------- ### Define a structure definition for AdsSymbol Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/symbol.html Example of a structure definition tuple used to map PLC structures to Python types, including variable names, types, array sizes, and string lengths. ```python structure_def = ( ('rVar', pyads.PLCTYPE_LREAL, 1), ('sVar', pyads.PLCTYPE_STRING, 2, 35), ('SVar1', pyads.PLCTYPE_STRING, 1), ('rVar1', pyads.PLCTYPE_REAL, 1), ('iVar', pyads.PLCTYPE_DINT, 1), ('iVar1', pyads.PLCTYPE_INT, 3), ) # i.e ('Variable Name', variable type, arr size (1 if not array), # length of string (if defined in PLC)) ``` -------------------------------- ### AdsClientConnection class initialization Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/testserver.html Initializes the connection handler with the provided socket and server instance, registering an exit cleanup. ```python def __init__(self, handler: "AbstractHandler", client: socket.socket, address: str, server: AdsTestServer, *args: Any, **kwargs: Any) -> None: self.handler = handler self.server = server self.client = client self.client_address = address # Make sure we clean up on exit atexit.register(self.close) # Server loop execution flag self._run = True super(AdsClientConnection, self).__init__(*args, **kwargs) ``` -------------------------------- ### Add route to testserver Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/testserver.rst.txt Configures the route to the testserver instance from a Python console. ```python >>> import pyads >>> pyads.add_route("127.0.0.1.1.1", "127.0.0.1") ``` -------------------------------- ### Connect to a remote device Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Basic connection lifecycle using open and close methods. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> plc.open() >>> plc.close() ``` -------------------------------- ### Connect to a remote device Source: https://pyads.readthedocs.io/en/latest/documentation/connection.html Demonstrates opening and closing a connection to a PLC. The connection can be managed manually or via a context manager. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> plc.open() >>> plc.close() ``` ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> with plc: >>> # ... ``` -------------------------------- ### Detect Operating System Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/utils.html Functions to check the current platform environment. ```python def platform_is_linux() -> bool: """Return True if current platform is Linux or Mac OS.""" return sys.platform.startswith("linux") or sys.platform.startswith("darwin") def platform_is_windows() -> bool: """Return True if current platform is Windows.""" # cli being .NET (IronPython) return sys.platform == "win32" or sys.platform == "cli" def platform_is_freebsd() -> bool: """Return True if current platform is FreeBSD.""" return sys.platform.startswith("freebsd") ``` -------------------------------- ### Device Information and State Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Methods to retrieve local address, device state, and server information. ```APIDOC ## get_local_address() ### Description Return the local AMS-address and the port number. ## read_state() ### Description Read the current ADS-state and the machine-state from the ADS-server. ## read_device_info() ### Description Read the name and the version number of the ADS-server. ``` -------------------------------- ### Platform Check Functions Source: https://pyads.readthedocs.io/en/latest/pyads.html Functions to determine the current operating system platform. ```APIDOC ## Platform Check Functions ### pyads.utils.platform_is_freebsd() -> bool Returns True if the current platform is FreeBSD. ### pyads.utils.platform_is_linux() -> bool Returns True if the current platform is Linux or Mac OS. ### pyads.utils.platform_is_windows() -> bool Returns True if the current platform is Windows. ``` -------------------------------- ### Open Connection to TwinCAT Router Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Establishes a connection to the TwinCAT message router and handles route addition on Linux systems. ```python def open(self) -> None: """Connect to the TwinCAT message router.""" if self._open: return if self.ams_net_id is None: self.ams_net_id = adsGetNetIdForPLC(self.ip_address) self._adr = AmsAddr(self.ams_net_id, self.ams_net_port) self._port = adsPortOpenEx() if linux: try: adsAddRoute(self._adr.netIdStruct(), self.ip_address) except ADSError: adsPortCloseEx(self._port) self._port = None raise self._open = True ``` -------------------------------- ### Read and write values by name Source: https://pyads.readthedocs.io/en/latest/documentation/connection.html Shows how to read and write PLC variables. Symbol information is cached by default, but this can be disabled. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1): >>> plc.open() >>> >>> plc.read_by_name('GVL.bool_value') # datatype will be queried and cached True >>> plc.read_by_name('GVL.bool_value') # cached datatype will be used True >>> plc.read_by_name('GVL.bool_value', cache_symbol_info=False) # datatype will not be cached and queried on each call True >>> plc.read_by_name('GVL.int_value', pyads.PLCTYPE_INT) # datatype is provided and will not be queried 0 >>> plc.write_by_name('GVL.int_value', 10) # write to target >>> plc.read_by_name('GVL.int_value') 10 >>> plc.close() ``` ```python >>> plc.read_by_name('GVL.wrong_name', pyads.PLCTYPE_BOOL) ADSError: ADSError: symbol not found (1808) ``` ```python >>> plc.read_by_name('GVL.sample_string', pyads.PLCTYPE_STRING) 'Hello World' >>> plc.write_by_name('GVL.sample_string', 'abc', pyads.PLCTYPE_STRING) >>> plc.read_by_name('GVL.sample_string', pyads.PLCTYPE_STRING) 'abc' ``` -------------------------------- ### Connection(ams_net_id, ams_net_port, ip_address) Source: https://pyads.readthedocs.io/en/latest/pyads.html Initializes a new connection to an ADS device. ```APIDOC ## Connection(ams_net_id, ams_net_port, ip_address) ### Description Initializes a class for managing the connection to an ADS device. ### Parameters - **ams_net_id** (str) - Optional - AMS net id of the remote device - **ams_net_port** (int) - Optional - Port of the remote device - **ip_address** (str) - Optional - The IP address of the device ``` -------------------------------- ### Manage PLC Handles Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Manually acquiring and releasing handles to optimize read/write performance. ```python >>> var_handle = plc.get_handle('global.bool_value') >>> plc.write_by_name('', True, pyads.PLCTYPE_BOOL, handle=var_handle) >>> plc.read_by_name('', pyads.PLCTYPE_BOOL, handle=var_handle) True >>> plc.release_handle(var_handle) ``` -------------------------------- ### Read and write structures with multiple datatypes Source: https://pyads.readthedocs.io/en/latest/documentation/connection.html Requires defining a structure tuple and using OrderedDict for data mapping. The PLC structure must use pack_mode 1. ```text {attribute 'pack_mode' := '1'} TYPE sample_structure : STRUCT rVar : LREAL; rVar2 : REAL; iVar : INT; iVar2 : ARRAY [1..3] OF DINT; sVar : STRING; END_STRUCT END_TYPE ``` ```python >>> structure_def = ( ... ('rVar', pyads.PLCTYPE_LREAL, 1), ... ('rVar2', pyads.PLCTYPE_REAL, 1), ... ('iVar', pyads.PLCTYPE_INT, 1), ... ('iVar2', pyads.PLCTYPE_DINT, 3), ... ('sVar', pyads.PLCTYPE_STRING, 1) ... ) >>> vars_to_write = OrderedDict([ ... ('rVar', 11.1), ... ('rar2', 22.2), ... ('iVar', 3), ... ('iVar2', [4, 44, 444]), ... ('sVar', 'abc')] ... ) >>> plc.write_structure_by_name('global.sample_structure', vars_to_write, structure_def) >>> plc.read_structure_by_name('global.sample_structure', structure_def) OrderedDict([('rVar', 11.1), ('rVar2', 22.2), ('iVar', 3), ('iVar2', [4, 44, 444]), ('sVar', 'abc')]) ``` -------------------------------- ### Use context manager for connection Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Automatically close the connection using the with statement. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', pyads.PORT_TC3PLC1) >>> with plc: >>> # ... ``` -------------------------------- ### Handle Device Notification Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/advanced_handler.html Placeholder handler for device notifications. ```python def handle_devicenote() -> bytes: """Handle a device notification.""" logger.info("Command received: DEVICE_NOTIFICATION") # No response data required return b"" ``` -------------------------------- ### Open ADS Port Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/pyads_ex.html Initializes a connection to the TwinCAT message router and returns the port number. ```python def adsPortOpenEx() -> int: """Connect to the TwinCAT message router. :rtype: int :return: port number ``` -------------------------------- ### Connection.get_all_symbols Source: https://pyads.readthedocs.io/en/latest/documentation/symbols.html Retrieves a list of all declared variables on the connected device. ```APIDOC ## Connection.get_all_symbols() Returns a list of objects containing metadata for all variables on the device, including index_group, index_offset, name, symtype, and comment. ``` -------------------------------- ### read_device_info() Source: https://pyads.readthedocs.io/en/latest/pyads.html Read the name and version number of the ADS-server. ```APIDOC ## read_device_info() ### Description Read the name and the version number of the ADS-server. ### Returns - **Tuple[str, AdsVersion]** - device name and version. ``` -------------------------------- ### is_open Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Property to check the current state of the ADS connection. ```APIDOC ## is_open ### Description Returns the current connection state of the ADS device. ### Returns - **bool** - True if the connection is open, False otherwise. ``` -------------------------------- ### Connection.read_structure_by_name / write_structure_by_name Source: https://pyads.readthedocs.io/en/latest/documentation/connection.html Methods for reading and writing complex structures defined in the PLC. ```APIDOC ## Connection.read_structure_by_name(name, structure_def) ## Connection.write_structure_by_name(name, value, structure_def) ### Description Handles complex structures using a tuple definition that maps variable names to types and counts. ### Parameters - **name** (str) - Required - The name of the structure on the PLC. - **structure_def** (tuple) - Required - A tuple defining the structure layout (name, type, count). - **value** (OrderedDict) - Required (for write) - The data to write as an OrderedDict. ``` -------------------------------- ### adsSyncReadDeviceInfoReqEx Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/pyads_ex.html Reads the name and version number of the ADS-server. ```APIDOC ## adsSyncReadDeviceInfoReqEx(port: int, address: AmsAddr) ### Description Reads the name and the version number of the ADS-server. ### Parameters - **port** (int) - Required - The local AMS port. - **address** (AmsAddr) - Required - The local or remote AmsAddr. ### Response - **Tuple[str, AdsVersion]** - Returns a tuple containing the device name and version. ``` -------------------------------- ### Handle ADS Write Control Request Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/testserver/advanced_handler.html Processes WRITE_CONTROL commands, returning an empty byte string as no response data is required. ```python def handle_writectrl() -> bytes: """Handle writectrl request.""" logger.info("Command received: WRITE_CONTROL") # No response data required return b"" ``` -------------------------------- ### Configure notification attributes Source: https://pyads.readthedocs.io/en/latest/documentation/symbols.html Specify custom notification attributes and user handles. ```python >>> attr = NotificationAttrib(length=sizeof(pyads.PLCTYPE_BOOL), max_delay=1.0, cycle_time=1.0) >>> user_handle = 123 >>> symbol.add_device_notification(my_func, attr=attr, user_handle=user_handle) ``` -------------------------------- ### Add device notifications Source: https://pyads.readthedocs.io/en/latest/documentation/symbols.html Attach callbacks to symbols to handle device notifications. ```python >>> symbol.add_device_notification(my_func) ``` -------------------------------- ### Retrieve and print all PLC symbols Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/symbols.rst.txt Fetches all available symbols from the PLC and iterates through their attributes to display them. ```python >>> symbols = plc.get_all_symbols() >>> print('\n'.join("%s: %s" % item for item in vars(symbols[0]).items())) index_group: 16448 index_offset: 384800 name: Constants.bFPUSupport symtype: BOOL comment: Does the target support multiple cores? ``` -------------------------------- ### pyads.structs.AmsAddr Source: https://pyads.readthedocs.io/en/latest/pyads.html Wrapper for SAmsAddr-structure to address an ADS device. ```APIDOC ## pyads.structs.AmsAddr(netid, port) ### Description Create a new AmsAddr object by a given netid and port. ### Methods - **amsAddrStruct()** -> SAmsAddr: Return the c-types structure SAmsAddr. - **netIdStruct()** -> SAmsNetId: Return the c-types structure SAmsNetId. - **setAdr(adrString)** -> None: Set the AMS-address according to the given IP-address. - **toString()** -> str: Textual representation of the AMS address. ``` -------------------------------- ### Define and Register a PLC Notification with Structure Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Uses a tuple-based structure definition to register a notification callback for reading multiple PLC data types at once. ```python >>> structure_def = ( ... ('rVar', pyads.PLCTYPE_LREAL, 1), ... ('rVar2', pyads.PLCTYPE_REAL, 1), ... ('iVar', pyads.PLCTYPE_INT, 1), ... ('iVar2', pyads.PLCTYPE_DINT, 3), ... ('sVar', pyads.PLCTYPE_STRING, 1)) >>> >>> size_of_struct = pyads.size_of_structure(structure_def) >>> >>> @plc.notification(ctypes.c_ubyte * size_of_struct) >>> def callback(handle, name, timestamp, value): ... values = pyads.dict_from_bytes(value, structure_def) ... print(values) >>> >>> attr = pyads.NotificationAttrib(size_of_struct) >>> plc.add_device_notification('global.sample_structure', attr, callback) ``` -------------------------------- ### Perform read and write operations Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/symbols.rst.txt Basic usage for reading from and writing to a symbol object. ```python >>> symbol.read() True >>> symbol.write(False) >>> symbol.read() False >>> plc.close() ``` -------------------------------- ### Registering Device Notifications Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/symbol.html Adds a device notification for the symbol and tracks the handle. ```python handles = self._plc.add_device_notification( (self.index_group, self.index_offset), attr, callback, user_handle ) self._handles_list.append(handles) return handles ``` -------------------------------- ### write_control(ads_state, device_state, data, plc_datatype) Source: https://pyads.readthedocs.io/en/latest/pyads.html Change the ADS state and the machine-state of the ADS-server. ```APIDOC ## write_control ### Description Change the ADS state and the machine-state of the ADS-server. ### Parameters - **ads_state** (int) - Required - New ADS-state, according to ADSTATE constants - **device_state** (int) - Required - New machine-state - **data** (Any) - Required - Additional data - **plc_datatype** (int) - Required - Datatype, according to PLCTYPE constants ``` -------------------------------- ### Use notification decorator Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Simplify notification handling by using a decorator to automatically convert ADS types to Python types. ```python >>> import pyads >>> plc = pyads.Connection('127.0.0.1.1.1', 48898) >>> plc.open() >>> >>> @plc.notification(pyads.PLCTYPE_INT) >>> def callback(handle, name, timestamp, value): >>> print( >>> '{1}: received new notitifiction for variable "{0}", value: {2}' >>> .format(name, timestamp, value) >>> ) >>> >>> plc.add_device_notification('GVL.intvar', pyads.NotificationAttrib(2), callback) >>> # Write to the variable to trigger a notification >>> plc.write_by_name('GVL.intvar', 123, pyads.PLCTYPE_INT) ``` -------------------------------- ### Write Data Synchronously Source: https://pyads.readthedocs.io/en/latest/_modules/pyads/connection.html Sends data synchronously to an ADS-device using index group and offset. ```python def write( self, index_group: int, index_offset: int, value: Any, plc_datatype: Type["PLCDataType"] ) -> None: """Send data synchronous to an ADS-device. :param int index_group: PLC storage area, according to the INDEXGROUP constants :param int index_offset: PLC storage address :param Any value: value to write to the storage address of the PLC :param Type["PLCDataType"] plc_datatype: type of the data given to the PLC, according to PLCTYPE constants """ if self._port is not None: return adsSyncWriteReqEx( self._port, self._adr, index_group, index_offset, value, plc_datatype ) ``` -------------------------------- ### Configure AdvancedHandler variables Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/testserver.rst.txt Defines and adds a variable to the AdvancedHandler for stateful interaction. ```python # Server code handler = AdvancedHandler() test_var = PLCVariable( "Main.my_var", bytes(8), ads_type=constants.ADST_REAL64, symbol_type="LREAL" ) handler.add_variable(test_var) ``` -------------------------------- ### Define Structure Definitions Source: https://pyads.readthedocs.io/en/latest/_sources/documentation/connection.rst.txt Defines the structure layout as tuples for pyads, showing both separate and nested definition styles. ```python >>> substructure_def = ( ... ('rVar', pyads.PLCTYPE_LREAL, 1), ... ('rVar2', pyads.PLCTYPE_REAL, 1), ... ('iVar', pyads.PLCTYPE_INT, 1), ... ('iVar2', pyads.PLCTYPE_DINT, 3), ... ('sVar', pyads.PLCTYPE_STRING, 1) ... ) >>> structure_def = ( ... ('rVar', pyads.PLCTYPE_LREAL, 1), ... ('structVar', substructure_def, 2) ... ) ``` ```python >>> structure_def = ( ... ('rVar', pyads.PLCTYPE_LREAL, 1), ... ('structVar', ( ... ('rVar', pyads.PLCTYPE_LREAL, 1), ... ('rVar2', pyads.PLCTYPE_REAL, 1), ... ('iVar', pyads.PLCTYPE_INT, 1), ... ('iVar2', pyads.PLCTYPE_DINT, 3), ... ('sVar', pyads.PLCTYPE_STRING, 1) ... ), 2) ... ) ``` -------------------------------- ### read_write(index_group, index_offset, plc_read_datatype, value, plc_write_datatype, return_ctypes=False, check_length=True) Source: https://pyads.readthedocs.io/en/latest/pyads.html Read and write data synchronously from/to an ADS-device. ```APIDOC ## read_write ### Description Read and write data synchronous from/to an ADS-device. ### Parameters - **index_group** (int) - Required - PLC storage area, according to the INDEXGROUP constants - **index_offset** (int) - Required - PLC storage address - **plc_read_datatype** (Type) - Required - Type of the data given to the PLC to respond to, or None to not read anything - **value** (Any) - Required - Value to write to the storage address of the PLC - **plc_write_datatype** (Type) - Required - Type of the data given to the PLC, or None to not write anything - **return_ctypes** (bool) - Optional - Return ctypes instead of python types if True (default: False) - **check_length** (bool) - Optional - Check whether the amount of bytes read matches the size of the read data type (default: True) ### Returns - **value** (Any) - The value read from the device. ```