import os import datetime import hashlib import hmac import json import time import requests import urllib3 from threading import Thread, Lock from typing import Optional, Dict from requests import Session from dotenv import load_dotenv from src.enums import ( Endpoint, WifiState, WifiPlanningState ) from src.logs import ( log_debug, log_info, log_warning, log_exception, log_error, ) load_dotenv() # Load environment variables from .env file URL_BASE = os.getenv('URL_BASE') APP_ID = os.getenv('APP_ID') APP_NAME = os.getenv('APP_NAME') APP_VERSION = os.getenv('APP_VERSION') DEVICE_NAME = os.getenv('DEVICE_NAME') TOKEN = os.getenv('TOKEN') TRACK_ID = os.getenv('TRACK_ID') urllib3.disable_warnings() class Fbxpy(): _instance: 'Fbxpy' = None lock: Lock current_session: Session last_use: int def __init__(self): super().__init__() self.lock = Lock() self.current_session = None self.last_use = 0 @classmethod def get_instance(cls): """ Get the singleton instance of Fbxpy. """ if cls._instance is None: cls._instance = Fbxpy() return cls._instance def update_last_use(self): """ Update the last use time of the session. """ self.last_use = time.time() def get_session(self) -> Session: """ Get the current session, creating it if necessary. """ if self.current_session is None: self.create_session() return self.current_session def check_time(self): """ Check if the session is still valid and close it if not. """ can_continue = True while can_continue: time.sleep(10) if self.current_session is None: can_continue = False elif (time.time() - self.last_use) >= 60: can_continue = False self.close_session() log_info("Session closed due to inactivity.") def connexion_post(self, method: Endpoint, data: Optional[Dict] = None): """ Send a POST request to the server. """ url = URL_BASE + method.value if data: data = json.dumps(data) with self.lock: result = json.loads(self.get_session().post(url, data=data).text) self.update_last_use() return result def connexion_post_without_connection(self, method: Endpoint, data: Optional[Dict] = None, session: Session = None): """ Send a POST request to the server without using the current session. """ url = URL_BASE + method.value if data: data = json.dumps(data) return json.loads(session.post(url, data=data).text) def connexion_get(self, method: Endpoint): """ Send a GET request to the server. """ url = URL_BASE + method.value with self.lock: result = json.loads(self.get_session().get(url).text) self.update_last_use() return result def connexion_get_without_connection(self, method: Endpoint, session: Session): """ Send a GET request to the server without using the current session. """ url = URL_BASE + method.value return json.loads(session.get(url).text) def connexion_put(self, method: Endpoint, data: Optional[Dict] = None): """ Send a PUT request to the server. """ url = URL_BASE + method.value if data: data = json.dumps(data) with self.lock: log_info(url) log_info(data) result = json.loads(self.get_session().put(url, data=data).text) self.update_last_use() return result def create_session(self) -> Session: """ Create a new session with the server. """ session = requests.session() session.verify = False challenge = str(self.connexion_get_without_connection(Endpoint.LOGIN, session)["result"]["challenge"]) token_bytes = bytes(TOKEN, 'latin-1') challenge_bytes = bytes(challenge, 'latin-1') password = hmac.new(token_bytes, challenge_bytes, hashlib.sha1).hexdigest() data = { "app_id": APP_ID, "app_version": APP_VERSION, "password": password } content = self.connexion_post_without_connection(Endpoint.LOGIN_SESSION, data, session) log_debug(content) session.headers = {"X-Fbx-App-Auth": content["result"]["session_token"]} self.current_session = session self.update_last_use() Thread(None, self.check_time, 'check_time_thread', (), {}).start() return session def close_session(self): """ Close the current session. """ if self.current_session is None: return try: self.connexion_post(Endpoint.LOGIN_LOGOUT) except Exception as e: log_exception(e) finally: self.current_session = None def get_wifi_state(self) -> WifiState: """ Get the current state of the WiFi. """ try: result = self.connexion_get(method=Endpoint.WIFI_AP) # Loop over the access points for ap in result["result"]: state = WifiState.get_by_value(ap["status"]["state"]) if state == WifiState.ACTIVE: if self.get_wifi_planning_state() == WifiPlanningState.TRUE: log_info("WiFi is active with planning.") return WifiState.ACTIVE_PLANIF else: log_info("WiFi is active without planning.") return WifiState.ACTIVE elif state in [WifiState.INACTIVE, WifiState.DISABLED]: log_warning("WiFi is inactive or disabled.") return WifiState.INACTIVE except Exception as e: log_exception(e) return WifiState.UNKNOWN def get_wifi_planning_state(self) -> WifiPlanningState: """ Get the current state of the WiFi planning. """ try: result = self.connexion_get(method=Endpoint.WIFI_PLANNING) if result["result"]["use_planning"]: return WifiPlanningState.TRUE else: return WifiPlanningState.FALSE except Exception as e: log_exception(e) return WifiPlanningState.UNKNOWN def set_wifi_planning_state(self, value: bool) -> bool: """ Set the state of the WiFi planning. """ try: result = self.connexion_put(Endpoint.WIFI_PLANNING, data={"use_planning": value}) if result["success"] and result["result"]["use_planning"]: return True except Exception as e: log_exception(e) return False def activate_wifi(self) -> bool: """ Activate the WiFi. """ result = self.connexion_put(method=Endpoint.WIFI_CONFIG, data={"enabled": True}) if result["success"]: return True else: return False def stop_wifi(self) -> bool: """ Stop the WiFi. """ try: result = self.connexion_put(method=Endpoint.WIFI_CONFIG, data={"enabled": False}) if result["success"] and not result["result"]["enabled"]: return True except Exception as e: log_exception(e) return False singleton = Fbxpy.get_instance() def check_wifi_state(bypass_check_time: bool = False): # Get current datetime current_time = datetime.datetime.now() log_info(current_time.strftime("%d/%m/%Y, %H:%M:%S")) # Check if minute is 29 or 59 if bypass_check_time or 29 == current_time.minute % 30: log_info("Checking WiFi state...") wifi_state = singleton.get_wifi_state() # Activate the wifi if needed if wifi_state == WifiState.INACTIVE: log_info("WiFi is inactive, trying to activate...") if singleton.activate_wifi(): log_info("WiFi activated successfully.") else: log_error("Failed to activate WiFi.") # Check the planning state and disable it if needed if singleton.get_wifi_planning_state() == WifiPlanningState.TRUE: log_info("WiFi planning is active, trying to disable...") if singleton.set_wifi_planning_state(False): log_info("WiFi planning disabled successfully.") else: log_error("Failed to disable WiFi planning.") # ===================================================================== if __name__ == "__main__": check_wifi_state(bypass_check_time=True) # Keep wifi up while True: try: check_wifi_state() except Exception as e: log_exception(f"An error occurred: {e}") finally: # Compute the number of seconds from now to the next_run_time date current_time = datetime.datetime.now().replace(microsecond=0) if current_time.minute < 29: next_run_time = current_time.replace(minute=29, second=0) else: next_run_time = current_time.replace(minute=59, second=0) seconds_until = (next_run_time - current_time).total_seconds() # Sleep for that duration log_info(f"Sleeping for {seconds_until} seconds until {next_run_time.strftime("%H:%M:%S")}.") time.sleep(seconds_until)