blob: 984ec16cd69303dea9cff7673fadc7feb1df76df (
plain) (
blame)
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
|
# This file is part of Senator Sperma
# Licensed under the GPL-3.0 License
# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
# https://codeberg.org/salmonik/senator-sperma
import json
import logging
import aiofiles
import os
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
settings_file = "data/settings.json"
@dataclass
class Settings:
volume: float = 1.0
baseurl: str = "http://localhost:8080"
_data: dict = field(default_factory=dict, init=False)
async def load(self) -> "Settings":
default_settings = {
"volume": 1.0,
"baseurl": "http://localhost:8080",
}
try:
if os.path.exists(settings_file):
async with aiofiles.open(settings_file, "r") as f:
content = await f.read()
if content.strip():
self._data = json.loads(content)
self._sync_from_data()
logger.info("Settings loaded into memory")
else:
self._data = default_settings.copy()
self._sync_from_data()
await self.save()
logger.warning("Empty settings.json, reset to defaults")
else:
self._data = default_settings.copy()
self._sync_from_data()
await self.save()
logger.info("Created default settings.json")
except (json.JSONDecodeError, Exception) as e:
logger.error(f"Error loading settings: {e}")
self._data = default_settings.copy()
self._sync_from_data()
return self
def _sync_from_data(self):
self.volume = self._data.get("volume", 1.0)
self.baseurl = self._data.get("baseurl", "http://localhost:8080")
def _sync_to_data(self):
self._data["volume"] = self.volume
self._data["baseurl"] = self.baseurl
async def save(self):
self._sync_to_data()
async with aiofiles.open(settings_file, "w") as f:
await f.write(json.dumps(self._data, indent=2))
_settings_instance: Settings | None = None
async def load_settings() -> Settings:
global _settings_instance
_settings_instance = Settings()
await _settings_instance.load()
return _settings_instance
async def save_settings():
if _settings_instance:
await _settings_instance.save()
def get_settings() -> Settings | None:
return _settings_instance
def get_volume() -> float:
return _settings_instance.volume if _settings_instance else 1.0
def get_baseurl() -> str:
return _settings_instance.baseurl.rstrip("/") if _settings_instance else "http://localhost:8080"
|
