Python 快速入门

直接使用 PLC4PY API

目前，你需要从 GitHub 仓库而不是 pypi 安装 PLC4Py。一旦我们确定 PLC4Py 已经可以发布，我们就会将其发布到 pypi。

进入 plc4py 目录并运行。这将会把 plc4py 安装到你的全局仓库中。

   pip install .

现在你就可以在应用中使用 PLC4Py 了。下面展示了一个最小示例。

import asyncio
from plc4py.PlcDriverManager import PlcDriverManager

connection_string = "modbus://127.0.0.1:5020"
driver_manager = PlcDriverManager()

async def communicate_with_plc():
    """
    Asynchronously communicates with a PLC using a PlcDriverManager.

    This function establishes a connection to the PLC defined by the connection_string.
    It builds a read request for a specific item ("Random Tag" in this case) using the connection's read request builder.
    The request is then executed asynchronously, and the response code is printed.
    """
    print(f"Connecting to plc: {connection_string}")
    async with driver_manager.connection(connection_string) as connection:
        print(f"Connected to {connection_string}")
        with connection.read_request_builder() as builder:
            print(f"Building read request")
            builder.add_item("Random Tag", "4x00001[10]")
            request = builder.build()
            print(f"Request built")
        print(f"Executing request")
        response = await connection.execute(request)
        print(f"Request executed")
        print(f"Response code: {response.response_code}")

asyncio.run(communicate_with_plc())

PLC4X 通常支持的功能非常有限，这并不是因为我们没有实现相关功能，而是因为 PLC 本身通常只支持非常有限的功能集。

PLC 所支持的、因而也由 PLC4X 支持的基本功能包括：

发现设备（PLC4Py 尚未支持）
列出 PLC 中的资源
读取数据
写入数据
订阅数据（PLC4Py 尚未支持）

总体而言，我们会尽量提供尽可能多的功能。因此，如果某种协议不支持基于订阅的通信，我们的目标是通过后台轮询来模拟这一功能，使其对用户透明（不过该模拟功能尚未实现，但已在我们的路线图中）。

但也存在一些我们无法模拟、或功能被有意禁用的情况：

如果 PLC 和/或协议不支持写入或浏览，我们确实无法提供该功能。

因此，我们使用元数据以编程方式检查某一功能是否可用。

读取数据

# Check if this connection support reading of data.
if not connection.is_read_supported():
    logger.error("This connection doesn't support reading.")

一旦确认某项功能可用，你就可以着手构建第一个请求了。方法是获取一个 ReadRequestBuilder：

# Create a new read request:
# - Give the single item requested an alias name
with connection.read_request_builder() as builder:
    builder.add_item("Random Tag 1", "4x00001[10]")
    builder.add_item("Random Tag 2", "4x00011")
    request = builder.build()

因此，如你所见，通过向请求中添加标签地址，最后调用 build 方法，即可准备好一个请求。

如果你使用的是 BrowseApi，那么你可能还会获得 Tag 对象。在这种情况下，只需使用 add_tag 并传入 PlcTag 对象，而不是地址字符串即可。

通过在请求对象上执行 execute 方法，即可将请求发送到 PLC：

try:
    response = await connection.execute(request)
except TimeOutException:
    # Handle timeout error
except ...
    # Handle all your other errors

通常，所有请求都是异步执行的。一旦请求被完整处理，回调就会被调用；如果一切正常，回调中将包含一个 ReadResponse，如果出现问题，则包含一个异常。

下面的示例将演示你可以使用的部分选项：

for tag_name in response.tag_names:
    if response.tags[tag_name].response_code == PlcResponseCode.OK:
        num_values: int = len(response.tags[tag_name].value)
        # If it's just one element, output just one single line.
        if num_values == 1:
            logger.info("Value[" + tag_name + "]: " + response.tags[tag_name].value)
        else:
            # If it's more than one element, output each in a single row.
            logger.info("Value[" + tag_name + "]:")
            for i in response.tags[tag_name].value.get_list():
                logger.info(" - " + str(i))
    else:
        # Something went wrong, to output an error message instead.
        logger.error("Error[" + tag_name + "]: " + response.tags[tag_name].name())

在 for 循环中，我们演示了用户如何遍历响应中的标签别名。对于普通的读请求，这些别名会由请求中的项预先确定；而对于订阅响应，响应中可能只包含被订阅项中的一部分。

在访问数据之前，建议先检查某个项是否被正确返回。这是通过给定别名的 response_code 属性来完成的。如果该值为 PlcResponseCode.OK，则一切正常；否则它可能是以下之一：

NOT_FOUND
ACCESS_DENIED
INVALID_ADDRESS
INVALID_DATATYPE
INTERNAL_ERROR
RESPONSE_PENDING

假设返回码为 OK，我们就可以继续访问数据了。

由于所有 PlcValue 项都支持 len 属性，用户可以通过调用 len(response.tags[tag_name].value) 来检查返回了某个类型的多少个项。

然后，你可以使用 response.tags[tag_name].value.get_list() 将 PlcList 中的值作为列表来处理。

写入数据

总体而言，写入数据的代码结构与读取数据非常相似。

因此，首先建议检查该连接是否具备写入数据的能力：

// Check if this connection support writing of data.
if not plc_connection.is_write_supported():
  logger.error("This connection doesn't support writing.")

一旦确定可以写入，我们就创建一个新的 PlcWriteRequest.Builder：

// Create a new write request:
// - Give the single item requested an alias name
// - Pass in the data you want to write (for arrays, pass in a list of values)
with connection.write_request_builder() as builder:
    builder.add_item("Random Tag 1", "4x00001[2]", PlcList([PlcINT(1), PlcINT(2)]))
    builder.add_item("Random Tag 2", "4x00011", PlcINT(1))
    request = builder.build()

与读请求相同，向 PLC 发送请求的方式是在请求对象上调用 execute 方法：

try:
    response = await connection.execute(request)
except TimeOutException:
    # Handle timeout error
except ...
    # Handle all your other errors

由于我们不需要处理数据本身，对于写请求，只需简单地检查每个字段的返回码即可。

for tag_name in response.tag_names:
    if response.tags[tag_name].response_code == PlcResponseCode.OK:
        logger.info("Value[" + tag_name + "]: updated");
    else:
        # Something went wrong, to output an error message instead.
        logger.error("Error[" + tag_name + "]: " + response.tags[tag_name].name())

订阅数据

即将推出

直接使用 PLC4PY API

读取数据

写入数据

订阅数据

评论