Skip to content

Refactor: centralize config, improve logging, and clean up code #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
d-roho wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Refactor: centralize config, improve logging, and clean up code #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
269 changes: 110 additions & 159 deletions MainApp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,115 +2,114 @@
"""
@author: d-roho
"""
import pywintypes
import logging
import sys
import time

import config
from gsi_pinger import pingerfunc
from win32gui import GetWindowText, GetForegroundWindow

# ---------------------------------------------------------------------------
# Logging setup — INFO to console, DEBUG available via -v flag
# ---------------------------------------------------------------------------
_log_level = logging.DEBUG if "-v" in sys.argv else logging.INFO
logging.basicConfig(
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%H:%M:%S",
level=_log_level,
)
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Windows-only imports (pause-and-play feature)
# ---------------------------------------------------------------------------
try:
import pywintypes
from win32gui import GetWindowText, GetForegroundWindow
_WIN32_AVAILABLE = True
except ImportError:
_WIN32_AVAILABLE = False
logger.warning(
"pywin32 not available — pause-and-play detection disabled on this platform."
)

def current_window():

# Returns name of current window for listener.py
window = GetWindowText(GetForegroundWindow())
return window
def current_window():
if not _WIN32_AVAILABLE:
return ""
return GetWindowText(GetForegroundWindow())


def welcome_message():

# Display Welcome Message unless disabled by "-w" arg
if "-w" not in sys.argv:
f = open('welcome.txt', 'r')
print(''.join([line for line in f]))
try:
with open("welcome.txt", "r") as f:
print("".join(f))
except OSError:
pass
print("CSGOPredictor Launched Successfully!")
if "-p" in sys.argv:
print("PAUSE-AND-PLAY = DISABLED")
else: print("PAUSE-AND-PLAY = ENABLED")
print("PAUSE-AND-PLAY =", "DISABLED" if "-p" in sys.argv else "ENABLED")
if "delay" in sys.argv:
delay_index = sys.argv.index("delay") + 1
delay_value = float(sys.argv[delay_index])
print("DELAY VALUE = " + str(delay_value) + " seconds")
else: print("DELAY VALUE = 0 seconds")
print(f"DELAY VALUE = {delay_value} seconds")
else:
print("DELAY VALUE = 0 seconds")
time.sleep(1.5)


def change_dir():

"""Changes working directory to match location of this python file"""
import os

# Changing working directory
dir_path = os.path.dirname(os.path.realpath(__file__))
os.chdir(str(dir_path))
print("current working directory is - " + str(dir_path))
logger.info("Working directory: %s", dir_path)
return dir_path


def import_model():

"""Imports trained Logistic Regression model"""
import sklearn_json as skljson

# Importing Model
print("Importing Model...")
logger.info("Importing model...")
model = skljson.from_json("CSGOPredictor")
print("Model Imported Successfully!")
logger.info("Model imported successfully.")
return model


def check_for_match_start():
def _wait_for_snapshot():
"""Block until pingerfunc returns a non-empty dict."""
result = {}
while not result:
result = pingerfunc()
return result

"""Pauses script till useful data is being recorded'
to generate predictions"""
test = {}
print("Waiting for CS:GO to be launched...")
while len(test.keys()) == 0:
test = pingerfunc()
print("CS:GO has been launched!")
print("Waiting for Match to begin...")

def check_for_match_start():
logger.info("Waiting for CS:GO to be launched...")
test = _wait_for_snapshot()
logger.info("CS:GO has been launched!")
logger.info("Waiting for match to begin...")
while test.get("allplayers") is None:
test = pingerfunc()
while test.get("map").get("phase") == 'warmup':
while test.get("map", {}).get("phase") == "warmup":
test = pingerfunc()

print("Match has Begun!")
logger.info("Match has begun!")


def match_start_check_postlaunch():

"""Same as check_for_match_start(), but used in cases where
CS:GO is known to have already been launched"""
test = {}
"""Wait until a live (non-warmup, non-empty) match snapshot arrives."""
while True:
try:
test = pingerfunc()
if len(test.get("allplayers").keys()) == 0:
time.sleep(1)
test = pingerfunc()
else:
return False
except AttributeError:
players = test.get("allplayers") if test else None
if players and len(players) > 0:
if test.get("map", {}).get("phase") != "warmup":
return
time.sleep(1)
continue

while True:
try:
test = pingerfunc()
if test.get("map").get("phase") == 'warmup':
time.sleep(1)
test = pingerfunc()
else:
return False
except AttributeError:
except Exception:
logger.debug("match_start_check_postlaunch: transient error, retrying", exc_info=True)
time.sleep(1)
continue


def parse_and_predict():

"""The main loop that parses logs and runs the predictive model.
Returns probability prediction of round outcome"""
window = current_window()
welcome_message()
change_dir()
Expand All @@ -120,147 +119,99 @@ def parse_and_predict():
from snapshot_parser import exception_handler, snapshot_formatter, snapshot_arrayfier
import exceptions
from listener import pause_detector, pause_screen

pause_counter = 0
delay_req = False
if "delay" in sys.argv:
delay_req = True
delay_req = "delay" in sys.argv
delay_value = 0.0
if delay_req:
delay_index = sys.argv.index("delay") + 1
delay_value = float(sys.argv[delay_index])

while True:
try:
if delay_req is True:
if delay_req:
time.sleep(delay_value)
# Checking is a Pause request was initiated in previous loop

# Check for pause request from previous loop iteration
if "-p" not in sys.argv:
from listener import raise_pause_screen # imports latest value
if raise_pause_screen is True: # Checks if pause was requested
from listener import raise_pause_screen
if raise_pause_screen:
pause_screen()

# Pinging for latest snapshot
# Ping for latest snapshot
snapshot = None
while snapshot is None:
snapshot = pingerfunc()

snapshot = exception_handler(snapshot)

# formatting snapshot
snapshot_formatted = snapshot_formatter(snapshot)

# parsing snapshot to create \attributes lists for predictive model
predictors = snapshot_arrayfier(snapshot_formatted)

"""Prediction"""

"""Freeze Time - Making Time prediction attribute default to 115 sec
during freezetime. This makes it so that the low time_left
during freezetime doesn't skew prediction towards Ts"""
# During freezetime substitute a neutral time value so low
# time_left doesn't bias predictions toward T-side.
if snapshot_formatted["phase_countdowns"].get("phase") == "freezetime":
predictors[4] = 115
predictors[4] = config.FREEZETIME_DEFAULT_SECONDS

# Running model with predictors
pred_nested = model.predict_proba([predictors])
pred = pred_nested[0] # converts list from nested to unnested
pred = list(pred) # converts numpy array to list
for i in range(2): # decimal -> %, and round values
pred[i] = round(pred[i]*100, 2)

"""Default Predictions - These are scenarios in which the winner of
the round has been decided (or is a virtual certainty) which the
predictive model is not able to account for when making prediction
"""
# Round Over
if snapshot_formatted["round"]["phase"] == "over":
if snapshot_formatted["round"].get("win_team") == "T":
pred = [0, 100]
if snapshot_formatted["round"].get("win_team") == "CT":
pred = [100, 0]

"""Virtual Round Win - Scenarios in which a team cannot lose,
but the round is still live"""
pred = [round(p * 100, 2) for p in pred_nested[0]]

# Bomb Timer < 5 seconds
if snapshot_formatted["phase_countdowns"].get("phase") == "bomb":
if float(snapshot_formatted["phase_countdowns"].get("phase_ends_in")) < 5.0:
# Override with deterministic outcomes when the round winner is known.
round_data = snapshot_formatted.get("round", {})
if round_data.get("phase") == "over":
if round_data.get("win_team") == "T":
pred = [0, 100]
elif round_data.get("win_team") == "CT":
pred = [100, 0]

"""Time to Defuse > Time left in Round - cant do this with
existing info, solution may be possible (more info from GSI?)"""

# Bomb Planted, All Ts dead - enough time to defuse - same as above

print(pred)
with open('predictions.txt', 'a') as fh: # writes predictions to txt file
fh.write(str(pred)+'\n')

# Check for Pause request (every 10 loops unless delay is specified)
if "-p" not in sys.argv:
if GetWindowText(GetForegroundWindow()) == window:
if delay_req is True: # when delay, check for pause after each loop
pause_detector()
elif pause_counter % 10 == 0:
# Bomb about to explode — T-side virtual win.
countdown = snapshot_formatted.get("phase_countdowns", {})
if countdown.get("phase") == "bomb":
try:
if float(countdown.get("phase_ends_in", 999)) < config.BOMB_NEAR_EXPIRY_SECONDS:
pred = [0, 100]
except (TypeError, ValueError):
pass

logger.info("Prediction: CT=%.2f%% T=%.2f%%", pred[0], pred[1])
with open("predictions.txt", "a") as fh:
fh.write(str(pred) + "\n")

# Poll for pause request (every 10 loops, or every loop when delayed)
if "-p" not in sys.argv and _WIN32_AVAILABLE:
if current_window() == window:
if delay_req or pause_counter % 10 == 0:
pause_detector()
pause_counter += 1
pause_counter += 1

# Exceptions
except exceptions.EmptyServer:

"""Raised by program when no players are found in the server.
Forces program to wait till at least one player is detected"""
print("Server is empty. Program will automatically resume once at least one player joins the server.")
logger.warning("Server is empty — waiting for a player to join...")
time.sleep(1)
print("Waiting...")
match_start_check_postlaunch()
print("Player(s) Detected!")
logger.info("Player(s) detected!")
time.sleep(1)
continue

except exceptions.MatchNotStarted:

"""Raised by program when player is not spectating a match.
Usually occurs when user goes into, then exits a match"""
print("You are not currently spectating a Match. Program will automatically resume when you begin spectating.")
logger.warning("Not spectating a match — waiting...")
time.sleep(1)
print("Waiting...")
match_start_check_postlaunch()
print("Match has Begun!")
logger.info("Match has begun!")
time.sleep(1)
continue

except exceptions.WarmUp:

# Raised by program when match being spectated is in warm up mode
print("Match is in Warm Up Phase. Predictions will begin after Warm Up.")
logger.info("Match is in warm-up — predictions will begin after warm-up.")
time.sleep(1)
print("Waiting...")
check_for_match_start() # For unknown reason, match_start_check_postlaunch() doesnt work here
check_for_match_start()
time.sleep(1)
continue

except KeyError:

"""A catch-all exception which restarts the loop. Should not occur.
Please report on GitHub if found."""
print("KeyError. Restarting loop.")
print("This should not occur. Please raise a Ticket on GitHub!")
logger.warning("KeyError in main loop — restarting. Please report on GitHub if frequent.")
time.sleep(5)
continue

except KeyboardInterrupt:

"""Catches Command Terminal keyboard interrupts. Helps exit program smoothly.
Without this, program raises several errors."""
print("Exiting Program")
logger.info("Keyboard interrupt received — exiting.")
time.sleep(0.5)
try:
sys.exit()
except:
sys.exit()

print("While loop in parse_and_predict somehow broken. This should not occur, please report on GitHub")
time.sleep(5)
sys.exit()
sys.exit(0)


if __name__ == '__main__':
if __name__ == "__main__":
parse_and_predict()
Loading