"""

This is an example of how to use the snap7 library to read and write data to a PLC.

It is used to manipulate a large DB object containing over 450 'rows' which represent valves

"""

import time

from db_layouts import rc_if_db_1_layout

from db_layouts import tank_rc_if_db_layout

from snap7 import Client, Row, DB

from util.db import print_row

client = Client()

client.connect("192.168.200.24", 0, 3)

def get_db1() -> None:

"""

Here we read out DB1, all data we is put in the all_data

variable and is a bytearray with the raw plc data

"""

all_data = client.db_get(1)

for i in range(400): # items in db

row_size = 130 # size of item

index = i * row_size

offset = index + row_size # end of row in db

print_row(all_data[index:offset])

def show_row(x: int) -> None:

"""

print data in DB of row/object X in

"""

row_size = 126

while True:

data = client.db_read(1, 4 + x * row_size, row_size)

row = Row(data, rc_if_db_1_layout, layout_offset=4)

print("name", row["RC_IF_NAME"])

print(row["RC_IF_NAME"])

break

# do some write action..

# do some check action..

def get_row(x: int) -> Row:

row_size = 126

data = client.db_read(1, 4 + x * row_size, row_size)

row = Row(data, rc_if_db_1_layout, layout_offset=4)

return row

def set_row(x: int, row: Row) -> None:

"""

We use db 1, use offset 4, we replace row x. To find the correct

start_index we mulitpy by row_size by x and we put the

byte array representation of row in the PLC

"""

row_size = 126

client.db_write(1, 4 + x * row_size, row_size, row._bytearray)

def open_row(row: Row) -> None:

"""

open a valve

"""

# row['AutAct'] = 1

row["Occupied"] = 1

row["BatchName"] = "test"

row["AutModLi"] = 1

row["ManModLi"] = 0

row["ModLiOp"] = 1

row["CloseAut"] = 0

row["OpenAut"] = 1

def close_row(row: Row) -> None:

"""

close a valve

"""

# print row['RC_IF_NAME']

row["BatchName"] = ""

row["Occupied"] = 0

row["CloseAut"] = 1

row["OpenAut"] = 0

def open_and_close() -> None:

for x in range(450):

row = get_row(x)

open_row(row)

set_row(x, row)

time.sleep(3)

for x in range(450):

row = get_row(x)

close_row(row)

set_row(x, row)

def set_part_db(start: int, size: int, _bytearray: bytearray) -> None:

data = _bytearray[start : start + size]

client.db_write(1, start, size, data)

# def write_data_db(dbnumber, all_data, size):

# area = snap7.types.S7AreaDB

# client.write_area(area, dbnumber, 0, size, all_data)

def open_and_close_db1() -> None:

t = time.time()

db1 = make_item_db(1)

all_data = db1._bytearray

print(f"row objects: {len(db1.index)}")

for x, (name, row) in enumerate(db1.index.items()):

open_row(row)

# set_part_db(4+x*126, 126, all_data)

t = time.time()

client.write_area(1, all_data, 4 + 126 * 450)

print(f"opening all valves took: {time.time() - t}")

print("sleep...")

time.sleep(5)

for x, (name, row) in enumerate(db1):

close_row(row)

# set_part_db(4+x*126, 126, all_data)

print(time.time() - t)

t = time.time()

client.write_area(1, all_data, 4 + 126 * 450)

print(f"closing all valves took: {time.time() - t}")

def read_tank_db() -> None:

db73 = make_tank_db()

print(len(db73))

for x, (name, row) in enumerate(db73):

print(row)

def make_item_db(db_number: int) -> DB:

t = time.time()

all_data = client.upload(db_number)

print(f"getting all data took: {time.time() - t}")

db1 = DB(

db_number, # the db we use

all_data, # bytearray from the plc

rc_if_db_1_layout, # layout specification

126, # size of the specification

450, # number of row's / specifocations

id_field="RC_IF_NAME", # field we can use to make row

layout_offset=4, # sometimes specification does not start a 0

db_offset=4, # At which point in all_data should we start

# parsing for data

)

return db1

def make_tank_db() -> DB:

tank_data = client.upload(73)

db73 = DB(73, tank_data, tank_rc_if_db_layout, 238, 2, id_field="RC_IF_NAME")

return db73

def print_tag() -> None:

db1 = make_item_db(1)

print(db1["5V315"])

def print_open() -> None:

db1 = make_item_db(1)

for x, (name, row) in enumerate(db1):

if row["BatchName"]:

print(row)

# read_tank_db()

# open_and_close()

# open_and_close_db1()

# time.sleep(1)

# show_row(2)

print_tag()

# print_open()

client.disconnect()