138 lines
5.4 KiB
Python
138 lines
5.4 KiB
Python
import json
|
|
import os
|
|
from typing import Dict, List, Any, Optional
|
|
from datetime import datetime
|
|
from persistence.models import *
|
|
from component.widget_log.log_handler import logger
|
|
|
|
class DataStore:
|
|
def __init__(self, storage_dir: str = "data/projects"):
|
|
self.storage_dir = storage_dir
|
|
self.current_project: Optional[str] = None
|
|
self.current_param: Optional[str] = None
|
|
self._ensure_storage_dir()
|
|
|
|
def _ensure_storage_dir(self):
|
|
"""确保存储目录存在"""
|
|
if not os.path.exists(self.storage_dir):
|
|
os.makedirs(self.storage_dir)
|
|
|
|
def _get_project_path(self, project_name: str) -> str:
|
|
"""获取项目文件路径"""
|
|
return os.path.join(self.storage_dir, f"{project_name}.json")
|
|
|
|
def save_project(self, project_name: str, description: str = "") -> bool:
|
|
"""创建或更新项目"""
|
|
try:
|
|
now = datetime.now().isoformat()
|
|
project_data = ProjectData(
|
|
name=project_name,
|
|
created_at=now if not self._project_exists(project_name) else self._get_project_created_time(project_name),
|
|
last_modified=now,
|
|
description=description,
|
|
params={}
|
|
)
|
|
self._save_project_data(project_name, project_data)
|
|
self.current_project = project_name
|
|
logger.info(f"项目 {project_name} 保存成功")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"保存项目失败: {e}")
|
|
return False
|
|
|
|
def add_param_to_project(self, project_name: str, param_name: str,
|
|
channel_data: Dict[int, Dict], description: str = "") -> bool:
|
|
"""向项目添加参数配置"""
|
|
try:
|
|
project_data = self.load_project(project_name)
|
|
if not project_data:
|
|
raise ValueError(f"Project {project_name} not found")
|
|
|
|
param_config = ParamConfig(
|
|
name=param_name,
|
|
created_at=datetime.now().isoformat(),
|
|
description=description,
|
|
channels=self._convert_to_channel_config(channel_data)
|
|
)
|
|
|
|
project_data.params[param_name] = param_config
|
|
project_data.last_modified = datetime.now().isoformat()
|
|
|
|
self._save_project_data(project_name, project_data)
|
|
logger.info(f"参数 {param_name} 添加到项目 {project_name} 成功")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"添加参数失败: {e}")
|
|
return False
|
|
|
|
def _convert_to_channel_config(self, channel_data: Dict[int, Dict]) -> Dict[int, ChannelConfig]:
|
|
"""转换通道数据为ChannelConfig格式"""
|
|
converted = {}
|
|
for channel_id, data in channel_data.items():
|
|
filters = [FilterConfig(**f) for f in data.get('filters', [])]
|
|
converted[channel_id] = ChannelConfig(
|
|
delay_data=data.get('delay_data', 0.0),
|
|
vol_data=data.get('vol_data', 0.0),
|
|
mix_left_data=data.get('mix_left_data', 0.0),
|
|
mix_right_data=data.get('mix_right_data', 0.0),
|
|
filters=filters
|
|
)
|
|
return converted
|
|
|
|
def load_project(self, project_name: str) -> Optional[ProjectData]:
|
|
"""加载项目数据"""
|
|
try:
|
|
file_path = self._get_project_path(project_name)
|
|
if not os.path.exists(file_path):
|
|
return None
|
|
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
return ProjectData(**data)
|
|
except Exception as e:
|
|
logger.error(f"加载项目失败: {e}")
|
|
return None
|
|
|
|
def list_projects(self) -> List[str]:
|
|
"""列出所有项目"""
|
|
try:
|
|
projects = []
|
|
for file in os.listdir(self.storage_dir):
|
|
if file.endswith('.json'):
|
|
projects.append(file[:-5])
|
|
return projects
|
|
except Exception as e:
|
|
logger.error(f"列出项目失败: {e}")
|
|
return []
|
|
|
|
def delete_project(self, project_name: str) -> bool:
|
|
"""删除项目"""
|
|
try:
|
|
file_path = self._get_project_path(project_name)
|
|
if os.path.exists(file_path):
|
|
os.remove(file_path)
|
|
if self.current_project == project_name:
|
|
self.current_project = None
|
|
logger.info(f"项目 {project_name} 删除成功")
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"删除项目失败: {e}")
|
|
return False
|
|
|
|
def _project_exists(self, project_name: str) -> bool:
|
|
"""检查项目是否存在"""
|
|
return os.path.exists(self._get_project_path(project_name))
|
|
|
|
def _get_project_created_time(self, project_name: str) -> str:
|
|
"""获取项目创建时间"""
|
|
if self._project_exists(project_name):
|
|
data = self.load_project(project_name)
|
|
return data.created_at if data else datetime.now().isoformat()
|
|
return datetime.now().isoformat()
|
|
|
|
def _save_project_data(self, project_name: str, project_data: ProjectData):
|
|
"""保存项目数据到文件"""
|
|
file_path = self._get_project_path(project_name)
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(asdict(project_data), f, indent=2, ensure_ascii=False) |