File size: 4,407 Bytes
fbf421d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
133
134
135
136
137
138
139
140
"""This module contains the settings for the Chattr app."""

from logging import getLogger
from pathlib import Path
from typing import List, Literal, Self

from dotenv import load_dotenv
from langchain_core.messages import SystemMessage
from pydantic import (
    BaseModel,
    DirectoryPath,
    Field,
    HttpUrl,
    RedisDsn,
    SecretStr,
    StrictStr,
    model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict

logger = getLogger(__name__)

load_dotenv()


class ModelSettings(BaseModel):
    url: HttpUrl = Field(default=None)
    name: StrictStr = Field(default=None)
    api_key: SecretStr = Field(default=None)
    temperature: float = Field(default=0.0, ge=0.0, le=1.0)
    system_message: SystemMessage = SystemMessage(
        content="You are a helpful assistant that can answer questions about the time and generate audio files from text."
    )

    @model_validator(mode="after")
    def check_api_key_exist(self) -> Self:
        """
        Ensure that an API key and model name are provided if a model URL is set.
        This method validates the presence of required credentials for the model provider.

        Returns:
            Self: The validated ModelSettings instance.

        Raises:
            ValueError: If the API key or model name is missing when a model URL is provided.
        """
        if self.url:
            if not self.api_key or not self.api_key.get_secret_value():
                raise ValueError(
                    "You need to provide API Key for the Model provider via `MODEL__API_KEY`"
                )
            if not self.name:
                raise ValueError("You need to provide Model name via `MODEL__NAME`")
        return self


class MemorySettings(BaseModel):
    url: RedisDsn = Field(default=RedisDsn(url="redis://localhost:6379"))


class VectorDatabaseSettings(BaseModel):
    name: StrictStr = Field(default="chattr")
    url: HttpUrl = Field(default=HttpUrl(url="http://localhost:6333"))


class MCPSettings(BaseModel):
    name: StrictStr = Field(default=None)
    url: HttpUrl = Field(default=None)
    command: StrictStr = Field(default=None)
    args: List[StrictStr] = Field(default=[])
    transport: Literal["sse", "stdio", "streamable_http", "websocket"] = Field(
        default=None
    )


class DirectorySettings(BaseModel):
    base: DirectoryPath = Field(default_factory=lambda: Path.cwd())
    assets: DirectoryPath = Field(default_factory=lambda: Path.cwd() / "assets")
    log: DirectoryPath = Field(default_factory=lambda: Path.cwd() / "logs")
    image: DirectoryPath = Field(
        default_factory=lambda: Path.cwd() / "assets" / "image"
    )
    audio: DirectoryPath = Field(
        default_factory=lambda: Path.cwd() / "assets" / "audio"
    )
    video: DirectoryPath = Field(
        default_factory=lambda: Path.cwd() / "assets" / "video"
    )

    @model_validator(mode="after")
    def create_missing_dirs(self) -> Self:
        """
        Ensure that all specified directories exist, creating them if necessary.
        This method checks and creates any missing directories defined in the DirectorySettings.

        Returns:
            Self: The validated DirectorySettings instance.
        """
        for directory in [
            self.base,
            self.assets,
            self.log,
            self.image,
            self.audio,
            self.video,
        ]:
            directory.mkdir(exist_ok=True)
            logger.info(f"Created directory: {directory}")
        return self


class Settings(BaseSettings):
    """Configuration for the Chattr app."""

    model_config = SettingsConfigDict(
        env_nested_delimiter="__",
        env_parse_none_str="None",
        env_file=".env",
        extra="ignore",
    )

    model: ModelSettings = ModelSettings()
    memory: MemorySettings = MemorySettings()
    vector_database: VectorDatabaseSettings = VectorDatabaseSettings()
    voice_generator_mcp: MCPSettings = MCPSettings(
        url="http://localhost:8080/gradio_api/mcp/sse",
        transport="sse",
        name="voice_generator",
    )
    video_generator_mcp: MCPSettings = MCPSettings(
        url="http://localhost:8002/gradio_api/mcp/sse",
        transport="sse",
        name="video_generator",
    )
    directory: DirectorySettings = DirectorySettings()


if __name__ == "__main__":
    print(Settings().model_dump())