aioisotp 0.1.1

为什么是异步的？

异步编程简化了一些可能的用例

• 单个连接上的全双工接收和发送。

• 同时在多个连接上进行通信。

• 功能寻址，其中发送一个请求，然后所有节点响应，然后按到达顺序处理响应。

• 实现或模拟多个服务器。

无需处理所有线程，以及所需的锁定机制。

安装

从 PyPI 安装

$ pip install aioisotp==0.1.1

文档

可以使用 Sphinx 构建基本文档

$ python setup.py build_sphinx

快速入门

下面是一个使用基于回调的协议实现的 echo 服务器示例，以及作为顺序读取器和写入器流实现的客户端。

import asyncio
import aioisotp


class EchoServer(asyncio.Protocol):

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        # Echo back the same data
        self.transport.write(data)


async def main():
    network = aioisotp.ISOTPNetwork('vcan0',
                                    interface='virtual',
                                    receive_own_messages=True)
    with network.open():
        # A server that uses a protocol
        transport, protocol = await network.create_connection(
            EchoServer, 0x1CDADCF9, 0x1CDAF9DC)

        # A client that uses streams
        reader, writer = await network.open_connection(
            0x1CDAF9DC, 0x1CDADCF9)

        writer.write(b'Hello world!')
        response = await reader.read(4095)
        assert response == b'Hello world!'


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

UDS

本包旨在启用使用需要ISO-TP的其他协议，其中最常见的是UDS。可以使用第三方库（如udsoncan或pyvit）来编码和解码有效负载。

import aioisotp
import udsoncan

...

reader, writer = await network.open_connection(0x1CDAF9DC, 0x1CDADCF9)

# Construct and send request
request = udsoncan.Request(
    udsoncan.services.ReadDataByIdentifier, data=b'\xF1\x90')
writer.write(request.get_payload())

# Wait for response and decode the payload
payload = await reader.read(4095)
response = udsoncan.Response.from_payload(payload)

print(response)
print(response.data)