| | |
| | |
| | |
| | |
| |
|
| | import threading |
| |
|
| | generation_state_lock = threading.Lock() |
| | is_currently_generating = False |
| | stop_generation_requested = False |
| | temporary_files_registry = {} |
| | temporary_files_lock = threading.Lock() |
| | memory_enforcement_lock = threading.Lock() |
| | background_cleanup_thread = None |
| | background_cleanup_stop_event = threading.Event() |
| | background_cleanup_trigger_event = threading.Event() |
| | text_to_speech_manager = None |
| |
|
| | def set_text_to_speech_manager(manager_instance): |
| | global text_to_speech_manager |
| | text_to_speech_manager = manager_instance |
| |
|
| | def get_text_to_speech_manager(): |
| | global text_to_speech_manager |
| | return text_to_speech_manager |
| |
|
| | def check_if_generation_is_currently_active(): |
| | with generation_state_lock: |
| | return is_currently_generating |
| |
|
| | def set_generation_active(is_active): |
| | global is_currently_generating |
| | with generation_state_lock: |
| | is_currently_generating = is_active |
| |
|
| | def set_stop_generation_requested(requested): |
| | global stop_generation_requested |
| | with generation_state_lock: |
| | stop_generation_requested = requested |
| |
|
| | def get_stop_generation_requested(): |
| | with generation_state_lock: |
| | return stop_generation_requested |
