plc.py

# PLC Server Settings
from datetime import datetime
from time import sleep

from plc_server_client import ApiClient, Configuration
from plc_server_client.api.plc_rest_controller_api import PlcRestControllerApi
from plc_server_client.model.plc_read_request import PlcReadRequest
from plc_server_client.model.plc_read_response import PlcReadResponse

from database_methods import get_all_entitys
from extract import (
    AutoGeneratedUUID,
    Parameter,
    AutoGeneratedField,
    StateProcessor,
    SyncDatabase,
)

PLC_SERVER_HOST = "192.168.167.196"
PLC_SERVER_PORT = "48080"
timeout = 1000


def get_api_client():
    api_client: ApiClient = ApiClient(
        configuration=Configuration(host=f"http://{PLC_SERVER_HOST}:{PLC_SERVER_PORT}")
    )
    return api_client


def get_controller_api(client_api):
    controller_api: PlcRestControllerApi = PlcRestControllerApi(api_client=client_api)
    return controller_api


def read_response(controller_api):
    response: PlcReadResponse = controller_api.multiread(
        PlcReadRequest(
            host="s7://192.168.167.210/0/0",
            fields=[
                "%DB444.DBD22:DINT",
                "%DB444.DBW20:INT",
                "%DB444:3.0:REAL",
                "%DB444:4.0:REAL",
                "%DB444:6.0:REAL",
                "%DB4560:12.0:UDINT",
            ],
            timeout=timeout,
        ),
        _check_return_type=False,
    )
    return response


def make_dict_from_response(response: PlcReadResponse):
    element_dict: dict = {}

    for i in range(0, len(response.fields)):
        element_dict.update(
            {response.fields[i].get("field"): response.fields[i].get("value")}
        )
    element_dict.update({"timestamp": f"{datetime.now()}"})
    return element_dict


if __name__ == "__main__":
    # thread der alle x sec ne anfrage macht.
    api_client: ApiClient = get_api_client()
    controller_api: PlcRestControllerApi = get_controller_api(api_client)

    def get_config() -> dict:
        return {
            "Cycle": {
                "fields": {
                    "uuid": AutoGeneratedUUID(),
                    "id": AutoGeneratedField(),
                    # "druck": Parameter("%DB444:4.0:REAL"),
                    "cycle": Parameter("%DB4560:12.0:UDINT"),
                },
                "primary_key": "id",
            },
            "ToolEquipped": {
                "fields": {
                    "id": AutoGeneratedField(),
                    "uuid": AutoGeneratedUUID(),
                    "tool_name": Parameter("%DB444:6.0:REAL"),
                },
                "primary_key": "id",
            },
            "MaterialEquipped": {
                "fields": {
                    "id": AutoGeneratedField(),
                    "uuid": AutoGeneratedUUID(),
                    "material_name": Parameter("%DB444.DBD22:DINT"),
                    "material_typ": Parameter("%DB444.DBW20:INT"),
                },
                "primary_key": "id",
            },
        }

    # threading.Timer(1.0,read_response,[controller_api]).start()
    end_result = []
    end_result.append(make_dict_from_response(read_response(controller_api)))
    processor: StateProcessor = StateProcessor(get_config())
    processor.init_context(end_result[0])
    db_sync: SyncDatabase = SyncDatabase(processor.get_result())
    db_sync.sync_database_with_index()
    sleep(1)
    while True:
        end_result.append(make_dict_from_response(read_response(controller_api)))
        processor.process_state(end_result[-1])
        print(end_result[-1])
        db_sync.set_end_result(processor.get_result())
        db_sync.sync_database_with_index()
        sleep(0.1)
    arrie = get_all_entitys()