Adds a real-world example that run a rpa in loop on event

command-line/python/rpa_event_loop/build/RPA_Event_Loop/config.json

@@ -0,0 +1,14 @@
+{
+  "read_battery_marco" : "read_battery_marco",
+  "switch_to_util_marco" : "switch_to_util_marco",
+  "switch_to_sol_marco" : "switch_to_sol_marco",
+  "path_autorun_html" : "C:\\ui_vision\\ui.vision.html",
+  "browser_path" : "C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe",
+  "voltage_cutoff_low": 50,
+  "voltage_cutoff_high": 80,
+  "timeout_seconds" : 10,
+  "wait_interval" : 20,
+  "num_executions" : 10,
+  "log_location" : "C:\\ui_vision\\logs"
+}
+

command-line/python/rpa_event_loop/readme.md

@@ -0,0 +1,59 @@
+### How to run
+
+#### Using exe:
+1. Update config.json file based on your environment and needs
+   1. Refer "Understanding config.json" section for details
+   2. The location of config.json is "dist/RPA_Event_Loop/config.json"
+2. Execute the program "RPA_Event_Loop.exe"
+   1. Tge the easiest way to do this to double clik on RPA_Event_Loop.exe
+   2. In case you need to check logs then 
+      1. Open CMD and change the directory to <>/RPA_Event_Loop/
+         1. This step is mandatory 
+         2. Executing the exe using /RPA_Event_Loop/RPA_Event_Loop.exe when current working dir is not /RPA_Event_Loop/ will cause an error
+      2. Start the exe
+
+#### Using Source Code:
+
+1. Install python 3.13 or above
+2. Open a CMD and change directory to RPA_Loop project 
+3. run "pip install -r .\requirements.txt"
+4. Update config.json file based on your environment and needs
+   1. Refer "Understanding config.json" section for details
+   2. The location of config.json is "src/config.json"
+5. execute "src/main.py"
+
+### Understanding config.json
+
+File contains various configurables for the program. Each of these is defined under 
+
+1. **read_battery_marco**: This is the location of the read_battery_marco in Ui.Vision macro inventory.
+2. **switch_to_util_marco**: This is the location of the switch_to_util_marco in Ui.Vision macro inventory.
+3. **switch_to_sol_marco**: This is the location of the switch_to_sol_marco in Ui.Vision macro inventory.
+2. **path_autorun_html**: This is the location where your ui.vision.html stored. The value must be a fully qualified path to a file.
+3. **browser_path**: This is the location where your browser binary stored. The value must be a fully qualified path to a file.
+4. **voltage_cutoff_low**: This is low cutoff value. When the output of read_battery_marco is less than this value switch_to_util_marco will execute.
+5. **voltage_cutoff_high**: This is high cutoff value. When the output of read_battery_marco is greater than this value switch_to_sol_marco will execute.
+4. **timeout_seconds**: This is timeout value for a macro execution. If the longest running marco out of 1,2,3 takes x second to execute set this value to x+10.
+5. **wait_interval**: Wait time between the execution of the flow in seconds. If you want the flow to run every 5 minutes set this to 300.
+6. **num_executions**: Number of times the flow should be executed. Set to 0 for an infinite loop.
+7. **log_location**: Location where you would like to store the logs.
+
+### Sample config.json
+```
+{
+  "read_battery_marco" : "read_battery_marco",
+  "switch_to_util_marco" : "switch_to_util_marco",
+  "switch_to_sol_marco" : "switch_to_sol_marco",
+  "path_autorun_html" : "C:\\ui_vision\\ui.vision.html",
+  "browser_path" : "C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe",
+  "voltage_cutoff_low": 50,
+  "voltage_cutoff_high": 80,
+  "timeout_seconds" : 10,
+  "wait_interval" : 300,
+  "num_executions" : 0,
+  "log_location" : "C:\\ui_vision\\logs" 
+}
+```
+
+### Logical Flow of the code
+![img.png](img.png)

command-line/python/rpa_event_loop/src/PlayFlow.py

@@ -0,0 +1,54 @@
+from datetime import datetime
+import os
+import subprocess
+import sys
+import time
+import json
+from PlayMacro import PlayMacro
+
+
+class PlayFlow:
+    def __init__(self):
+        self.read_battery_marco = None
+        self.switch_to_util_marco = None
+        self.switch_to_sol_marco = None
+        self.voltage_cutoff_low = None
+        self.voltage_cutoff_high = None
+        self.wait_interval = None
+        self.num_executions = None
+        self.on_utility = False
+        self.update_config()
+
+    def update_config(self):
+        fh = open("./config.json", 'r')
+        config = json.load(fh)
+        fh.close()
+        self.read_battery_marco = PlayMacro(macro=config.get("read_battery_marco"))
+        self.switch_to_util_marco = PlayMacro(macro=config.get("switch_to_util_marco"))
+        self.switch_to_sol_marco = PlayMacro(macro=config.get("switch_to_sol_marco"))
+        self.voltage_cutoff_low = config.get("voltage_cutoff_low")
+        self.voltage_cutoff_high = config.get("voltage_cutoff_high")
+
+    def strat_flow(self):
+        bat_capacity = float(self.read_battery_marco.play_macro())
+        time.sleep(2)
+        if bat_capacity < self.voltage_cutoff_low:
+            print(f"{datetime.now()} Battery Capacity at {bat_capacity} switching to utility")
+            if not self.on_utility:
+                self.switch_to_util_marco.play_macro()
+                self.on_utility = True
+            else:
+                print(f"{datetime.now()} Already on utility")
+
+        if bat_capacity > self.voltage_cutoff_high:
+            print(f"{datetime.now()} Battery Capacity at {bat_capacity} switching to Solar")
+            if self.on_utility:
+                self.switch_to_sol_marco.play_macro()
+                self.on_utility = False
+            else:
+                print(f"{datetime.now()}  Already on Solar")
+
+
+if __name__ == '__main__':
+    flow_obj = PlayFlow()
+    flow_obj.strat_flow()

command-line/python/rpa_event_loop/src/PlayMacro.py

@@ -0,0 +1,107 @@
+import json
+import os
+import subprocess
+import time
+from datetime import datetime
+
+import pygetwindow as gw
+
+
+class PlayMacro:
+    def __init__(self, macro, var1='-', var2='-', var3='-'):
+        self.var1 = var1
+        self.var2 = var2
+        self.var3 = var3
+        self.macro = macro
+        self.ui_vision_windows = None
+        self.path_autorun_html = None
+        self.browser_path = None
+        self.timeout_seconds = None
+        self.wait_interval = None
+        self.num_executions = None
+        self.log_location = None
+        self.update_config()
+
+    def get_ui_vision_windows(self):
+        self.ui_vision_windows = []
+        windows = gw.getAllWindows()
+        for window in windows:
+            if "UI.Vision".lower() in str(window.title).lower():
+                self.ui_vision_windows.append(window)
+        print(f"{datetime.now()} Found {len(self.ui_vision_windows)} ui.vision windows")
+
+    def minimize_ui_vision_windows(self):
+        self.get_ui_vision_windows()
+        for window in self.ui_vision_windows:
+            window.minimize()
+        print(f"{datetime.now()} ui.vision windows minimized")
+
+    def close_ui_vision_windows(self):
+        self.get_ui_vision_windows()
+        for window in self.ui_vision_windows:
+            window.close()
+        print(f"{datetime.now()} ui.vision windows closed")
+
+    def update_config(self):
+        fh = open("./config.json", 'r')
+        config = json.load(fh)
+        fh.close()
+        self.path_autorun_html = config.get("path_autorun_html")
+        self.browser_path = config.get("browser_path")
+        self.timeout_seconds = config.get("timeout_seconds")
+        self.log_location = config.get("log_location")
+        if not isinstance(self.macro, str):
+            raise f"macro must be a string value of macro location in UI vision app"
+        if not os.path.exists(self.browser_path):
+            raise f"browser path {self.browser_path} not found"
+        if not os.path.exists(self.path_autorun_html):
+            raise f"autorun html path {self.path_autorun_html} not found"
+        if not os.path.exists(self.log_location):
+            raise f"log location {self.log_location} not found"
+        if not isinstance(self.timeout_seconds, int):
+            raise f"timeout seconds value:{self.timeout_seconds} is not a integer"
+
+    def play_macro(self):
+        result = None
+        log = f'log_{datetime.now().strftime('%m-%d-%Y_%H_%M_%S')}.txt'
+        path_log = os.path.join(self.log_location, log)
+        print(f"{datetime.now()} Log File will show up at {path_log}")
+        args = f'file:///{self.path_autorun_html}?macro={self.macro}&cmd_var1={self.var1}&cmd_var2{self.var2}&cmd_var3{self.var3}&closeRPA=0&direct=1&savelog={path_log}'
+
+        proc = subprocess.Popen([self.browser_path, args])
+
+        time.sleep(2)
+
+        self.minimize_ui_vision_windows()
+
+        status_runtime = 0
+        while not os.path.exists(path_log) and status_runtime < self.timeout_seconds:
+            print(f"{datetime.now()} Waiting for macro to finish, seconds={status_runtime}")
+            time.sleep(1)
+            status_runtime = status_runtime + 1
+
+        if status_runtime < self.timeout_seconds:
+            with open(path_log) as f:
+                status_text = f.readline()
+                if 'Status=OK' in status_text:
+                    print(f"{datetime.now()} {status_text}")
+                    log_data = f.readlines()
+                    for log_line in log_data[::-1]:
+                        if "[echo] Result" in log_line:
+                            print(f"{datetime.now()} {log_line}")
+                            result = log_line.split(':')[1]
+        else:
+            status_text = f"Macro did not complete withing the time given: {self.timeout_seconds} seconds"
+            print(f"{datetime.now()} {status_text}")
+        proc.kill()
+        self.close_ui_vision_windows()
+        return result
+
+
+if __name__ == '__main__':
+    fh = open("./config.json", 'r')
+    config = json.load(fh)
+    fh.close()
+    macro_obj = PlayMacro(macro=config.get("switch_to_sol_marco"))
+    result = macro_obj.play_macro()
+    print(f"Result is {result}")

command-line/python/rpa_event_loop/src/config.json

@@ -0,0 +1,14 @@
+{
+  "read_battery_marco" : "read_battery_marco",
+  "switch_to_util_marco" : "switch_to_util_marco",
+  "switch_to_sol_marco" : "switch_to_sol_marco",
+  "path_autorun_html" : "C:\\ui_vision\\ui.vision.html",
+  "browser_path" : "C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe",
+  "voltage_cutoff_low": 50,
+  "voltage_cutoff_high": 80,
+  "timeout_seconds" : 10,
+  "wait_interval" : 10,
+  "num_executions" : 10,
+  "log_location" : "C:\\ui_vision\\logs"
+}
+

command-line/python/rpa_event_loop/src/main.py

@@ -0,0 +1,26 @@
+import json
+import time
+
+from PlayFlow import PlayFlow
+
+
+def start_monitoring():
+    fh = open("./config.json", 'r')
+    config = json.load(fh)
+    fh.close()
+    wait_interval = config.get("wait_interval")
+    num_executions = config.get("num_executions")
+    flow_obj = PlayFlow()
+    if num_executions > 0:
+        while num_executions > 0:
+            flow_obj.strat_flow()
+            time.sleep(wait_interval)
+            num_executions = num_executions - 1
+    else:
+        while True:
+            flow_obj.strat_flow()
+            time.sleep(wait_interval)
+
+
+if __name__ == '__main__':
+    start_monitoring()