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
|
"""
Base service class with common functionality.
"""
import logging
import os
import shutil
from pathlib import Path
from typing import Any, Optional, TypeVar, Dict
import decky
from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME
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:
self.log = decky.logger
else:
self.log = logger
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
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:
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
f.flush()
os.fsync(f.fileno())
path.chmod(mode)
self.log.info(f"Wrote to {path}")
except (OSError, IOError, PermissionError) as e:
self.log.error(f"Failed to write to {path}: {e}")
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
|
