monitory-scenario-simulator/utils.py at develop · Fac2Real/monitory-scenario-simulator · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
# utils.py
import asyncio
import streamlit as st
import traceback

def run_async_in_thread(loop, coro):
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(coro)
    except Exception as e:
        try:
            st.error(f"비동기 스레드 오류: {e}")
        except Exception: # st.error가 실패할 수 있는 경우 (예: 앱 종료 중)
            pass
        print(f"비동기 스레드 오류: {e}") # 디버깅을 위해 콘솔에 출력
        traceback.print_exc()
    finally:
        print(f"Async thread loop for {str(coro)} processing finished.")


async def start_scenario_task(scenario_obj):
    """
    단일 시나리오 객체의 current_state 및 stop_event를 기반으로 실행 루프를 관리합니다.
    또한 전환이 진행 중인지 확인하기 위해 transition_in_progress_event를 사용합니다.
    """
    scenario_name = scenario_obj.__class__.__name__
    # 객체에 고유한 경로 속성이 있는 경우 식별자 추가 (ZoneScenario, EquipScenario 모두 해당)
    if hasattr(scenario_obj, 'normal_path') and scenario_obj.normal_path:
        try:
            scenario_name += f" ({scenario_obj.normal_path.split(r'\\')[-1]})" # 원시 문자열 경로 분할
        except: # 경로가 예상과 다를 경우를 대비한 예외 처리
             scenario_name += " (path error)"


    print(f"APP: [{scenario_name}] 관리자 작업 시작 중")

    # 초기 시작 시 상태를 "normal"로 설정하고 stop_event를 해제합니다.
    # transition_in_progress_event도 "set"(전환 중 아님)으로 설정되어 있는지 확인합니다.
    if scenario_obj.current_state == "stopped": # 새로운 시작일 때만
        scenario_obj.current_state = "normal"
        if hasattr(scenario_obj, 'stop_event'): # stop_event가 있는지 확인
            scenario_obj.stop_event.clear()
        if hasattr(scenario_obj, 'transition_in_progress_event'): # transition_in_progress_event가 있는지 확인
            scenario_obj.transition_in_progress_event.set() # 자유로운 상태로 설정

    while hasattr(scenario_obj, 'stop_event') and not scenario_obj.stop_event.is_set():
        # 새 안정 상태 루프를 실행하기 전에 진행 중인 전환이 완료될 때까지 기다립니다.
        if hasattr(scenario_obj, 'transition_in_progress_event'):
            # print(f"APP: [{scenario_name}] '{scenario_obj.current_state}' 상태 루프 전에 전환 이벤트 대기 중 (is_set={scenario_obj.transition_in_progress_event.is_set()})")
            await scenario_obj.transition_in_progress_event.wait() # 이벤트가 clear()되면 여기서 차단됨
            # print(f"APP: [{scenario_name}] 전환 이벤트 설정됨. '{scenario_obj.current_state}' 상태로 진행")
        else: # transition_in_progress_event가 없는 경우 (예: EquipScenario가 아직 업데이트되지 않은 경우)
            print(f"APP: [{scenario_name}] transition_in_progress_event 없음. 직접 진행.")


        if scenario_obj.stop_event.is_set(): # 대기 후 다시 중지 이벤트 확인
            print(f"APP: [{scenario_name}] 전환 대기 후 중지 이벤트 감지. 종료 중.")
            break

        current_loop_state = scenario_obj.current_state

        if current_loop_state == "normal":
            # print(f"APP: [{scenario_name}] NORMAL 데이터 루프 진입 ({scenario_obj.normal_path})")
            await scenario_obj.execute_scenario_loop(scenario_obj.normal_path, "normal")
        elif current_loop_state == "warn":
            # print(f"APP: [{scenario_name}] WARN 데이터 루프 진입 ({scenario_obj.warn_path})")
            await scenario_obj.execute_scenario_loop(scenario_obj.warn_path, "warn")
        elif current_loop_state == "stopped":
            print(f"APP: [{scenario_name}] 상태가 'stopped'입니다. 관리자 작업 종료 중.")
            break
        else:
            print(f"APP: [{scenario_name}] 알 수 없는 상태 '{current_loop_state}'. 0.1초 대기.")
            await asyncio.sleep(0.1) # 알 수 없는 상태에서 무한 루프 방지

        # execute_scenario_loop가 반환되면 다음 중 하나 때문일 수 있습니다:
        # 1. 외부 상태 변경 (current_state != 실행 중이던 state_parameter)
        # 2. stop_event 설정됨
        # 3. transition_in_progress_event가 해제됨 (새 전환 시작 신호)
        # 외부 루프가 다시 평가하고, await transition_in_progress_event.wait()가 #3을 처리합니다.
        # print(f"APP: [{scenario_name}] '{current_loop_state}'에 대한 주 루프 종료됨. 다음 작업 확인 중.")
        await asyncio.sleep(0.01) # 다른 작업(예: UI 업데이트)을 위한 약간의 양보

    print(f"APP: [{scenario_name}] 관리자 작업 종료됨.")


async def change_scenario_state_task(scenario_obj, new_state):
    """
    전환을 처리하고 시나리오 객체의 current_state를 업데이트합니다.
    이 시나리오 객체에 대해 실행 중인 start_scenario_task가 새 상태를 선택합니다.
    """
    scenario_name = scenario_obj.__class__.__name__
    if hasattr(scenario_obj, 'normal_path') and scenario_obj.normal_path:
        try:
            scenario_name += f" ({scenario_obj.normal_path.split(r'\\')[-1]})"
        except:
            scenario_name += " (path error)"

    previous_internal_state = scenario_obj.current_state
    # print(f"APP: [{scenario_name}] 상태를 '{previous_internal_state}'에서 '{new_state}'(으)로 변경 요청")

    # transition_in_progress_event가 있고 설정되어 있는지 확인 (전환 중이 아님)
    can_transition = True
    if hasattr(scenario_obj, 'transition_in_progress_event'):
        if not scenario_obj.transition_in_progress_event.is_set():
            can_transition = False
            print(f"APP: [{scenario_name}] 전환이 이미 진행 중이므로 '{new_state}'(으)로의 상태 변경 건너뜀.")

    if not can_transition and new_state != "stop": # 중지 요청은 항상 진행되어야 함
        return


    if new_state == "warn":
        if previous_internal_state != "warn":
            # print(f"APP: [{scenario_name}] 'normal_to_warn' 전환 실행 중.")
            await scenario_obj.handle_state_change_to_warn() # 전환 데이터를 재생하고 이벤트를 설정/해제
            scenario_obj.current_state = "warn" # 중요: 상태 업데이트
            # print(f"APP: [{scenario_name}] 상태가 공식적으로 'warn'으로 설정됨.")
        # else:
            # print(f"APP: [{scenario_name}] 이미 'warn' 상태이거나 해당 상태로 전환 중입니다. 조치 없음.")
    elif new_state == "normal":
        if previous_internal_state != "normal":
            # print(f"APP: [{scenario_name}] 'warn_to_normal' 전환 실행 중.")
            await scenario_obj.handle_state_change_to_normal() # 전환 데이터를 재생하고 이벤트를 설정/해제
            scenario_obj.current_state = "normal" # 중요: 상태 업데이트
            # print(f"APP: [{scenario_name}] 상태가 공식적으로 'normal'으로 설정됨.")
        # else:
            # print(f"APP: [{scenario_name}] 이미 'normal' 상태이거나 해당 상태로 전환 중입니다. 조치 없음.")
    elif new_state == "stop":
        # print(f"APP: [{scenario_name}] 'stop' 핸들러 실행 중.")
        await scenario_obj.handle_state_change_to_stop() # stop_event 및 current_state = "stopped" 설정
        # print(f"APP: [{scenario_name}] 'stop' 핸들러 완료됨.")
    # else:
        # print(f"APP: [{scenario_name}] 알 수 없는 new_state '{new_state}' 요청됨.")


async def run_all_scenarios_async_explicit(scenario_objects_with_paths):
    # scenario_objects_with_paths는 (scenario_obj, initial_path) 튜플의 리스트입니다.
    # initial_path는 여기서는 사용되지 않지만, 함수 시그니처 일관성을 위해 유지할 수 있습니다.
    tasks = []
    for sc_obj, _ in scenario_objects_with_paths: # initial_path는 무시됨
        tasks.append(start_scenario_task(sc_obj))
    await asyncio.gather(*tasks)

async def change_all_states_async_explicit(scenario_objects, new_state):
    tasks = []
    for sc_obj in scenario_objects:
        tasks.append(change_scenario_state_task(sc_obj, new_state))
    await asyncio.gather(*tasks)