Python通過snap7庫與西門子S7-1200建立S7通信，讀寫存儲器數據，順便寫個流水燈

1.snap7 簡介

snap7 是一個基於以太網與S7系列的西門子PLC通訊的開源庫。

支持包括S7系列的S7-200、S7-200 Smart、S7-300、S7-400、S7-1200以及S7-1500的以太網通信。

適用系統
支持32/64位英特爾/ AMD的所有平台。
例如：Windows ( 除了 windows Me和95)；Linux和類Linux（樹莓派，UBeagleBone Black，DOO 等）;BSD；Oracle Solaris ；Apple OSX

支持語言
Pascal；C#；C++；C；LabVIEW；Python；Node.js；Java，其中介紹比較多的是Python。

snap7官方網站

http://snap7.sourceforge.net/

https://pypi.org/project/python-snap7/

https://python-snap7.readthedocs.io/en/latest/

2.S7通信

西門子S7系列PLC采用以下兩種通訊方式：
1) 開放式的TCP\IP，可以用於連接PLC與其他非西門子硬件
2) 西門子自己開發的S7 Protocol以太網通訊協議，用於西門子內部硬件通訊

這兩者的傳輸報文是不一樣的，如下圖：

西門子數存儲到二進制時方式是大端模式（BIG-Endian），而我們的普通電腦常常為小端模式（Liitle-Endian）。
大端模式是指數據的低位保存在內存的高地址中，而數據的高位保存在內存的低地址中.
小端模式是指數據的低位保存在內存的低地址中，而數據的高位保存在內存的高地址中。
例如：雙字 DWORD 0X2F11214C
PLC

PC

所以數據需要進行轉換。

3.安裝snap7庫

pip install python-snap7

筆者使用的是Python3.6.4和python-snap7 1.1.0，安裝完成后，環境就算搭建好了。通過一個連接測試代碼試試，判斷下環境是否搭建正常。

注意自己新建的文件名不能是snap7，會和庫文件沖突！

import snap7
client = snap7.client.Client()
client.connect('192.168.0.1', 0, 1)
client.disconnect()

如果是下圖提示，則環境正常（192.168.0.1的PLC不存在）。

如果是下圖提示，則環境異常（snap7庫安裝不正確）。

4.讀寫PLC

4.1配置S7-1200

環境搭建正常后，在正式建立通信前PLC還需做些配置工作，主要是開發自身的讀寫權限。具體參照下圖配置：

設置訪問級別

設置DB塊的屬性

通過上述配置，PLC可以正常通信了。

4.2使用snap7讀寫存儲器

python-snap7重要的兩個方法是read_area和write_area，通過這兩個方法就能讀和寫PLC的對應存儲地址。

摘自client.py

def read_area(self, area: Areas, dbnumber: int, start: int, size: int) -> bytearray:
    """Reads a data area from a PLC
    With it you can read DB, Inputs, Outputs, Merkers, Timers and Counters.

    Args:
        area: area to be read from.
        dbnumber: number of the db to be read from. In case of Inputs, Marks or Outputs, this should be equal to 0.
        start: byte index to start reading.
        size: number of bytes to read.

    Returns:
        Buffer with the data read.

    Raises:
        :obj:`ValueError`: if the area is not defined in the `Areas`

    Example:
        >>> import snap7
        >>> client = snap7.client.Client()
        >>> client.connect("192.168.0.1", 0, 0)
        >>> buffer = client.read_area(snap7.types.Areas.DB, 1, 10, 4)  # Reads the DB number 1 from the byte 10 to the byte 14.
        >>> buffer
        bytearray(b'\\x00\\x00')
    """
    if area not in Areas:
        raise ValueError(f"{area} is not implemented in snap7.types")
    elif area == Areas.TM:
        wordlen = WordLen.Timer
    elif area == Areas.CT:
        wordlen = WordLen.Counter
    else:
        wordlen = WordLen.Byte
    type_ = snap7.types.wordlen_to_ctypes[wordlen.value]
    logger.debug(
        f"reading area: {area.name} dbnumber: {dbnumber} start: {start}: amount {size}: wordlen: {wordlen.name}={wordlen.value}")
    data = (type_ * size)()
    result = self._library.Cli_ReadArea(self._pointer, area.value, dbnumber, start,
                                        size, wordlen.value, byref(data))
    check_error(result, context="client")
    return bytearray(data)


@error_wrap
def write_area(self, area: Areas, dbnumber: int, start: int, data: bytearray) -> int:
    """Writes a data area into a PLC.

    Args:
        area: area to be write.
        dbnumber: number of the db to be write to. In case of Inputs, Marks or Outputs, this should be equal to 0.
        start: byte index to start writting.
        data: buffer to be write.

    Returns:
        Snap7 error code.

    Exmaple:
        >>> import snap7
        >>> client = snap7.client.Client()
        >>> client.connect("192.168.0.1", 0, 0)
        >>> buffer = bytearray([0b00000001])
        >>> client.write_area(snap7.types.Areas.DB, 1, 10, buffer)  # Writes the bit 0 of the byte 10 from the DB number 1 to TRUE.
    """
    if area == Areas.TM:
        wordlen = WordLen.Timer
    elif area == Areas.CT:
        wordlen = WordLen.Counter
    else:
        wordlen = WordLen.Byte
    type_ = snap7.types.wordlen_to_ctypes[WordLen.Byte.value]
    size = len(data)
    logger.debug(f"writing area: {area.name} dbnumber: {dbnumber} start: {start}: size {size}: "
                 f"wordlen {wordlen.name}={wordlen.value} type: {type_}")
    cdata = (type_ * len(data)).from_buffer_copy(data)
    return self._library.Cli_WriteArea(self._pointer, area.value, dbnumber, start,
                                       size, wordlen.value, byref(cdata))

《SIMATIC S7-1200 可編程控制器系統手冊》節4.2.1有如下描述：

PLC的數據存儲通過“變量”的形式與存儲區間關聯，分為輸入(I)、輸出(O)、位存儲(M)和數據塊(DB)。程序在訪問對應(I/O)存儲區時，是通過訪問CPU的過程映像對相應地址進行操作的。具體對應關系如下：

故python-snap7中定義的Areas含義為

'PE': 0x81, #input

'PA': 0x82, #output

'MK': 0x83, #bit memory

'DB': 0x84, #DB

'CT': 0x1C, #counters

'TM': 0x1D, #Timers

現在離讀寫PLC還差最后一步，就是起始地址如何確定呢？

對於M3.4，對應的就是M(0x83)，起始地址是3，對應bit位是4。

4.3讀寫示例

import struct
import time
import snap7

def plc_connect(ip, rack=0, slot=1):
    '''
    連接初始化
    :param ip:
    :param rack: 通常為0
    :param slot: 根據plc安裝，一般為0或1
    :return:
    '''
    client = snap7.client.Client()
    client.connect(ip, rack, slot)
    return client

def plc_con_close(client):
    """
    連接關閉
    :param client:
    :return:
    """
    client.disconnect()

def test_mk10_1(client):
    """
    測試M10.1
    :return:
    """
    area = snap7.types.Areas.MK
    dbnumber = 0
    start = 10
    amount = 1
    print('初始值',end='')
    mk_data = client.read_area(area, dbnumber, start, amount)
    print(mk_data)#struct.unpack('!c', mk_data)
    print('置1')
    client.write_area(area, dbnumber, start, b'\x01')
    print('當前值',end='')
    mk_cur = client.read_area(area, dbnumber, start, amount)
    print(mk_cur)

def test_mk_w201(client):
    """
    測試MW201,數據類型為word
    :param client:
    :return:
    """
    area = snap7.types.Areas.MK
    dbnumber = 0
    amount = 2
    start = 201
    print(u'初始值')
    mk_data = client.read_area(area, dbnumber, start, amount)
    print(struct.unpack('!h', mk_data))
    print(u'置12')
    client.write_area(area, dbnumber, start, b'\x12')
    print(u'當前值')
    mk_cur = client.read_area(area, dbnumber, start, amount)
    print(struct.unpack('!h', mk_cur))
    time.sleep(3)
    print(u'置3')
    client.write_area(area, dbnumber, start, b'\x02')
    print(u'當前值')
    mk_cur = client.read_area(area, dbnumber, start, amount)
    print(struct.unpack('!h', mk_cur))

def test_q0_0(client):
    """
    測試Q0.0，會使其輸出高電平
    :return:
    """
    area = snap7.types.Areas.PA
    dbnumber = 0
    start = 0
    amount = 1
    print('初始值',end='')
    mk_data = client.read_area(area, dbnumber, start, amount)
    print(mk_data)#struct.unpack('!c', mk_data)
    print('置1')
    client.write_area(area, dbnumber, start, b'\x01')
    print('當前值',end='')
    mk_cur = client.read_area(area, dbnumber, start, amount)
    print(mk_cur)

if __name__ == "__main__":
    client_fd = plc_connect('192.168.0.2')
    # test_mk10_1(client_fd)
    # test_mk_w201(client_fd)
    test_q0_0(client_fd)
    plc_con_close(client_fd)


'''
'PE': 0x81, #input, I
'PA': 0x82, #output, Q
'MK': 0x83, #bit memory, M
'DB': 0x84, #DB, DBX
'CT': 0x1C, #counters
'TM': 0x1D, #Timers
'''

也可通過db_read()和db_write()讀寫DB塊

import snap7
client = snap7.client.Client()
client.connect('192.168.0.2', 0, 1)
plc_db1 = client.db_read(1, 0, 3)  # 讀取數據塊db1,起始字節,讀取長度
print(plc_db1) # hex(plc_db1[0])
client.db_write(1, 0, b'\x11') # 寫入數據塊db1,起始字節,數據hex
plc_db1 = client.db_read(1, 0, 3)  # 讀取數據塊db1,起始字節,讀取長度
print(plc_db1)
client.disconnect()

讀寫MW變量時，要在變量表中先創建變量，並燒錄進S7-1200

5.流水燈

import struct
import time
import snap7

def plc_connect(ip, rack=0, slot=1):
    '''
    連接初始化
    :param ip:
    :param rack: 通常為0
    :param slot: 根據plc安裝，一般為0或1
    :return:
    '''
    client = snap7.client.Client()
    client.connect(ip, rack, slot)
    return client

def plc_con_close(client):
    """
    連接關閉
    :param client:
    :return:
    """
    client.disconnect()

def ledtrip(client):
    """
    跑馬燈，使Q0.0~5循環亮起
    :return:
    """
    area = snap7.types.Areas.PA
    dbnumber = 0
    start = 0
    delayTime = 0.5
    for i in range(10):
        client.write_area(area, dbnumber, start, bytearray([0b00000001]))
        time.sleep(delayTime)
        client.write_area(area, dbnumber, start, bytearray([0b00000010]))
        time.sleep(delayTime)
        client.write_area(area, dbnumber, start, bytearray([0b00000100]))
        time.sleep(delayTime)
        client.write_area(area, dbnumber, start, bytearray([0b00001000]))
        time.sleep(delayTime)
        client.write_area(area, dbnumber, start, bytearray([0b00010000]))
        time.sleep(delayTime)
        client.write_area(area, dbnumber, start, bytearray([0b00100000]))
        time.sleep(delayTime)
    client.write_area(area, dbnumber, start, b'\x00')


if __name__ == "__main__":
    client_fd = plc_connect('192.168.0.2')
    ledtrip(client_fd)
    plc_con_close(client_fd)

參考

https://pypi.org/project/python-snap7/

https://python-snap7.readthedocs.io/en/latest/

https://blog.csdn.net/zxpbuct/article/details/80079698

https://blog.csdn.net/lcb411/article/details/101147181

https://www.toutiao.com/a6589203413941092868

http://www.6dm.club/index.php/2018/04/07/