74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
from core.models.system.SystemState import SystemState
|
|
from core.models.session.SessionState import SessionState
|
|
from core.models.enum_models import ProtocolChoice
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
@dataclass
|
|
class GlobalState:
|
|
systemwide_active: bool = False
|
|
protocol: ProtocolChoice = None
|
|
firewalled: bool = False
|
|
dns_set: bool = False
|
|
systemwide_profile_id: Optional[int] = None
|
|
session_profile_ids: list[int] = field(default_factory=list)
|
|
verified: bool = False
|
|
|
|
class ProfileStateRepository:
|
|
_instance = None
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super(ProfileStateRepository, cls).__new__(cls)
|
|
cls._instance._initialize()
|
|
return cls._instance
|
|
|
|
def _initialize(self):
|
|
self.global_state = GlobalState()
|
|
# Initial load: Sync memory with whatever is currently on disk
|
|
sys = SystemState.get()
|
|
if sys:
|
|
self.global_state.systemwide_profile_id = sys.profile_id
|
|
self.global_state.firewalled = sys.firewalled
|
|
self.global_state.dns_set = sys.dns_set
|
|
|
|
# Load existing session IDs from disk into the list
|
|
self.global_state.session_profile_ids = [s.id for s in SessionState.all()]
|
|
|
|
def turn_off_system_connection(self):
|
|
"""Clear the system-wide memory cache."""
|
|
self.global_state.systemwide_profile_id = None
|
|
self.global_state.protocol = None
|
|
self.global_state.firewalled = False
|
|
self.global_state.dns_set = False
|
|
self.global_state.systemwide_active = False
|
|
|
|
|
|
def sync_system_state(self, state: SystemState):
|
|
"""Called by SystemStateController to update memory cache."""
|
|
self.global_state.systemwide_profile_id = state.profile_id
|
|
self.global_state.protocol = state.protocol
|
|
self.global_state.firewalled = state.firewalled
|
|
self.global_state.dns_set = state.dns_set
|
|
if state.profile_id:
|
|
self.global_state.systemwide_active = True
|
|
|
|
|
|
def sync_session_added(self, profile_id: int):
|
|
"""Called by SessionStateController to update memory cache."""
|
|
if profile_id not in self.global_state.session_profile_ids:
|
|
self.global_state.session_profile_ids.append(profile_id)
|
|
|
|
def sync_session_removed(self, profile_id: int):
|
|
"""Called by SessionStateController to update memory cache."""
|
|
if profile_id in self.global_state.session_profile_ids:
|
|
self.global_state.session_profile_ids.remove(profile_id)
|
|
|
|
def is_profile_active(self, profile_id: int) -> bool:
|
|
# Check system profile
|
|
if self.global_state.systemwide_active and self.global_state.systemwide_profile_id == profile_id:
|
|
return True
|
|
# Check session profiles
|
|
return profile_id in self.global_state.session_profile_ids
|
|
|