Merge pull request #136 from RedDeadDepresso/master

Added some features and refactoring

core/Baas_thread.py

@@ -2,10 +2,11 @@
 from datetime import datetime
 import cv2
 from core.exception import ScriptError
-from core.notification import notify
+from core.notification import notify, toast
 from core.scheduler import Scheduler
 from core import position, picture
 from core.utils import Logger
+from device_operation import process_api
 
 from core.pushkit import push
 import numpy as np
@@ -58,7 +59,7 @@
 
 
 class Baas_thread:
-    def __init__(self, config, logger_signal=None, button_signal=None, update_signal=None):
+    def __init__(self, config, logger_signal=None, button_signal=None, update_signal=None, exit_signal=None):
         self.dailyGameActivity = None
         self.activity_name = None
         self.config_set = config
@@ -91,6 +92,7 @@ def __init__(self, config, logger_signal=None, button_signal=None, update_signal
         self.total_assault_difficulty_names = ["NORMAL", "HARD", "VERYHARD", "HARDCORE", "EXTREME", "INSANE", "TORMENT"]
         self.button_signal = button_signal
         self.update_signal = update_signal
+        self.exit_signal = exit_signal
         self.stage_data = {}
         self.activity_name = None
 
@@ -196,6 +198,16 @@ def check_process_running(self, process_name):
             if proc.info['name'] == process_name:
                 return True
         return False
+
+    def terminate_process(self, process_name):
+        """
+        终止指定名称的进程
+        """
+        for proc in psutil.process_iter(['pid', 'name']):
+            if proc.info['name'] == process_name:
+                proc.terminate()
+                return True
+        return False
 
     def check_process_running_from_pid(self, pid):
         # 代码来源于BAAH
@@ -242,16 +254,14 @@ def start_check_emulator_stat(self, emulator_strat_stat, wait_time):
                     return False
                 return True
             else:
-                self.lnk_path = self.config.get("program_address")
-                self.convert_lnk_to_exe(self.lnk_path)
                 self.file_path = self.config.get("program_address")
                 self.process_name = self.extract_filename_and_extension(self.file_path)
-                if self.check_process_running(self.process_name):
+                if process_api.is_running(self.file_path) is not None:
                     self.logger.info(f"-- Emulator Process {self.process_name} is running --")
                     return True
                 else:
                     self.logger.warning(f"-- Emulator Process {self.process_name} is not running, start Emulator --")
-                    subprocess.Popen(self.file_path)
+                    process_api.start(self.file_path)
                     self.logger.info(f" Start wait {wait_time} seconds for emulator to start. ")
                     while self.flag_run:
                         time.sleep(0.01)
@@ -260,7 +270,7 @@ def start_check_emulator_stat(self, emulator_strat_stat, wait_time):
                             break
                     else:
                         return False
-                    if self.check_process_running(self.process_name):
+                    if process_api.is_running(self.file_path) is not None:
                         self.logger.info(f"Emulator Process {self.process_name} started SUCCESSFULLY")
                         return True
                     else:
@@ -372,6 +382,8 @@ def thread_starter(self):
                         self.task_finish_to_main_page = False
                     self.scheduler.update_valid_task_queue()
                     time.sleep(1)
+                    if self.flag_run: # allow user to stop script before then action
+                        self.handle_then()
         except Exception as e:
             notify(title='', body='任务已停止')
             self.logger.error(e.__str__())
@@ -706,3 +718,60 @@ def refresh_hard_tasks(self):
             except Exception as e:
                 self.logger.error(e.__str__())
         self.config_set.set("unfinished_hard_tasks", self.config['unfinished_hard_tasks'])
+
+    def handle_then(self):
+        action = self.config_set["then"]
+        if action == '无动作' or not self.scheduler.is_wait_long(): # Do Nothing 
+            return 
+        elif action == '退出 Baas': # Exit Baas
+            self.exit_baas()
+        elif action == '退出 模拟器': # Exit Emulator
+            self.exit_emulator()
+        elif action == '退出 Baas 和 模拟器':  # Exit Baas and Emulator
+            self.exit_emulator()
+            self.exit_baas()
+        elif action == '关机': # Shutdown
+            self.shutdown()
+        self.signal_stop() # avoid rerunning then action in case of error
+
+    def exit_emulator(self):
+        self.logger.info(f"-- BAAS Exit Emulator --")
+        if self.config['emulatorIsMultiInstance']:
+            name = self.config.get("multiEmulatorName")
+            num = self.config.get("emulatorMultiInstanceNumber")
+            self.logger.info(f"-- Exit Multi Emulator --")
+            self.logger.info(f"EmulatorName: {name}")
+            self.logger.info(f"MultiInstanceNumber: {num}")
+            device_operation.stop_simulator_classic(name, num)
+        else:
+            self.file_path = self.config.get("program_address")
+            if not process_api.terminate(self.file_path):
+                self.logger.error("Emulator exit failed")
+                return False
+        return True
+
+    def exit_baas(self):
+        if self.exit_signal is not None:
+            self.exit_signal.emit(0)        
+
+    def shutdown(self):
+        try:
+            self.start_shutdown()
+            answer = toast(title='BAAS: Cancel Shutdown?',
+                            body='All tasks have been completed: shutting down. Do you want to cancel?',
+                            button='Cancel',
+                            duration='long'
+            )
+            # cancel button pressed
+            if answer == {'arguments': 'http:Cancel', 'user_input': {}}:
+                self.cancel_shutdown()
+        except:
+            self.logger.error("Failed to shutdown. It may be due to a lack of administrator privileges.")
+
+    def start_shutdown(self):
+        self.logger.info("Running shutdown")
+        subprocess.run(["shutdown", "-s", "-t", "60"])
+
+    def cancel_shutdown(self):
+        self.logger.info("Shutdown cancelled")
+        subprocess.run(["shutdown", "-a"])

core/default_config.py

@@ -281,10 +281,11 @@
     "name": "新的配置",
     "purchase_arena_ticket_times": "0",
     "screenshot_interval": "0.3",
+    "autostart": false,
+    "then": "无动作",
     "program_address": "None",
     "open_emulator_stat": false,
     "emulator_wait_time": "180",
-    "multi_emulator_check": false,
     "ArenaLevelDiff": 0,
     "maxArenaRefreshTimes": 10,
     "createPriority": "花>Mo>情人节>果冻>色彩>灿烂>光芒>玲珑>白金>黄金>铜>白银>金属>隐然",

core/notification.py

@@ -1,9 +1,12 @@
 import os
+
 # Check if the current OS is Windows
 try:
-    from win10toast import ToastNotifier as _notify
+    from win11toast import notify as _notify
+    from win11toast import toast as _toast
 except ImportError:
     _notify = None
+    _toast = None
 app_id = 'BlueArchiveAutoScript.exe'
 icon_path = '/gui/assets/logo.png'
 
@@ -28,3 +31,15 @@ def notify(title=None, body=None):
         app_id='BlueArchiveAutoScript.exe',
         icon=root_path + icon_path,
     )
+
+def toast(title=None, body=None, button=None, duration=None):
+    root_path = get_root_path()
+    return _toast(
+        title=title,
+        body=body,
+        app_id='BlueArchiveAutoScript.exe',
+        icon=root_path + icon_path,
+        button=button, 
+        duration=duration
+        )
+

core/scheduler.py

@@ -32,11 +32,8 @@ def _read_config(self):
                         self.event_map[item['func_name']] = item['event_name']
 
     def _commit_change(self):
-        """event_config只能被switch修改,调度时在内存中操作"""
         with open(self.event_config_path, 'w', encoding='utf-8') as f:
             json.dump(self._event_config, f, ensure_ascii=False, indent=2)
-        # with open(self._display_config_path, 'w', encoding='utf-8') as f:
-        #     json.dump(self._display_config, f, ensure_ascii=False, indent=2)
 
     @classmethod
     def get_next_time(cls, hour, minute, second):
@@ -94,21 +91,16 @@ def heartbeat(self):
 
     def update_valid_task_queue(self):
         self._read_config()
-        _valid_event = [x for x in self._event_config if x['enabled']]                                      # filter out disabled event
-        _valid_event = [x for x in self._event_config if x['enabled'] and x['next_tick'] <= time.time()]    # filter out event not ready
+        time_since_epoch = time.time()
+        now = datetime.now()
+        time_since_midnight = self.convert_to_seconds(now.hour, now.minute, now.second)
+
+        _valid_event = [x for x in self._event_config if x['enabled'] and x['next_tick'] <= time_since_epoch and \
+                        not self.is_disable_period(x, time_since_midnight)]    # filter out event not ready
         _valid_event = sorted(_valid_event, key=lambda x: x['priority'])                                    # sort by priority
-        current_time = datetime.now(timezone.utc).timestamp() % 86400
+
         self._valid_task_queue = []
         for i in range(0, len(_valid_event)):
-            f = True
-            for j in range(0, len(_valid_event[i]["disabled_time_range"])):  # current task not in disable time range
-                start = _valid_event[i]["disabled_time_range"][j][0][0] * 3600 + _valid_event[i]["disabled_time_range"][j][0][1] * 60 + _valid_event[i]["disabled_time_range"][j][0][2]
-                end = _valid_event[i]["disabled_time_range"][j][1][0] * 3600 + _valid_event[i]["disabled_time_range"][j][1][1] * 60 + _valid_event[i]["disabled_time_range"][j][1][2]
-                if start <= current_time <= end:
-                    f = False
-                    break
-            if not f:
-                continue
             self._waitingTaskDisplayQueue.append(_valid_event[i]['event_name'])
             thisTask = {
                 "pre_task": [],
@@ -129,9 +121,40 @@ def update_valid_task_queue(self):
             thisTask["post_task"] = temp
             self._valid_task_queue.append(thisTask)
 
+    def convert_to_seconds(self, hour, minute, second) -> float:
+        return hour * 3600 + minute * 60 + second
+
+    def is_disable_period(self, event_list, time_since_midnight) -> bool:
+        disabled = event_list["disabled_time_range"]
+        for period in disabled:
+            start = self.convert_to_seconds(*period[0])
+            end = self.convert_to_seconds(*period[1])
+            if start <= time_since_midnight <= end:
+                return True
+        return False
+
+    def is_wait_long(self) -> bool:
+        """
+        allows tactical challenge to be fully completed and triggers then action
+        by determining whether the wait is less than 2 minutes
+        """
+        time_since_epoch = time.time()
+        now = datetime.now()
+        time_since_midnight = self.convert_to_seconds(now.hour, now.minute, now.second)
+
+        _valid_event = [x for x in self._event_config if x['enabled'] and \
+                         x['next_tick'] > time_since_epoch and \
+                            not self.is_disable_period(x, time_since_midnight)]
+        if _valid_event:
+            event_list = min(_valid_event, key=lambda x: x['next_tick'])
+            next_tick = event_list['next_tick']
+            difference = next_tick - time_since_epoch
+            return difference > 120
+
+        return True
+
     def getWaitingTaskList(self):
         return self._waitingTaskDisplayQueue
 
     def getCurrentTaskName(self):
         return self._currentTaskDisplay
-

device_operation/mumu_manager_api.py

@@ -5,10 +5,15 @@
 from .get_adb_address import get_simulator_port
 
 
-def mumu12_control_api_backend(multi_instance_number=0, operation="start"):
+def mumu12_control_api_backend(simulator_type, multi_instance_number=0, operation="start"):
     # 读取注册表中的键值
-    key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
-                         r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\MuMuPlayer-12.0")
+    if simulator_type == "mumu":
+        key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
+                            r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\MuMuPlayer-12.0")
+    else:
+        key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
+                            r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\MuMuPlayerGlobal-12.0")
+
     icon_path, _ = winreg.QueryValueEx(key, "DisplayIcon")
     icon_path = icon_path.replace('"', '') # 去除获取到的文件位置两侧的双引号
     winreg.CloseKey(key)

device_operation/process_api.py

@@ -0,0 +1,74 @@
+import os
+import psutil
+import pythoncom
+import subprocess
+import win32com.client
+
+from typing import Union
+
+
+def match_lists(target_args: list[str], process_cmdline: list[str]):
+    # target_args and process_cmdline have different format for their strings, so this method check if they're equals
+    for string in target_args:
+        cleaned_str = string.strip('"\'')
+        if cleaned_str not in process_cmdline:
+            return False
+    return True
+
+def extract_args(emulator_path: str) -> tuple[str, list[str]]:
+    # Resolve the emulator's actual executable path from a shortcut if provided
+    if emulator_path.endswith('.lnk'):
+        shell = win32com.client.Dispatch("WScript.Shell")
+        shortcut = shell.CreateShortCut(emulator_path)
+        emulator_path = os.path.abspath(shortcut.Targetpath)
+
+        # Extract command line arguments from the shortcut
+        arguments = shortcut.Arguments
+        arguments = arguments.split(" ")
+        if arguments == [""]:
+            arguments = None
+    else:
+        # No .lnk file, so there are no additional arguments
+        arguments = None
+
+    return emulator_path, arguments
+
+def start(emulator_path: str):
+    if os.path.isfile(emulator_path):
+        directory = os.path.dirname(emulator_path)
+        path, args = extract_args(emulator_path)
+        subprocess.Popen([path] + args, shell=True, cwd=directory)
+
+def is_running(emulator_path) -> Union[psutil.Process, None]:
+    # Initialize the COM library
+    pythoncom.CoInitialize()
+
+    emulator_path, target_args = extract_args(emulator_path)
+
+    # Iterate over all running processes
+    for proc in psutil.process_iter(attrs=['pid', 'name', 'exe', 'cmdline']):
+        try:
+            process_info = proc.info
+            process_exe = process_info.get('exe')  # Use get to handle potential None values
+
+            if process_exe is not None:
+                process_cmdline = process_info.get('cmdline', '')
+                # Compare the executable path and command line arguments
+                if (os.path.normcase(process_exe) == os.path.normcase(emulator_path)):
+                    if target_args is None or (process_cmdline and match_lists(target_args, process_cmdline)):
+                        p = psutil.Process(process_info['pid'])
+                        return p   # Emulator process is running
+        except Exception as e:
+            print(e)
+    return None
+
+def terminate(emulator_path: str) -> bool:
+    p = is_running(emulator_path)
+    try:
+        if p is not None:
+            p.terminate()
+            return True  # Emulator process terminated successfully
+    except Exception as e:
+        print(e)
+
+    return False  # Emulator process not found or termination failed

device_operation/start_simulator.py

@@ -45,7 +45,7 @@ def bst_read_registry_key(region):
         command = f""" "{bst_read_registry_key('cn')}" --instance {find_display_name(multi_instance, read_registry_key('cn'))}"""
         subprocess.Popen(command,shell=True)
         return get_simulator_port(simulator_type, multi_instance)
-    if simulator_type == "mumu":
+    if simulator_type in ["mumu", "mumu_global"]:
         if multi_instance == None:
             multi_instance = 0
-        return mumu12_control_api_backend(multi_instance, 'start')
+        return mumu12_control_api_backend(simulator_type, multi_instance, 'start')

device_operation/stop_simulator.py

@@ -58,7 +58,7 @@ def stop_simulator_classic(simulator_type, multi_instance=None):
         if pid:
             os.system(f'taskkill /F /PID {pid[0]}')
 
-    if simulator_type == "mumu":
+    if simulator_type in ["mumu", "mumu_global"]:
         if multi_instance == None:
             multi_instance = 0
-        mumu12_control_api_backend(multi_instance, 'stop')
+        mumu12_control_api_backend(simulator_type, multi_instance, 'stop')