Source code for diva.config.module_config
# Copyright 2024 Mews
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and limitations under the License.
from gc import collect
from typing import TYPE_CHECKING, Union
from diva import parameters
from diva.config.services import AsksMissings
from diva.config.services import CompletionWithLast
from diva.config.services import ConfigCreation
from diva.config.services import GetMissings
from diva.tools import append_if_not_in, datetime_from_str
from diva.logging.logger import crossing_point
if TYPE_CHECKING:
from diva.chat import Prompt
from diva.config import Config
from diva.chat import ModuleChat
[docs]
class ModuleConfig:
"""
This module supervises all tasks related to the search of graph parameters (config) in the user prompt.
The strategies for subtasks (in config.services) are instantiated as private arguments of the class.
Attributes
----------
config: instance of Config
with getter. The instance contains all the retrieved parameters in the user_prompt for graph building
history_config: list['Config']
with getter. List of instances of Config. Newly created Config (for each user prompt categorized as
visualisation) are appended to this list. The list can be cleared with the function clear_history()
missings: list[str]
with getter. List of missing mandatory graph parameters (climate variable, start time, end time, location)
Methods
-------
prompt_to_config(prompt="", verbose=parameters.verbose)
Searches for graph parameters in the user_prompt
complete_missings(verbose=parameters.verbose):
Completes graph parameters not found in the user_prompt with the data from the last Config object in the
history.
ask_missings() -> str:
Generates a text string to ask for the missing parameters (if any). Returns a text string.
clear_history(clear_chat_history=True)
Clear the config history and the chat history if the corresponding parameter is set to True (default)
"""
def __init__(self, module_chat):
"""
Since the user prompt is written in the chat, this class needs to access the chat module.
Parameters
----------
module_chat: instance of ModuleChat
"""
print("NEW INSTANCE CREATED")
self.__module_chat: 'ModuleChat' = module_chat
self.__history_config: list['Config'] = []
self.__config: Union['Config', None] = None
self.__got_missings: bool = False # if True, no need to get them a second time
# ---
self.__creation_strategy = ConfigCreation()
self.__completion_strategy = CompletionWithLast()
self.__ask_missings_strategy = AsksMissings()
self.__get_missings_strategy = GetMissings()
# ---
self.__module_chat.set_config_creation_strategy(self.__creation_strategy)
# others
[docs]
@crossing_point("Function that searches for the graph parameters in the user prompt")
def prompt_to_config(self, prompt: Union[str, 'Prompt'] = "", verbose: bool = parameters.verbose):
"""
Searches for the graph parameters in the user prompt.
Parameters
----------
prompt: str or instance of Prompt
Default to empty string. If this parameter is an empty string, it fetches the instance of prompt from
the instance of chat associated with this class. Else, if the parameters is a non-empty string, the
method uses it to instantiate Prompt, and stores the instance in the chat instance.
verbose: bool
Default value from diva.parameters.verbose.
"""
# ------------------------------------------------------------
# clear config historic if more than 3 consecutive discussions
n = len(self.__module_chat.prompt_history)
if n > 3:
is_viz = [True for prompt in self.__module_chat.prompt_history[n - 4:n - 1] if "discussion" in prompt.type]
if len(is_viz) == 3:
self.clear_history()
# ---
if str(prompt) == "":
prompt = self.__module_chat.prompt
else:
if type(prompt) is str:
self.__module_chat.create_user_prompt(prompt)
prompt = self.__module_chat.prompt
# ---
self.__manage_history()
self.__config = self.__creation_strategy(prompt=prompt, verbose=verbose)
self.__config.set_last(self.__get_last_config())
self.__history_config.append(self.__config)
self.__got_missings = False
# The up-to-be-created config needs to look at the last config to fill missings values.
# Thus, I attach a reference to the last config in the up-to-be-created config to make this process easier
print("last config from module config: ", self.__config.last)
self.complete_missings(verbose)
[docs]
@crossing_point("Function that completes the missing parameters with the data from the last Config instance in history")
def complete_missings(self, verbose: bool = parameters.verbose):
"""
Completes missing parameters with the data from the last Config instance in history.
Parameters
----------
verbose: bool
Default value from diva.parameters.verbose.
"""
if not self.__got_missings:
self.__get_missings_strategy.define_missings(config=self.config, verbose=verbose)
self.__got_missings = True
self.__completion_strategy(self.config)
self.__add_disclaimer()
# get_missings
missings = self.__get_missings_strategy(self.config)
self.__config.set_missings(missings)
[docs]
def ask_missings(self) -> str:
"""
Creates a text string to asks the missing mandatory graph parameters.
Returns
-------
str
A text string asking for the missing mandatory graph parameters.
"""
# when run, got_missings shall be true because done either in prompt_to_config or complete_missings
demand_to_user = self.__ask_missings_strategy(
missings=self.missings
)
self.__module_chat.chat.set_demand_of_info(demand_to_user)
return demand_to_user
def __manage_history(self):
""" Keeps only the most recent Config instances (between 3 and 6) because DIVA does not need a large history. """
if len(self.__history_config) > 6:
self.__history_config = self.__history_config[-3:]
collect()
[docs]
def clear_history(self, clear_chat_history=True):
""" Clears the history of configs and the chat history (by default) """
self.__history_config = []
if clear_chat_history:
self.__module_chat.clear_history()
self.__config = None
def __add_disclaimer(self):
""" Adds disclaimer messages, for example warnings that the predictions are not weather forecasts when the user
asks data in a time range close to today, in the future, and a warning that past data may have not been
recently updated when the user asks data between the last data update and today. """
types = []
start_times = self.__config.start_time.split(", ")
end_times = self.__config.end_time.split(", ")
for i in range(0, len(start_times)):
if start_times[i] != "Unknown" and end_times[i] != "Unknown":
dt_start = datetime_from_str(start_times[i])
dt_end = datetime_from_str(end_times[i])
# show disclaimer if time_max_data is in the displayed range
if dt_start <= parameters.time_max_data <= dt_end:
types = append_if_not_in(types, "update_freq")
if dt_start <= parameters.today_datetime <= dt_end:
types = append_if_not_in(types, "forecast")
# show disclaimer if start_time or end_time is in between time_max and today
if parameters.time_max_data < dt_start < parameters.today_datetime:
types = append_if_not_in(types, "forecast")
if dt_end > parameters.today_datetime:
types = append_if_not_in(types, "update_freq")
if parameters.time_max_data < dt_end < parameters.today_datetime:
types = append_if_not_in(types, "update_freq")
# show disclaimer if start_time or end_time is less -/+ 28 days from today
timedelta = dt_start - parameters.today_datetime
if 0 < timedelta.days < 28:
types = append_if_not_in(types, "forecast")
if types:
self.__module_chat.add_disclaimer(types)
def __get_last_config(self) -> Union['Config', None]:
""" function to get the last Config instance in the history. Returns the last Config instance, if any, and
None otherwise. """
if len(self.__history_config) == 0:
last_config = None
else:
last_config = self.__history_config[-1]
return last_config
@property
def config(self) -> 'Config':
return self.__config
@property
def history_config(self) -> list['Config']:
return self.__history_config
@property
def missings(self) -> list[str]:
""" Missings is not an attribute of the class. This function retrieves the missing parameters
from the class instantiated in the private attribute get_missings_strategy. Returns a list of string. """
return self.__get_missings_strategy.missings