Python Script Integration

你要做什么？
├── C++ 调用 Python 函数 → 场景 A：Python 模块导入与函数调用
├── Python 后台线程处理数据，不阻塞 UI → 场景 B：Python 多线程
├── Python 线程向 C++ 主线程推送数据 → 场景 C：跨线程数据回传
├── Python 脚本访问 C++ 数据管理器 → 场景 D：Python 访问 C++ 数据
├── Python 脚本弹出 UI 对话框 → 场景 E：Python 调用 UI
├── Python 脚本向日志窗口输出 → 场景 F：Python 日志通道
└── Python 脚本部署与更新 → 场景 G：脚本部署

┌──────── C++ 主线程（Qt 事件循环） ────────┐
│                                           │
│  MyWorker (C++ Worker)                    │
│    ├ DAPyModule m_pyModule                 │
│    │   └ import("MyPackage")              │
│    ├ DAPyModule m_pyDataAnalysisModule      │
│    │   └ = m_pyModule.attr("data_analysis")│
│    ├ DAPyModule m_threadStatusMgrModule     │
│    │   └ = DAWorkbench.DAPyBase             │
│    │      .attr("thread_status_manager")    │
│    │                                        │
│    ├ 调用 Python 函数 (同步)                │
│    │   └ m_pyDataAnalysisModule->attr(     │
│    │      "process_zip_data_thread")(path)  │
│    │      → 返回 taskid                     │
│    │                                        │
│    └ QTimer::singleShot 轮询线程状态        │
│        └ get_task_status(taskid)           │
│            → 返回 dict {is_running, progress, message, ...} │
│                                           │
└───────────────────────────────────────────┘
          ↑ callInMainThread          ↑ QTimer 轮询
          │ (Qt 信号槽跨线程)           │ (C++ 主动查询)
┌───────── Python 后台线程 ──────────────────┐
│                                           │
│  threading.Thread(target=process_zip_data)│
│    ├ 处理数据（pandas/numpy）              │
│    ├ status.update_progress(p, msg)        │
│    ├ status.update_custom_data(k, v)       │
│    ├ status.finish(success, msg)           │
│    └ 回调函数: signal_handler              │
│        .callInMainThread(add_data_func)    │
│        → 在主线程中操作 DataManager         │
│                                           │
└───────────────────────────────────────────┘

#include "DAPyModule.h"
#include "DAPybind11InQt.h"  // 必须在所有 pybind11 头文件之前
#include "DAPybind11QtCaster.hpp"

bool MyWorker::initializePythonEnv()
{
    try {
        // 1. 导入插件自己的 Python 包
        m_pyModule = std::make_unique<DA::DAPyModule>();
        if (!m_pyModule->import("MyPackage")) {
            daCritical << "Failed to import MyPackage";
            return false;
        }

        // 2. 获取子模块（包内的 .py 文件）
        m_pyDataModule = std::make_unique<DA::DAPyModule>();
        *m_pyDataModule = m_pyModule->attr("data_analysis");

        // 3. 导入平台提供的线程状态管理器
        DA::DAPyModule daWorkbench("DAWorkbench.DAPyBase");
        m_threadStatusMgr = std::make_unique<DA::DAPyModule>();
        *m_threadStatusMgr = daWorkbench.attr("thread_status_manager");

        m_isPythonValid = true;
        return true;
    } catch (const std::exception& e) {
        m_pyModule.reset();
        m_pyDataModule.reset();
        qCritical() << e.what();
        return false;
    }
}

#include "DAPybind11QtCaster.hpp"  // Qt ↔ Python 类型转换

void MyWorker::callPythonFunction()
{
    try {
        // 获取函数对象
        auto func = m_pyDataModule->attr("my_function");
        if (func.is_none()) {
            qCritical() << "function not found";
            return;
        }

        // 传递 QString 参数（DAPybind11QtCaster 自动转换）
        auto arg = DA::PY::toPyObject(QString("some_path"));
        auto result = func(arg);

        // 获取返回值
        std::string taskid = result.cast<std::string>();
        m_taskID = taskid;

    } catch (const std::exception& e) {
        qCritical() << e.what();
    }
}

PyScripts/
└── MyPackage/              ← 包名（C++ import 的名字）
    ├── __init__.py         ← 包入口，导出子模块
    ├── data_analysis.py    ← 数据分析脚本
    └── utils.py            ← 工具函数

# __init__.py
from . import data_analysis, utils

# data_analysis.py
def my_function(path: str) -> str:
    """C++ 通过 attr("my_function") 调用此函数"""
    import pandas as pd
    df = pd.read_csv(path)
    return f"processed_{len(df)}_rows"

// QString → pybind11::object
auto pyStr = DA::PY::toPyObject(QString("hello"));

// pybind11::dict → QVariant
QVariant var = DA::PY::fromPyVariant(pyDict);

// 判断能否转为 QDateTime
bool canConvert = DA::PY::canCastToQDateTime(pyObj);

# data_analysis.py
import threading
import DAWorkbench.DAPyBase.thread_status_manager as tsm

def process_zip_data_thread(zip_path: str) -> str:
    """
    主入口：启动后台线程处理数据
    返回: taskid（字符串），空字符串表示启动失败
    """
    try:
        # 创建任务状态追踪器
        taskid, status = tsm.create_task_with_status("process zip data")

        # 启动后台线程
        thread = threading.Thread(
            target=_process_data_internal,
            args=(zip_path, status),
            daemon=True
        )
        thread.start()
        return taskid
    except Exception as e:
        logger.error(f"启动线程失败: {e}")
        return ""

def _process_data_internal(zip_path: str, status: tsm.ProcessingStatus):
    """后台线程的实际处理函数"""
    try:
        status.start()
        status.update_progress(0, "开始处理")
        status.update_custom_data("zip_path", zip_path)

        # ... 数据处理逻辑 ...
        for i, file in enumerate(files):
            status.update_progress(
                (i / len(files)) * 100,
                f"处理文件 {i}/{len(files)}: {file}"
            )
            # ... 处理每个文件 ...

        status.update_custom_data("result_count", len(results))
        status.finish(True, f"完成，共处理 {len(results)} 个文件")
    except Exception as e:
        status.finish(False, f"失败: {e}")

void MyWorker::startBackgroundTask()
{
    if (!isPythonValid()) return;

    try {
        // 调用 Python 函数启动后台线程
        auto func = m_pyDataModule->attr("process_zip_data_thread");
        auto arg = DA::PY::toPyObject(m_zipPath);
        auto result = func(arg);

        if (result.is_none()) return;

        m_taskID = result.cast<std::string>();

        // 显示进度条
        DA::DAStatusBarInterface* statusBar = m_ui->getStatusBar();
        statusBar->showProgressBar();
        statusBar->setProgressText(QString(u8"正在处理..."));

        // 启动轮询
        checkTaskStatus();
    } catch (const std::exception& e) {
        qCritical() << e.what();
    }
}

void MyWorker::checkTaskStatus()
{
    try {
        // 释放 GIL，让 Python 后台线程能运行
        {
            pybind11::gil_scoped_release release;
            QThread::sleep(1);  // 短暂休眠
        }

        // 调用 Python 获取任务状态
        auto get_status = m_threadStatusMgr->attr("get_task_status");
        pybind11::dict status = get_status(m_taskID);

        bool is_running = status["is_running"].cast<bool>();

        if (is_running) {
            // 更新进度
            double progress = status["progress"].cast<double>();
            std::string message = status["message"].cast<std::string>();
            double elapsed = status["elapsed_seconds"].cast<double>();

            DA::DAStatusBarInterface* statusBar = m_ui->getStatusBar();
            statusBar->setProgress(progress);
            statusBar->setProgressText(
                QString::fromStdString(message) +
                QString(u8" 已用时: %1:%2")
                    .arg(static_cast<int>(elapsed) / 60, 2, 10, QChar('0'))
                    .arg(static_cast<int>(elapsed) % 60, 2, 10, QChar('0'))
            );

            // 继续轮询（20ms 后再次检查）
            QTimer::singleShot(20, this, &MyWorker::checkTaskStatus);
        } else {
            // 任务完成
            bool is_success = status["is_success"].cast<bool>();
            QString msg = QString::fromStdString(status["message"].cast<std::string>());

            DA::DAStatusBarInterface* statusBar = m_ui->getStatusBar();
            statusBar->hideProgressBar();
            statusBar->setBusy(false);

            if (is_success) {
                statusBar->showMessage(QString(u8"完成!"));
                // 获取自定义数据
                pybind11::dict custom_data = status["custom_data"];
                QString zip_path = QString::fromStdString(
                    custom_data["zip_path"].cast<std::string>());
                // 处理完成后的逻辑
            } else {
                statusBar->showMessage(QString(u8"失败! ") + msg);
            }

            QApplication::processEvents();
        }
    } catch (const std::exception& e) {
        qCritical() << e.what();
    }
}

# 创建任务
taskid, status = tsm.create_task_with_status("task name")

# 状态控制
status.start()                    # 开始
status.update_progress(50, "处理中")  # 更新进度
status.pause()                    # 暂停
status.resume()                   # 恢复
status.cancel("用户取消")           # 取消
status.finish(True, "完成消息")     # 完成（成功/失败）

# 自定义数据
status.update_custom_data("key", value)
value = status.get_custom_data("key", default)

# 查询
info = status.get_status()        # 返回状态字典
is_active = status.is_active()    # 是否活跃
is_finished = status.is_finished() # 是否已结束

# data_analysis.py
import da_app, da_data

def process_zip_data_thread(zip_path: str) -> str:
    taskid, status = tsm.create_task_with_status("process zip data")

    def internal_callback(result_dict):
        """后台线程完成后的回调，通过 callInMainThread 在主线程写入数据"""
        if result_dict is not None:
            signal_handler = da_app.getCore().getPythonSignalHandler()
            if signal_handler:
                def add_data_in_main_thread():
                    """此函数在 Qt 主线程中执行"""
                    datamanager = da_app.getCore().getDataManagerInterface()
                    for name, df in result_dict.items():
                        data = da_data.DAData(df)
                        data.setName(name)
                        data.setDescribe(name)
                        datamanager.addData(data)

                signal_handler.callInMainThread(add_data_in_main_thread)

    thread = threading.Thread(
        target=_process_data,
        args=(zip_path, internal_callback, status),
        daemon=True
    )
    thread.start()
    return taskid

// 在 Worker 初始化时连接数据添加信号
void MyWorker::initialize(DA::DACoreInterface* core, MyUI* ui)
{
    m_core = core;
    m_ui = core->getUiInterface();

    DA::DADataManagerInterface* datamgr = m_core->getDataManagerInterface();
    // 连接数据添加信号
    connect(datamgr, &DA::DADataManagerInterface::dataAdded,
            this, &MyWorker::onDataAdded);
}

void MyWorker::onDataAdded()
{
    // 数据已通过 Python 线程 → callInMainThread → DataManager::addData 添加
    // 在这里重新分析数据
    analysisDatas();
    Q_EMIT datasAdded();  // 通知 UI 更新
}

Python 后台线程                     C++ 主线程（Qt 事件循环）
    │                                     │
    ├ signal_handler.callInMainThread(    │
    │   lambda: add_data())               │
    │   └ 函数包装为 FunctionWrapper       │
    │   └ 存入 m_functionMap (mutex)       │
    │   └ emit executeRequested(id) ──────┼──→ Qt 跨线程队列连接
    │                                     │
    │   (Python 线程继续执行)              ├ onExecuteRequested(id)
    │                                     │   └ 从 map 取出 FunctionWrapper
    │                                     │   └ 执行函数（操作 DataManager）
    │                                     │

import da_app, da_data

def my_python_function():
    """Python 脚本中访问 C++ DataManager"""
    datamanager = da_app.getCore().getDataManagerInterface()

    # 按通配符查找数据表
    datas = datamanager.findDatas("*module*")
    for d in datas:
        name = d.getName()
        df = d.toDataFrame()  # 转为 pandas DataFrame
        # 处理 DataFrame...

    # 获取特定名称的数据
    datas = datamanager.findDatas("*module_8")
    if datas:
        df = datas[0].toDataFrame()

import da_app, da_data
import pandas as pd

def write_data_to_manager():
    """将 DataFrame 写入 DataManager（必须在主线程调用）"""
    signal_handler = da_app.getCore().getPythonSignalHandler()
    signal_handler.callInMainThread(lambda: _do_write())

def _do_write():
    datamanager = da_app.getCore().getDataManagerInterface()
    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
    data = da_data.DAData(df)
    data.setName("my_data_table")
    data.setDescribe("description")
    datamanager.addData(data)

import da_data

# DAData 包装 pandas DataFrame
data = da_data.DAData(df)    # 从 DataFrame 创建
df = data.toDataFrame()      # 转回 DataFrame
data.setName("name")        # 设置名称
data.setDescribe("desc")    # 设置描述
name = data.getName()        # 获取名称
is_null = data.isNull()     # 是否为空
data_id = data.id()          # 获取 ID

import da_app

def my_function():
    ui = da_app.getCore().getUiInterface()

    # 选择文件夹
    path = ui.getExistingDirectory("选择保存路径")
    if len(path) == 0:
        return  # 用户取消

    # 选择文件
    # file_path = ui.getOpenFileName("选择文件", "CSV (*.csv)")

from DAWorkbench.DAPyBase.form_builder import FormBuilder, option

def show_config_dialog():
    ui = da_app.getCore().getUiInterface()

    config = (FormBuilder("参数设置")
        .group("basic", "基本设置")
        .enum("machine", label="机型",
              default="GMV-224WM/S",
              options=[option("GMV-224WM/S", "GMV-224WM/S"),
                       option("GMV-252WM/S", "GMV-252WM/S")],
              description="选择机型")
        .bool("save_word", label="是否生成报告",
              default=True,
              description="勾选将生成 Word 报告")
        .end_group()
        .build()
    )

    result = ui.getConfigValues(config, "my_config_cache_key")
    if len(result) == 0:
        return  # 用户取消

    machine = result["machine"]
    save_word = result["save_word"]