AI Agent:flow:FlowSelect.py:ソースコード

from flow.FlowBase import FlowBase
from flow.FlowSequentialBase import FlowSequentialBase
from flow.FlowCriterionAndBranching  import FlowCriterionAndBranching


class FlowSelect(FlowBase):
    def __init__(self, criterion):
        super().__init__()
        self.set_category("flow")
        self.set_type("select")
        self.branching_process_dict = {}
        self.criterion_and_branching = FlowCriterionAndBranching()  # 条件による判別のみを担当
        self.criterion_and_branching.set_criterion(criterion)

    def append_branching(self, branching):
        if branching not in self.branching_process_dict.keys():
            self.branching_process_dict[branching] = FlowSequentialBase()
            self.criterion_and_branching.append_branching(branching)

    def append_process(self, process, branching):
        # 各分岐の処理内容を追加する
        if branching not in self.branching_process_dict.keys():
            self.branching_process_dict[branching] = FlowSequentialBase()
            self.criterion_and_branching.append_branching(branching)
        self.branching_process_dict[branching].append_process(process)

    def set_criterion_and_branching(self, criterion, branching_list):
        # #判断条件と分岐の内容を設定
        self.criterion_and_branching.set_criterion(criterion)
        self.criterion_and_branching.set_branching(branching_list)
        for key in branching_list:
            if key not in self.branching_process_dict.keys():
                self.branching_process_dict[key] = FlowSequentialBase()

    def set_select_type(self, type):
        self.criterion_and_branching.set_select_type(type)

    def set_threshold(self, type):
        self.criterion_and_branching.set_threshold(type)

    def run(self, command=None, pre_respons=None, flow_data=None):
        if None is not self.criterion_and_branching:
            key = self.criterion_and_branching.check()
            if None is not key:
                result = ""
                if key in self.branching_process_dict.keys():

                    result, pre_respons =\
                        self.branching_process_dict[key].run(command,
                                                             pre_respons,
                                                             key)  # is_return, result
                    print("select.run.result", result)
                    if "break" == result:
                        return "break", pre_respons
                    if "continue" == result:
                        return "continue", pre_respons
                    if "return" == result:
                        return "return", pre_respons
                    return "", pre_respons
                else:
                    print("選択肢が見つかりませんでした。", key)

            else:
                print("条件で正常に選択できませんでした。")

            return "", None
        else:
            print("条件が設定されていません。self_criterion_and_branchingで設定してください。")
            return "", None



    # ────── ここから追加 ──────
    def create_case(self, when):
        """
        `select` に case を追加する。
        :param when: 分岐キー（例: "value1", "value2"）
        Returns:
            FlowSequentialBase: case 用のシーケンシャルフロー
        """
        # case 用のシーケンシャルフローを確保
        case_flow = FlowSequentialBase()
        # 分岐リストに key を登録
        if when not in self.branching_process_dict:
            self.branching_process_dict[when] = FlowSequentialBase()
            self.criterion_and_branching.append_branching(when)

        # case_flow は self.branching_process_dict[when] と同じインスタンスを使う
        self.branching_process_dict[when] = case_flow

        # 再帰的に子ステップを構築
        return case_flow
    # ────── ここまで ──────