Интеграция компьютерного зрения в АСУ ТП. IEC 61499 + Python + CV = OpenFB

git clone https://github.com/rtsoft-sdc/openFB.git
cd openFb
python3 -m venv .venv 

Активировать окружение

source .venv/bin/activate            # На Linux
#.venv\Scripts\activate              # На Windows

pip install -r requirements.txt

Запустить среду исполнения openFb

python3 run.py


...
2026-07-03 21:01:29,882][ERROR][MainThread] Could not find fboot definition file. Awaiting deployment.
[INFO]	OPCUA server is running on 0.0.0.0:4840
[2026-07-03 21:01:29,882][INFO][MainThread] TCP server starting up on 0.0.0.0 port 61499
[INFO]	Openfb is up and running on 0.0.0.0:61499
[2026-07-03 21:01:29,882][INFO][MainThread] waiting for a connection...

cd ~/workspace/cvdemo/Type Library
python3 ~/openFb/tools/fbt2py ~/workspace/cvdemo/Type Library/QRDETECTOR.fbt
mv ./QRDETECTOR.py ~/openfb/openfb/resources/function_blocks/openCV/trackers
cp ./QRDETECTOR.fbt ~/openfb/openfb/resources/function_blocks/openCV/trackers

import logging

class QRDETECTOR:
    """ Simple FB with one algorithm """

    def __init__(self):
        # Input Variables
        self.QI = False  # Type: BOOL
        self.QUEUE_ID = ''  # Type: STRING
        self.IMG_ID = 0  # Type: UDINT

        # Output Variables
        self.QO = False  # Type: BOOL
        self.STATUS = 0  # Type: STRING
        self.VAL = 0  # Type: STRING
    def __del__(self):
        # TODO Insert your code here 
        pass 
    def service_REQ(self):
        """ Event Handler for REQ """
        logging.info("Execution logic for REQ triggered")
        # Implement internal algorithms here

    def schedule(self,IN_EVENT_NAME,EVNT_CNTR,QI,QUEUE_ID,IMG_ID): 
        self.QI= QI
        self.QUEUE_ID= QUEUE_ID
        self.IMG_ID= IMG_ID
        if IN_EVENT_NAME == "REQ":
            # TODO Insert your code here
            self.service_REQ()
            return EVNT_CNTR,self.QO,self.STATUS,self.VAL

from openfb.resources.function_blocks.openCV.globalVideoMemory import GlobalVideoMemory
...
def service_REQ(self):
    queue_id = GlobalVideoMemory.get_queue_id(self.QUEUE_ID)
    img = GlobalVideoMemory.get(queue_id=queue_id, img_id=self.IMG_ID)

import cv2
from openfb.resources.function_blocks.openCV.globalVideoMemory import GlobalVideoMemory
import logging

class QRDETECTOR():
    """ Simple FB with one algorithm """
    def __init__(self):
        # Input Variables
        self.QI = False  # Type: BOOL
        self.QUEUE_ID = ''  # Type: STRING
        self.IMG_ID = 0  # Type: UDINT

        # Output Variables
        self.QO = False  # Type: BOOL
        self.STATUS = 0  # Type: STRING
        self.VAL = ""  # Type: STRING
        self.detector = cv2.QRCodeDetector()
    def service_REQ(self):
        """ Event Handler for REQ """
        # Implement internal algorithms here        
        queue_id = GlobalVideoMemory.get_queue_id(self.QUEUE_ID)
        img = GlobalVideoMemory.get(queue_id=queue_id, img_id=self.IMG_ID)
        if img is not None:
                    data, bbox, straight_qrcode = self.detector.detectAndDecode(img)
                    if bbox is not None:
                        self.QO = True
                        self.VAL = data
                    else:
                        self.QO = False
                        self.VAL =""         	
    def schedule(self,IN_EVENT_NAME,EVNT_CNTR,QI,QUEUE_ID,IMG_ID): 
        self.QI= QI
        self.QUEUE_ID= QUEUE_ID
        self.IMG_ID= IMG_ID
        if IN_EVENT_NAME == "REQ":
            # TODO Insert your code here
            self.service_REQ()
            return EVNT_CNTR,self.QO,self.STATUS,self.VAL