我们尝试使用Python开发一个极简IEC61499 运行时，主要目的是为AI Code 构筑一整套模型（也可以说是IEC61499 本体论），在这个平台上AI Code 可以自动化生成功能块，和功能能块网络（应用）。

初步的实验表明，在这样的模型下，AI 能够迅速，准确地生成和测试符合规范的功能块代码。使我们看到，将程序分解成为规范的小型模块化结构对于AI Code 是十分有益的，每个功能块完成单一的功能。便于生成和调测，也利于人类看懂和审查。

这个Python 的IEC61499 运行时与我们过去编写的C++ 代码也有很大的区别。这次实现了完全对象化设计。将运行时过去的功能，诸如读取数据，写入数据，检索事件队列，建立功能块类型实例这些功能全部写入功能块和运行时的类中。程序架构十分简单。

几种种功能块的实现

基本功能块 Basic Function Block (BFB)

基本功能块是一类同步计算的功能块，核心是ECC 状态机和Action 构成。ECC 在EventTrigger 中实现，Action 作为功能块类的内部函数。

复合功能块Composite Function Block (CFB)

这是由多个功能块通过功能块网络实现的功能块，目前没有实现

服务功能 Service Function Block (SFB)

这是与外部资源关联的功能块，大多数是异步操作的方式运行。在这个运行时中，我采纳了Python 的Thread 线程结构。

总而言之，AI Code 工程中，模块的结构尽可能规范和清晰，有助于AI 的理解和生成。多一点规范，少一点灵活性。

基本功能块基类

这是功能块的基类（Basic Class），其它功能块🈶它派生出来。在这个基类中，唯一缺少的是EventTrigger 。它是在派生的类中实现的。

class Variable:
  def __init__(self, name, datatype="BOOL", value=""):
          self.name = name
          self.datatype = datatype
          self.value = value
class Event:
  def __init__(self, name, comment=""):
          self.name = name
          self.comment = comment
class BasicFB:
  def __init__(self, Name, FBType="", Comment="",
               InputEvents=None, OutputEvents=None,
               InputVars=None, OutputVars=None, Position=None):
          self.Name = Name
          self.FBType = FBType
          self.Comment = Comment
          self.InputEvents = InputEvents if InputEvents is not None else []
          self.OutputEvents = OutputEvents if OutputEvents is not None else []
          self.InputVars = InputVars if InputVars is not None else []
          self.OutputVars = OutputVars if OutputVars is not None else []
          self.Position = Position
          self.ecc_status = "Ready"
          # 回指针：所属的功能块网络，由 FunctionBlockNetwork 在构造时回填。
          # 只有挂到网络上的实例才非 None；sendOutputEvent 依赖它把输出事件
          # 经网络转发到下游功能块。
          self.network = None
  def addInputEvent(self, event, comment=""):
      self.InputEvents.append(Event(event, comment))
  def addOutputEvent(self, event, comment=""):
      self.OutputEvents.append(Event(event, comment))
  def addInputVar(self, var, datatype="BOOL", comment="", value=""):
      # 兼容两种调用：传入已构造的 Variable，或传 (name, datatype, comment[, value])
      if isinstance(var, Variable):
          self.InputVars.append(var)
      else:
          self.InputVars.append(Variable(var, datatype, value))
  def addOutputVar(self, var, datatype="BOOL", comment="", value=""):
      if isinstance(var, Variable):
          self.OutputVars.append(var)
      else:
          self.OutputVars.append(Variable(var, datatype, value))
  def setVarValue(self, var_name, value):
         # 同时覆盖输入与输出变量：E_CTU/E_CTD 的 CV、Q 都是输出变量，
         # 内部动作要用 setVarValue 写它们；PV 是输入变量，外部要写它。
         for var in self.InputVars:
             if var.name == var_name:
                 var.value = value
                 return
         for var in self.OutputVars:
             if var.name == var_name:
                 var.value = value
                 return
  def getVarValue(self, var_name):
         for var in self.InputVars:
             if var.name == var_name:
                 return var.value
         for var in self.OutputVars:
             if var.name == var_name:
                 return var.value
         return None
  def sendOutputEvent(self, event):
         # 功能块内部产生输出事件时调用：交给功能块网络转发到下游输入事件。
         # 没挂功能块网络（单元测试里）则忽略，保证可单独实例化测试。
         if self.network is not None:
             self.network.NotifyEvent(self.Name, event)

功能块库

**E_CYCLE **

这是一个周期性定时器功能块。他内部使用了python 的线程机制（Thread），所有的

import threading
import time
from BasicFB import *
from utilities import parseTime

class E_CYCLE(BasicFB):
    def __init__(self, name, position=None):
        super().__init__(name, "E_CYCLE", "Periodic event generator", Position=position)
        self.addInputEvent("START", "Start the periodic generation of events")
        self.addInputEvent("STOP", "Stop the generation of events")
        self.addOutputEvent("EO", "Periodically triggered output event")
        self.addInputVar("DT", "TIME", "cycle time")
        self.setVarValue("DT", 1000)  # 默认周期 1000 ms
        self._running = False
        self._thread = None

    def timerThread(self):
        while self._running:
            dt = parseTime(self.getVarValue("DT"))
            time.sleep(dt / 1000.0)  # 毫秒转秒
            if self._running:
                self.sendOutputEvent("EO")  # 经网络转发 EO

    def eventTrigger(self, event):
        if event == "START":
            if not self._running:
                self._running = True
                self._thread = threading.Thread(target=self.timerThread, daemon=True)
                self._thread.start()
        elif event == "STOP":
            self._running = False  # 线程在下次循环检测到后自行退出

E_CTD

from BasicFB import *

class E_CTD(BasicFB):
    def __init__(self, name, position=None):
        super().__init__(name, "E_CTD", "Count Down Counter", Position=position)
        self.addInputEvent("CD", "Count Down")
        self.addInputEvent("LD", "Load counter value")
        self.addOutputEvent("CDO", "Count Down Output Event")
        self.addOutputEvent("LDO", "Load Output Event")
        self.addInputVar("PV", "UINT", "Preset Value")
        self.addOutputVar("Q", "BOOL", "CV <= 0")
        self.addOutputVar("CV", "UINT", "Counter Value")
        # 初值 CV=0 => Q=1：首次 EI 时 G=1，触发 LD 装入 PV，从而自举启动循环递减。
        self.setVarValue("CV", 0)
        self.setVarValue("Q", 1)

    def eventTrigger(self, event):
        match self.ecc_status:
            case "Ready":
                if event == "LD":
                    self.ecc_status = "Counting"
                    self.actionLoad()
                elif event == "CD":
                    self.ecc_status = "Counting"
                    self.actionCountDown()
            case "Counting":
                if event == "CD":
                    self.actionCountDown()
                elif event == "LD":
                    self.actionLoad()

    def actionLoad(self):
        PV = self.getVarValue("PV")
        self.setVarValue("CV", PV)
        self.setVarValue("Q", 0)
        print(f"[{self.Name}] LD -> CV={PV} (loaded)")
        self.sendOutputEvent("LDO")

    def actionCountDown(self):
        CV = self.getVarValue("CV")
        if CV > 0:
            CV -= 1
            self.setVarValue("CV", CV)
            print(f"[{self.Name}] CD -> CV={CV}")
            if CV == 0:
                self.setVarValue("Q", 1)        # 减到 0，切换信号置位
                self.sendOutputEvent("CDO")
        else:
            print(f"[{self.Name}] CD ignored, CV already 0")

E_SWITCH

from BasicFB import *

class E_SWITCH(BasicFB):
    def __init__(self, name, position=None):
        super().__init__(name, "E_SWITCH",
                         "Switching (demultiplexing) an event based on boolean input G",
                         Position=position)
        self.addInputEvent("EI", "Event Input")
        self.addOutputEvent("EO0", "Output, switched from EI when G=0")
        self.addOutputEvent("EO1", "Output, switched from EI when G=1")
        self.addInputVar("G", "BOOL", "Switch EI to EO0 when G=0, to EO1 when G=1", 0)

    def eventTrigger(self, event):
        if event == "EI":
            G = self.getVarValue("G")
            if G == 0:
                self.sendOutputEvent("EO0")
            elif G == 1:
                self.sendOutputEvent("EO1")
            # G 非法值时丢弃事件

运行时（runtime class）

from myApp import App

class runTime:
    def __init__(self, App):
        self.App = App

    def run(self):
        # 启动周期发生器；EO 会经网络转发，驱动整个功能块网络。
        self.App.TriggerEvent("E_CYCLE_1", "START")

    def stop(self):
        self.App.TriggerEvent("E_CYCLE_1", "STOP")

功能块网络基类

from FunctionBlock import *


class FunctionBlockNetwork:
     def __init__(self, Name, Comment, FunctionBlocks,
                  EventConnections=None, DataConnections=None):
         self.Name = Name
         self.Comment = Comment
         self.FunctionBlocks = FunctionBlocks
         # 连接由上层(myApp)直接分好类填入：事件连接转发事件，数据连接搬运变量值。
         # 每条连接是 (源FB名, 源引脚名, 目的FB名, 目的引脚名) 四元组。
         self.EventConnections = EventConnections if EventConnections is not None else []
         self.DataConnections = DataConnections if DataConnections is not None else []
         # 回填网络指针：功能块内部 sendOutputEvent 依赖它把输出事件转发回网络。
         for fb in self.FunctionBlocks:
             fb.network = self
         self._fbByName = {fb.Name: fb for fb in self.FunctionBlocks}

     def __repr__(self):
        FB_list = [fb.Name for fb in self.FunctionBlocks]
        ev = [f"{s}.{p}->{d}.{q}" for s, p, d, q in self.EventConnections]
        dt = [f"{s}.{p}->{d}.{q}" for s, p, d, q in self.DataConnections]
        return ('FB Network :name=' + self.Name
                + ' Function Blocks:' + str(FB_list)
                + ' EventConnections:' + str(ev)
                + ' DataConnections:' + str(dt))

     def ReadVarValue(self, FB_name, var_name):
         fb = self._fbByName.get(FB_name)
         return fb.getVarValue(var_name) if fb else None

     def WriteVarValue(self, FB_name, var_name, value):
         fb = self._fbByName.get(FB_name)
         if fb:
             fb.setVarValue(var_name, value)

     def propagateData(self, FB_name):
         for sfb, spin, dfb, dpin in self.DataConnections:
             if dfb != FB_name:
                 continue
             src = self._fbByName.get(sfb)
             dst = self._fbByName.get(dfb)
             if src is None or dst is None:
                 continue
             value = src.getVarValue(spin)
             if value is not None:
                 dst.setVarValue(dpin, value)

     def TriggerEvent(self, FB_name, event):
         fb = self._fbByName.get(FB_name)
         if fb is None:
             return
         self.propagateData(FB_name)     # 只刷新本 FB 的输入数据，再驱动 ECC
         fb.eventTrigger(event)

     def NotifyEvent(self, FB_name, event):
         """某功能块产生了输出事件：按事件连接转发到下游输入事件。"""
         for sfb, spin, dfb, dpin in self.EventConnections:
             if sfb == FB_name and spin == event:
                 self.TriggerEvent(dfb, dpin)

主程序

import sys
import termios
import tty
from App import App
from runTime import  runTime


def _read_any_key():
    try:
        fd = sys.stdin.fileno()
        old = termios.tcgetattr(fd)
    except (ValueError, termios.error):
     
        try:
            return input()
        except EOFError:
            return ""
    try:
        tty.setcbreak(fd)
        return sys.stdin.read(1)
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old)


def main():
    runtime = runTime(App())
    print("FB network loaded:", runtime.App.Name)
    print("Blocks:", [fb.Name for fb in runtime.App.FunctionBlocks])
    runtime.run()
    print("\n网络运行中，按任意键退出...")
    _read_any_key()
    runtime.stop()
    print("已停止。")


if __name__ == "__main__":
    raise SystemExit(main())

运行

python main.py

功能块开发

在这个模型的框架之下，AI 生成功能块代码显得轻松自然。这里我使用GLM 4.5 大模型做了一些实验。

提示词

生成一个PID 算法的功能块.

AI 回答

迅速完成算法，测试，并且做出项目小结。

沿用项目现有风格（from BasicFB import *、ECC eventTrigger + action*、sendOutputEvent），实现了一个位置式 PID 控制器：

事件接口（IEC 61499 风格）

• INIT → INITO：清零内部状态（积分项、历史误差）

• REQ → CNF：周期计算，读 SP/PV 算 OUT

• RST → CNF：仅复位积分项，保留增益参数

变量

• 输入：SP、PV、Kp、Ki、Kd、Ts（采样周期）、AutoMan（自动/手动）、ManualIn、OutMin、OutMax

• 输出：OUT、Error、Integral、Derivative

控制逻辑

• u = Kp·e + Ki·∫e + Kd·de/dt，积分用 error·Ts 累加，微分用 (e - e_last)/Ts

• 抗积分饱和：积分项本身也限幅，避免长期饱和后恢复迟滞

• 输出限幅：_clamp 兜底，并处理上下限配错的情形

• 手动模式（AutoMan=0）：直通 ManualIn 并同步误差显示，不更新积分

• Ts=0 时退化为纯 PD，避免除零

测试覆盖了自动连续计算、手动切换、复位三个路径，输出符合预期（首拍 P=100/I=25/D=50→限幅 100，稳态后微分归零）。

需要我再加一个对应的单元测试文件，或改成增量式 PID（输出 ΔU）吗？