Spaces:
Running
Running
File size: 2,537 Bytes
4ed8ca0 |
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 |
import os
import json
import pandas as pd
from datetime import date, timedelta, datetime
from typing import Annotated
# Define custom annotated types
# VerboseType = Annotated[bool, "Whether to print data to console. Default to True."]
SavePathType = Annotated[str, "File path to save data. If None, data is not saved."]
# def process_output(data: pd.DataFrame, tag: str, verbose: VerboseType = True, save_path: SavePathType = None) -> None:
# if verbose:
# print(data.to_string())
# if save_path:
# data.to_csv(save_path)
# print(f"{tag} saved to {save_path}")
def save_output(data: pd.DataFrame, tag: str, save_path: SavePathType = None) -> None:
if save_path:
data.to_csv(save_path)
print(f"{tag} saved to {save_path}")
def get_current_date():
return date.today().strftime("%Y-%m-%d")
def register_keys_from_json(file_path):
with open(file_path, "r") as f:
keys = json.load(f)
for key, value in keys.items():
os.environ[key] = value
def decorate_all_methods(decorator):
def class_decorator(cls):
for attr_name, attr_value in cls.__dict__.items():
if callable(attr_value):
setattr(cls, attr_name, decorator(attr_value))
return cls
return class_decorator
def get_next_weekday(date):
if not isinstance(date, datetime):
date = datetime.strptime(date, "%Y-%m-%d")
if date.weekday() >= 5:
days_to_add = 7 - date.weekday()
next_weekday = date + timedelta(days=days_to_add)
return next_weekday
else:
return date
# def create_inner_assistant(
# name, system_message, llm_config, max_round=10,
# code_execution_config=None
# ):
# inner_assistant = autogen.AssistantAgent(
# name=name,
# system_message=system_message + "Reply TERMINATE when the task is done.",
# llm_config=llm_config,
# is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
# )
# executor = autogen.UserProxyAgent(
# name=f"{name}-executor",
# human_input_mode="NEVER",
# code_execution_config=code_execution_config,
# default_auto_reply="",
# is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
# )
# assistant.register_nested_chats(
# [{"recipient": assistant, "message": reflection_message, "summary_method": "last_msg", "max_turns": 1}],
# trigger=ConversableAgent
# )
# return manager
|