1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
"""
Base service class with common functionality.
"""
import os
import shutil
from pathlib import Path
from typing import Any, Optional, TypeVar, Dict
from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME
# Generic type for response dictionaries
ResponseType = TypeVar('ResponseType', bound=Dict[str, Any])
class BaseService:
"""Base service class with common functionality"""
def __init__(self, logger: Optional[Any] = None):
"""Initialize base service
Args:
logger: Logger instance, defaults to decky.logger if None
"""
if logger is None:
import decky
self.log = decky.logger
else:
self.log = logger
# Initialize common paths using pathlib
self.user_home = Path.home()
self.local_lib_dir = self.user_home / LOCAL_LIB
self.local_share_dir = self.user_home / VULKAN_LAYER_DIR
self.lsfg_script_path = self.user_home / SCRIPT_NAME
self.lsfg_launch_script_path = self.user_home / SCRIPT_NAME # ~/lsfg launch script
self.config_dir = self.user_home / CONFIG_DIR
self.config_file_path = self.config_dir / CONFIG_FILENAME
def _ensure_directories(self) -> None:
"""Create necessary directories if they don't exist"""
self.local_lib_dir.mkdir(parents=True, exist_ok=True)
self.local_share_dir.mkdir(parents=True, exist_ok=True)
self.config_dir.mkdir(parents=True, exist_ok=True)
self.log.info(f"Ensured directories exist: {self.local_lib_dir}, {self.local_share_dir}, {self.config_dir}")
def _remove_if_exists(self, path: Path) -> bool:
"""Remove a file if it exists
Args:
path: Path to the file to remove
Returns:
True if file was removed, False if it didn't exist
Raises:
OSError: If removal fails
"""
if path.exists():
try:
path.unlink()
self.log.info(f"Removed {path}")
return True
except OSError as e:
self.log.error(f"Failed to remove {path}: {e}")
raise
else:
self.log.info(f"File not found: {path}")
return False
def _write_file(self, path: Path, content: str, mode: int = 0o644) -> None:
"""Write content to a file
Args:
path: Target file path
content: Content to write
mode: File permissions (default: 0o644)
Raises:
OSError: If write fails
"""
try:
# Write directly to the file
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
f.flush() # Ensure data is written to disk
os.fsync(f.fileno()) # Force filesystem sync
# Set permissions
path.chmod(mode)
self.log.info(f"Wrote to {path}")
except Exception:
self.log.error(f"Failed to write to {path}")
raise
def _success_response(self, response_type: type, message: str = "", **kwargs) -> Any:
"""Create a standardized success response
Args:
response_type: The TypedDict response type to create
message: Success message
**kwargs: Additional response fields
Returns:
Success response dict
"""
response = {
"success": True,
"message": message,
"error": None
}
response.update(kwargs)
return response
def _error_response(self, response_type: type, error: str, message: str = "", **kwargs) -> Any:
"""Create a standardized error response
Args:
response_type: The TypedDict response type to create
error: Error description
message: Optional message
**kwargs: Additional response fields
Returns:
Error response dict
"""
response = {
"success": False,
"message": message,
"error": error
}
response.update(kwargs)
return response
|
