1import logging 

2import os 

3import pickle 

4from dataclasses import dataclass, field 

5from pathlib import Path 

6from typing import Any, Dict, Optional, Tuple, Type, Union 

7 

8import dataclasses_json 

9from dataclasses_json import config, DataClassJsonMixin 

10 

11from .adapters import ConfigurationStore 

12from .ui.state import UIState 

13 

14 

15# JSON decoder and encoder translations 

16def encode_path(path: Path) -> str: 

17 return str(path.resolve()) 

18 

19 

20dataclasses_json.cfg.global_config.decoders[Path] = Path 

21dataclasses_json.cfg.global_config.decoders[Optional[Path]] = ( # type: ignore 

22 lambda p: Path(p) if p else None 

23) 

24 

25 

26dataclasses_json.cfg.global_config.encoders[Path] = encode_path 

27dataclasses_json.cfg.global_config.encoders[ 

28 Optional[Path] # type: ignore 

29] = encode_path 

30 

31 

32@dataclass 

33class ProviderConfiguration: 

34 id: str 

35 name: str 

36 ground_truth_adapter_type: Type 

37 ground_truth_adapter_config: ConfigurationStore 

38 caching_adapter_type: Optional[Type] = None 

39 caching_adapter_config: Optional[ConfigurationStore] = None 

40 

41 def migrate(self): 

42 self.ground_truth_adapter_type.migrate_configuration( 

43 self.ground_truth_adapter_config 

44 ) 

45 if self.caching_adapter_type: 

46 self.caching_adapter_type.migrate_configuration(self.caching_adapter_config) 

47 

48 def clone(self) -> "ProviderConfiguration": 

49 return ProviderConfiguration( 

50 self.id, 

51 self.name, 

52 self.ground_truth_adapter_type, 

53 self.ground_truth_adapter_config.clone(), 

54 self.caching_adapter_type, 

55 ( 

56 self.caching_adapter_config.clone() 

57 if self.caching_adapter_config 

58 else None 

59 ), 

60 ) 

61 

62 def persist_secrets(self): 

63 self.ground_truth_adapter_config.persist_secrets() 

64 if self.caching_adapter_config: 

65 self.caching_adapter_config.persist_secrets() 

66 

67 def asdict(self) -> Dict[str, Any]: 

68 def get_typename(key: str) -> Optional[str]: 

69 key += "_type" 

70 if isinstance(self, dict): 

71 return type_.__name__ if (type_ := self.get(key)) else None 

72 else: 

73 return type_.__name__ if (type_ := getattr(self, key)) else None 

74 

75 return { 

76 "id": self.id, 

77 "name": self.name, 

78 "ground_truth_adapter_type": get_typename("ground_truth_adapter"), 

79 "ground_truth_adapter_config": dict(self.ground_truth_adapter_config), 

80 "caching_adapter_type": get_typename("caching_adapter"), 

81 "caching_adapter_config": dict(self.caching_adapter_config or {}), 

82 } 

83 

84 

85def encode_providers( 

86 providers_dict: Dict[str, Union[ProviderConfiguration, Dict[str, Any]]] 

87) -> Dict[str, Dict[str, Any]]: 

88 return { 

89 id_: ( 

90 config 

91 if isinstance(config, ProviderConfiguration) 

92 else ProviderConfiguration(**config) 

93 ).asdict() 

94 for id_, config in providers_dict.items() 

95 } 

96 

97 

98def decode_providers( 

99 providers_dict: Dict[str, Dict[str, Any]] 

100) -> Dict[str, ProviderConfiguration]: 

101 from sublime_music.adapters import AdapterManager 

102 

103 def find_adapter_type(type_name: str) -> Type: 

104 for available_adapter in AdapterManager.available_adapters: 

105 if available_adapter.__name__ == type_name: 

106 return available_adapter 

107 raise Exception(f"Couldn't find adapter of type {type_name}") 

108 

109 return { 

110 id_: ProviderConfiguration( 

111 config["id"], 

112 config["name"], 

113 ground_truth_adapter_type=find_adapter_type( 

114 config["ground_truth_adapter_type"] 

115 ), 

116 ground_truth_adapter_config=ConfigurationStore( 

117 **config["ground_truth_adapter_config"] 

118 ), 

119 caching_adapter_type=( 

120 find_adapter_type(cat) 

121 if (cat := config.get("caching_adapter_type")) 

122 else None 

123 ), 

124 caching_adapter_config=( 

125 ConfigurationStore(**(config.get("caching_adapter_config") or {})) 

126 ), 

127 ) 

128 for id_, config in providers_dict.items() 

129 } 

130 

131 

132@dataclass 

133class AppConfiguration(DataClassJsonMixin): 

134 version: int = 5 

135 cache_location: Optional[Path] = None 

136 filename: Optional[Path] = None 

137 

138 # Providers 

139 providers: Dict[str, ProviderConfiguration] = field( 

140 default_factory=dict, 

141 metadata=config(encoder=encode_providers, decoder=decode_providers), 

142 ) 

143 current_provider_id: Optional[str] = None 

144 _loaded_provider_id: Optional[str] = field(default=None, init=False) 

145 

146 # Players 

147 player_config: Dict[str, Dict[str, Union[Type, Tuple[str, ...]]]] = field( 

148 default_factory=dict 

149 ) 

150 

151 # Global Settings 

152 song_play_notification: bool = True 

153 offline_mode: bool = False 

154 allow_song_downloads: bool = True 

155 download_on_stream: bool = True # also download when streaming a song 

156 prefetch_amount: int = 3 

157 concurrent_download_limit: int = 5 

158 

159 # Deprecated. These have also been renamed to avoid using them elsewhere in the app. 

160 _sol: bool = field(default=True, metadata=config(field_name="serve_over_lan")) 

161 _pn: int = field(default=8282, metadata=config(field_name="port_number")) 

162 _rg: int = field(default=0, metadata=config(field_name="replay_gain")) 

163 

164 @staticmethod 

165 def load_from_file(filename: Path) -> "AppConfiguration": 

166 config = AppConfiguration() 

167 try: 

168 if filename.exists(): 

169 with open(filename, "r") as f: 

170 config = AppConfiguration.from_json(f.read()) 

171 except Exception: 

172 pass 

173 

174 config.filename = filename 

175 return config 

176 

177 def __post_init__(self): 

178 # Default the cache_location to ~/.local/share/sublime-music 

179 if not self.cache_location: 

180 path = Path(os.environ.get("XDG_DATA_HOME") or "~/.local/share") 

181 path = path.expanduser().joinpath("sublime-music").resolve() 

182 self.cache_location = path 

183 

184 self._state = None 

185 self._loaded_provider_id = None 

186 self.migrate() 

187 

188 def migrate(self): 

189 for _, provider in self.providers.items(): 

190 provider.migrate() 

191 

192 if self.version < 6: 

193 self.player_config = { 

194 "Local Playback": {"Replay Gain": ["no", "track", "album"][self._rg]}, 

195 "Chromecast": { 

196 "Serve Local Files to Chromecasts on the LAN": self._sol, 

197 "LAN Server Port Number": self._pn, 

198 }, 

199 } 

200 

201 self.version = 6 

202 self.state.migrate() 

203 

204 @property 

205 def provider(self) -> Optional[ProviderConfiguration]: 

206 return self.providers.get(self.current_provider_id or "") 

207 

208 @property 

209 def state(self) -> UIState: 

210 if not (provider := self.provider): 

211 return UIState() 

212 

213 # If the provider has changed, then retrieve the new provider's state. 

214 if self._loaded_provider_id != provider.id: 

215 self.load_state() 

216 

217 return self._state 

218 

219 def load_state(self): 

220 self._state = UIState() 

221 if not (provider := self.provider): 

222 return 

223 

224 self._loaded_provider_id = provider.id 

225 if (state_filename := self._state_file_location) and state_filename.exists(): 

226 try: 

227 with open(state_filename, "rb") as f: 

228 self._state = pickle.load(f) 

229 except Exception: 

230 logging.exception(f"Couldn't load state from {state_filename}") 

231 # Just ignore any errors, it is only UI state. 

232 self._state = UIState() 

233 

234 self._state.__init_available_players__() 

235 

236 @property 

237 def _state_file_location(self) -> Optional[Path]: 

238 if not (provider := self.provider): 

239 return None 

240 

241 assert self.cache_location 

242 return self.cache_location.joinpath(provider.id, "state.pickle") 

243 

244 def save(self): 

245 # Save the config as YAML. 

246 self.filename.parent.mkdir(parents=True, exist_ok=True) 

247 json = self.to_json(indent=2, sort_keys=True) 

248 with open(self.filename, "w+") as f: 

249 f.write(json) 

250 

251 # Save the state for the current provider. 

252 if state_filename := self._state_file_location: 

253 state_filename.parent.mkdir(parents=True, exist_ok=True) 

254 with open(state_filename, "wb+") as f: 

255 pickle.dump(self.state, f)