ORPA-pyOpenRPA/Robot/pyRobotName_Settings.py

#### Settings file - all initialization goes here!
import sys #For std operations
import logging #Logger for robot
import datetime #For test purpose
import os # os defs
import socket # Detect VM name
from pyOpenRPA.Robot import OrchestratorConnector # Lib for connection with orchestrator

## # # # # # # # # # # # # # # # # ##
# GLOBAL PARAMETERS
gSettings = None
# Auto detect version which should be started
gHostnameVersionDict={ # Dict for versions
    "MACHINE NAME DEV":"DEV", "MACHINE NAME TST":"TST", "MACHINE NAME PRD":"PRD"
}
gADLogin = "LOGIN"; gADPassword = "PASSWORD"
gResourceFullPath = os.path.join(os.getcwd(),"..\\Resources\\")
gSourceLocationRelative = r"..\Sources" # Path to Source folder, which contains package. Need if you want to import some py files from source package
sys.path.insert(0, os.path.abspath(gSourceLocationRelative)) # Add to has feature to import module from Source package pyPackageName
from pyPackage import pyModule # Import pyPackage from Source folder

## # # # # # # # # # # # # # # # # ##
# ORCHESTRATOR CONNECTION PARAMETERS
lOrchestratorHost="localhost"; lOrchestratorPort=8080
lOrchestratorProtocol="http"; lOrchestratorAuthToken="ORCHESTRATOR AUTH TOKEN"

## # # # # # # # # # # # # # # # # ##
gVMHostNameStr = socket.gethostname().upper() # Get VM host name
gVersionStr = gHostnameVersionDict[gVMHostNameStr] # DEV Or TST or PRD
gCWDAbsPathStr = os.getcwd()

## # # # # # # # # # # # # # # # # ##
# INIT GLOBAL DICT WITH ALL SETTINGS
if __name__ == "__main__":
    gSettings={
        "VersionStr":gVersionStr,  # DEV Or TST or PRD
        "Package_1": { # package description
            "TempFolderPath": os.path.abspath("Temp"),  # Temp location settings
            "Folder": { # Listen network folder
                "FolderReadyPostfixStr":"Ready", # Folder ready postfix
                "FolderPathList": [  # DEV is below.
                    { "PathStr": os.path.abspath(r"Test1"), "TypeStr": "TYPE1" },
                    { "PathStr": os.path.abspath(r"Test2"), "TypeStr": "TYPE2" }
                ] if gVersionStr == "DEV" else [ # DEV is upper, TST is below
                    {"PathStr": os.path.abspath(r"Test1"), "TypeStr": "TYPE1"},
                ]  if gVersionStr == "TST" else [ # TST is upper, PRD is below
                    {"PathStr": os.path.abspath(r"Test2"), "TypeStr": "TYPE2"},
                ] if gVersionStr == "PRD" else [], # PRD is upper, None is below
            },
        },
        "Package_2": { "Key1":"Value1" }, # package description
        "Package_3": { "Key1":"Value1" }, # package description
        "BI": { # BI TOOL TO COLLECT BUSINESS EFFECTS
            "VersionStr": gVersionStr, # DEV, TST, PRD
            "ADUserStr": gADLogin, # USER SESSION, where robot is starting
            "VMNameStr": gVMHostNameStr # Test VM
        },
        "Log": { # LOGGER
            "Instance": logging.getLogger("pyFord"),
            "ShowInConsoleBool" : True, #True - Show exception in console window
            "FolderPathStr": f"Logs", #Folder path to save all logs
        },
        "Orchestrator": { # ORCHESTRATOR
            "Storage": {
                "O2R_Reset": {}, # Orchestrator to robot reset storage
                "R2O_Const": {}, # Robot to orchestrator constant storage
                "O2R_Const": {} # Orchestrator to robot constant storage
            },
            "TerminateDef": OrchestratorConnector.IntervalTerminateAll, # Call this function when program must be shutted down (to kill threads from OrchestratorConnector)
        }
    }

    ## # # # # # # # # # # # # # # # # ##
    # ORCHESTRATOR CONNECTOR INIT
    gSettings["OrchestratorConnector"]={
        "IntervalDataSendAsync": [
            #{
            #    "Interval": 1.1,
            #    "RobotStorage": gSettings,
            #    "RobotStorageKey": "Statistics",
            #    "OrchestratorKeyList": ["Storage", "pyFord","RobotToOrchestrator","Statistics"],
            #    "OrchestratorProtocol": lOrchestratorProtocol, "OrchestratorHost": lOrchestratorHost,
            #    "OrchestratorPort": lOrchestratorPort, "OrchestratorAuthToken": lOrchestratorAuthToken
            #}
        ],
        "IntervalDataReceiveResetAsync": [
            {
                "Interval": 3,
                "RobotStorage": gSettings,
                "RobotStorageKey": "OrchestratorToRobotReset",
                "RobotResetValue": {"TurnOffSafeBool":False},
                "OrchestratorKeyList": ["Storage", "pyFord","OrchestratorToRobotReset"],
                "OrchestratorProtocol": lOrchestratorProtocol, "OrchestratorHost": lOrchestratorHost,
                "OrchestratorPort": lOrchestratorPort, "OrchestratorAuthToken": lOrchestratorAuthToken
            }
        ],
        "IntervalDataReceiveAsync": [ ]
    }
    OrchestratorConnector.LoggerSetLevelError() # Turn off many warnings from orchestrator connector logger
    OrchestratorConnector.ConfigurationInit(gSettings["OrchestratorConnector"]) # Run OrchestratorConnector initialization

    ## # # # # # # # # # # # # # # # # ##
    # LOGGER CONFIG
    if not os.path.exists(gSettings["Log"]["FolderPathStr"]):
        os.makedirs(gSettings["Log"]["FolderPathStr"])
    lL=gSettings["Log"]["Instance"]
    lL.setLevel(logging.INFO if gVersionStr == "PRD" else logging.DEBUG) # Log debug if not prd
    # create the logging file handler
    lLFH = logging.FileHandler(os.path.join(gSettings["Log"]["FolderPathStr"], datetime.datetime.now().strftime("%Y_%m_%d") + ".log"), encoding ="UTF-8")
    lLF = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    lLFH.setFormatter(lLF)
    lL.addHandler(lLFH) # add handler to logger object
    if gSettings["Log"]["ShowInConsoleBool"]: # Add console output
        handler = logging.StreamHandler(sys.stdout); handler.setFormatter(lLF); lL.addHandler(handler)

    ## # # # # # # # # # # # # # # # # ##
    # START ROBOT
    lL.info("# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #")
    lL.info("ИМЯ РОБОТА, НАЧАЛО ВЫПОЛНЕНИЯ")
    lL.info("# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #")
    lL.info(f"ХОСТ: {gVMHostNameStr}, ВЕРСИЯ РОБОТА: {gVersionStr}, РАБОЧАЯ ДИРЕКТОРИЯ: {gCWDAbsPathStr}") # Init info
    pyModule.pyModule(inGSettings = gSettings) # Execute main def in Source/pyPackage.pyModule.pyModule