150 lines
6.2 KiB
Python
150 lines
6.2 KiB
Python
import sys
|
||
import os
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
|
||
|
||
from typing import Dict, List, Optional, Any
|
||
from persistence.data_store import DataStore
|
||
|
||
class DataStoreManager:
|
||
_instance = None
|
||
|
||
def __new__(cls):
|
||
if cls._instance is None:
|
||
cls._instance = super().__new__(cls)
|
||
cls._instance._store = DataStore()
|
||
return cls._instance
|
||
|
||
@property
|
||
def current_project(self) -> Optional[str]:
|
||
return self._store.current_project
|
||
|
||
@property
|
||
def current_param(self) -> Optional[str]:
|
||
return self._store.current_param
|
||
|
||
@current_param.setter
|
||
def current_param(self, param_name: str):
|
||
self._store.current_param = param_name
|
||
|
||
def create_project(self, name: str, description: str = "") -> bool:
|
||
"""创建新项目"""
|
||
return self._store.save_project(name, description)
|
||
|
||
def save_param(self, project_name: str, param_name: str,
|
||
channel_settings: Dict[int, Dict], description: str = "") -> bool:
|
||
"""保存参数配置"""
|
||
success = self._store.add_param_to_project(project_name, param_name,
|
||
channel_settings, description)
|
||
if success:
|
||
self._store.current_param = param_name
|
||
return success
|
||
|
||
def get_project(self, name: str) -> Optional[Dict]:
|
||
"""获取项目数据"""
|
||
project_data = self._store.load_project(name)
|
||
return project_data.__dict__ if project_data else None
|
||
|
||
def get_param_data(self, project_name: str, param_name: str) -> Dict:
|
||
"""获取参数数据"""
|
||
return self._store.load_param_data(project_name, param_name)
|
||
|
||
def get_projects(self) -> List[str]:
|
||
"""获取所有项目列表"""
|
||
return self._store.list_projects()
|
||
|
||
def get_params(self, project_name: str) -> List[str]:
|
||
"""获取项目的所有参数列表"""
|
||
return self._store.list_params(project_name)
|
||
|
||
def remove_project(self, name: str) -> bool:
|
||
"""删除项目"""
|
||
return self._store.delete_project(name)
|
||
|
||
def remove_param(self, project_name: str, param_name: str) -> bool:
|
||
"""删除参数"""
|
||
return self._store.delete_param(project_name, param_name)
|
||
|
||
def update_param_value(self, project_name: str, param_name: str,
|
||
parameter_path: str, new_value: Any) -> bool:
|
||
"""更新参数值"""
|
||
try:
|
||
# 加载参数数据
|
||
param_data = self._store.load_param_data(project_name, param_name)
|
||
|
||
# 解析参数路径，更新对应的值
|
||
parts = parameter_path.split('.')
|
||
if parts[0] == 'dataset' and parts[1] == 'tuning_parameters':
|
||
if parts[2] == 'mix_parameters':
|
||
# 例如: dataset.tuning_parameters.mix_parameters[0].mix_left_data
|
||
idx = int(parts[3].split('[')[1].split(']')[0])
|
||
field = parts[4]
|
||
if idx in param_data:
|
||
param_data[idx][field] = new_value
|
||
elif parts[2] == 'eq_parameters':
|
||
# 例如: dataset.tuning_parameters.eq_parameters[0].fc
|
||
idx = int(parts[3].split('[')[1].split(']')[0])
|
||
field = parts[4]
|
||
channel_id = idx // 20 # 假设每个通道最多20个滤波器
|
||
filter_idx = idx % 20
|
||
|
||
if channel_id in param_data and 'filters' in param_data[channel_id]:
|
||
filters = param_data[channel_id]['filters']
|
||
if filter_idx < len(filters):
|
||
filters[filter_idx][field] = new_value
|
||
elif parts[2] == 'delay_parameters':
|
||
# 例如: dataset.tuning_parameters.delay_parameters[0].delay_data
|
||
idx = int(parts[3].split('[')[1].split(']')[0])
|
||
field = parts[4]
|
||
if idx in param_data:
|
||
param_data[idx]['delay_data'] = new_value
|
||
elif parts[2] == 'volume_parameters':
|
||
# 例如: dataset.tuning_parameters.volume_parameters[0].vol_data
|
||
idx = int(parts[3].split('[')[1].split(']')[0])
|
||
field = parts[4]
|
||
if idx in param_data:
|
||
param_data[idx]['vol_data'] = new_value
|
||
|
||
# 保存更新后的参数数据
|
||
return self._store.add_param_to_project(project_name, param_name, param_data)
|
||
except Exception as e:
|
||
return False
|
||
|
||
def update_project(self, project_name: str, project_data: Dict) -> bool:
|
||
"""更新项目数据"""
|
||
try:
|
||
# 将字典转换为ProjectData对象
|
||
from persistence.models import ProjectData
|
||
from dataclasses import asdict
|
||
|
||
# 如果传入的是字典，需要转换为ProjectData对象
|
||
if isinstance(project_data, dict):
|
||
# 确保params是字典而不是列表
|
||
if 'params' in project_data and not isinstance(project_data['params'], dict):
|
||
project_data['params'] = {}
|
||
|
||
# 创建ProjectData对象
|
||
from persistence.models import ProjectData
|
||
project_data_obj = ProjectData(**project_data)
|
||
else:
|
||
project_data_obj = project_data
|
||
|
||
# 更新最后修改时间
|
||
from datetime import datetime
|
||
project_data_obj.last_modified = datetime.now().isoformat()
|
||
|
||
# 保存项目元数据
|
||
self._store._save_project_metadata(project_name, project_data_obj)
|
||
return True
|
||
except Exception as e:
|
||
from component.widget_log.log_handler import logger
|
||
logger.error(f"更新项目失败: {e}")
|
||
import traceback
|
||
logger.error(traceback.format_exc())
|
||
return False
|
||
|
||
@classmethod
|
||
def get_instance(cls) -> 'DataStoreManager':
|
||
"""获取 DataStoreManager 实例"""
|
||
if cls._instance is None:
|
||
cls._instance = DataStoreManager()
|
||
return cls._instance |