from langchain_core.tools import tool
from pydantic import BaseModel, Field
import tools.command_list
import flow.flow_controller
from tools.program_called_command_list import save_text as __save_text
from tools.program_called_command_list import load_text as __load_text
from tools.file_list import _create_folder
from tools import file_list
import tools.tools_define
from tools.program_called_command_list import wait_safety
import os
g_Working_file = ""
g_text_lines = []
DATA_SPACE_DIR = tools.tools_define.DATA_SPACE_DIR
CURRENT_DIR = os.path.join(DATA_SPACE_DIR,"code/")
_create_folder(CURRENT_DIR)
class programProgramInput(BaseModel):
code: str = Field(..., description="実行するprogramコード")
@tool(args_schema=programProgramInput)
def program_error_check(code) -> str:
"""programプログラムを実行してその出力を得る。"""
wait_safety() # リクエストの制限を越えないために必要
return flow.flow_controller.python_error_check(code)
@tool
def get_code_error() -> bool:
"""
program_error_checkを使用した結果がエラーだったかどうかを得る。
"""
wait_safety() # リクエストの制限を越えないために必要
return flow.flow_controller.get_code_error()
def __check_data_has_line_number(data):
lines = data.split("\n")
if 0 < len(lines):
for line in lines:
line = line.strip()
if line == "":
continue
if 0 < len(line.split("|")):
if line.split("|")[0].isdecimal():
continue
return False
return True
return False
def _load_program_file_line_number_append(file_name):
global g_text_lines
g_text_lines = __load_text(os.path.join(CURRENT_DIR, file_name)).split("\n")
result =""
for i in range(len(g_text_lines)):
result +=str(i+1).rjust(5) +"|"+g_text_lines[i]
return result
def _save_program_data(file_name, data):
global g_text_lines
if __check_data_has_line_number(data):
_save_program_appended_line_number_data(file_name, data)
else:
g_text_lines = data.split("\n")
_save_as_woring_program_data(file_name)
def _save_program_appended_line_number_data(file_name, data):
global g_text_lines
g_text_lines = data.split("\n")
result =""
for line, i in zip(g_text_lines, range(len(g_text_lines))):
g_text_lines[i] = line.split("|")[1]
result += g_text_lines[i]+"\n"
__save_text(os.path.join(CURRENT_DIR, file_name), result)
class SaveprogramFile(BaseModel):
file_name: str = Field()
data: str = Field()
@tool
def get_working_program_file_name():
"""
現在編集中のprogramコードのファイル名を返します。
"""
global g_Working_file
wait_safety() # リクエストの制限を越えないために必要
return g_Working_file
@tool(args_schema=SaveprogramFile)
def save_program_data(file_name, data):
"""
programコードを保存します。
Args:
file_name:ファイル名:
data:programコード
"""
wait_safety() # リクエストの制限を越えないために必要
_save_program_data("code/" + file_name, data)
class OpenProgramFile(BaseModel):
file_name: str = Field()
@tool(args_schema=OpenProgramFile)
def open_program_file_with_line_number(file_name):
"""
programコードを作業ファイルとして開きます。
読み込んだコードを行番号付きで返します。
Args:
file_name:ファイル名:
Returns:
読み込んだprogramコードを行番号付きで返します
"""
global g_Working_file
g_Working_file = file_name
wait_safety() # リクエストの制限を越えないために必要
return _load_program_file_line_number_append(file_name)
@tool
def cat_program_file(file_name):
"""
programコードを返します。
open_program_file_with_line_numberで開いた作業中データは更新されません。
insert_program_data,delete_program_data,update_program_data,append_program_data で更新されるのはopen_program_file_with_line_numberで開いた作業中データのみです。
また、save_program_woring_data で保存されるのはopen_program_file_with_line_numberで開いたファイル名になります。
プログラムコードは行番号付きで返します。
Args:
file
file_name:ファイル名:
Returns:
programコードを返します。
"""
wait_safety() # リクエストの制限を越えないために必要
# load_program_file では作業データが更新されてしまう。
data = __load_text(os.path.join(CURRENT_DIR, file_name)).split("\n")
result =""
for i in range(len(data)):
result +=str(i+1).rjust(5) + "|" + data[i]
return result
def _save_as_woring_program_data(file_name):
global g_Working_file
g_Working_file = file_name
_save_woring_program_data()
@tool(args_schema=OpenProgramFile)
def save_as_woring_program_data(file_name):
"""
現在編集中のprogramコードを名前を付けて保存します。
Args:
file_name:ファイル名:
"""
wait_safety() # リクエストの制限を越えないために必要
_save_as_woring_program_data(file_name)
def _save_woring_program_data():
global g_text_lines
global g_Working_file
file_name = g_Working_file
result =""
for line in g_text_lines:
result += line + "\n"
__save_text(os.path.join(CURRENT_DIR, file_name), result)
@tool
def save_woring_program_data():
"""
現在編集中のprogramコードを上書き保存します。
Args:
file_name:ファイル名:
"""
wait_safety() # リクエストの制限を越えないために必要
_save_woring_program_data()
# 今回はこれはtoolとしない。AIには行番号付きのデータだけ扱わせる。
def _get_Working_program_data():
"""
現在作業中のprogramコードを返します。
Returns:
str:現在作業中のprogramコードを返します。
"""
global g_text_lines
result =""
for line in g_text_lines:
result += line + "\n"
return result
@tool
def get_working_program_data_line_number_append():
"""
現在編集中のprogramコードの
行番号の入ったテキストを返します。
Returns:
str:現在編集中のprogramコードの行番号の入ったテキストを返します。
"""
global g_text_lines
wait_safety() # リクエストの制限を越えないために必要
result =""
for i in range(len(g_text_lines)):
result +=str(i+1).rjust(5) +"|"+g_text_lines[i]
return result
class InsrtprogramData(BaseModel):
insert_line: int = Field()
data: str = Field()
@tool(args_schema=InsrtprogramData)
def insert_program_data(insert_line, data):
"""
現在編集中のprogramコードにデータを挿入します。
:param insert_line: 挿入する行番号。この行の前にデータが挿入される
:param data: 挿入するprogramコードのデータ
"""
global g_text_lines
wait_safety() # リクエストの制限を越えないために必要
# ライン番号を出力するとき1がはじめなのでもらった値を-1する必要がある。
insert_line -= 1
if __check_data_has_line_number(data):
lines = data.split("\n")
for line in lines:
line_buf = line.split("|")[1]
g_text_lines.insert(insert_line, line_buf)
insert_line += 1
else:
lines = data.split("\n")
for line in lines:
g_text_lines.insert(insert_line, line)
insert_line += 1
class DeleteprogramData(BaseModel):
start_line: int = Field()
end_line: int = Field()
@tool(args_schema=DeleteprogramData)
def delete_program_data(start_line, end_line):
"""
現在編集中のprogramコードからデータを削除します。
:param start_line: 削除を開始する行番号
:param end_line: 削除を終了する行番号、1行だけ消すときはstart_lineとend_lineを同じ値にする
"""
global g_text_lines
wait_safety() # リクエストの制限を越えないために必要
start_line -= 1
# ライン番号を出力するとき1がはじめなのでもらった値を-1する必要がある。
for _ in range(start_line, end_line):
g_text_lines.pop(start_line)
class UpdatePyrhonData(BaseModel):
start_line: int = Field()
end_line: int = Field()
data: str = Field()
@tool(args_schema=UpdatePyrhonData)
def update_program_data(start_line, end_line, data):
"""
現在編集中のprogramコードのデータを更新します。
:param start_line: 更新を開始する行番号
:param end_line: 更新を終了する行番号、1行だけ更新するときはstart_lineとend_lineを同じ値にする
:param data: 更新後のprogramコードのデータ
"""
global g_text_lines
wait_safety() # リクエストの制限を越えないために必要
start_line -= 1
# ライン番号を出力するとき1がはじめなのでもらった値を-1する必要がある。
delete_program_data(start_line, end_line)
insert_program_data(start_line, data)
class AppendprogramData(BaseModel):
data: str = Field()
@tool(args_schema=AppendprogramData)
def append_program_data(data):
"""
現在編集中のprogramコードにデータを追記します。
:param data: 追記するprogramコードのデータ
"""
global g_text_lines
wait_safety() # リクエストの制限を越えないために必要
if __check_data_has_line_number(data):
lines = data.split("\n")
for line in lines:
line_buf = line.split("|")[1]
g_text_lines.append(line_buf)
else:
lines = data.split("\n")
for line in lines:
g_text_lines.append(line)
#==============================
# プログラムのファイルリスト管理ツール
def _get_program_file_descriptions() -> str:
"""
プログラムのデータの保存先フォルダ内のファイルリストを取得します。
:return: ファイルパスのリスト
:rtype: List[str]
"""
return file_list._get_and_create_all_file_descriptions(CURRENT_DIR)
@tool
def get_program_file_list() -> str:
"""
プログラムのデータの保存先フォルダ内のファイルリストを取得します。
:return: ファイルパスのリスト
:rtype: List[str]
"""
wait_safety() # リクエストの制限を越えないために必要
return _get_program_file_descriptions()
class SetProgressFileDescriptionInput(BaseModel):
file_path: str = Field(..., description="プログラムのファイルのパス")
description: str = Field(..., description="ファイルの説明")
note: str = Field("", description="ファイルのメモ書き")
@tool(args_schema = SetProgressFileDescriptionInput)
def set_program_file_description(file_path: str, description: str, note: str="") -> None:
"""
プログラムのデータの保存先フォルダ内のファイルリストにファイルの説明を追加します。
この機能呼び出しの前にget_program_file_listを実行してファイルリストを初期化してください。
Args:
:param file_path: ファイルのパス
:type file_path: str
:param description: ファイルの説明
:type description: str
:param note: ファイルのメモ書き
:type note: str
"""
wait_safety() # リクエストの制限を越えないために必要
file_list._set_file_description(file_path, description, note)
def _save_program_file_descriptions() -> None:
file_list._save_file_descriptions()
@tool
def save_program_file_descriptions() -> None:
"""
プログラムのファイルリスト管理ツールのデータの保存先フォルダ内の
ファイルの説明を保存します。
この機能呼び出しの前にget_program_file_listを実行してファイルリストを初期化してください。
"""
wait_safety() # リクエストの制限を越えないために必要
file_list._save_file_descriptions()
def get_tools_list():
return [
program_error_check,
get_code_error,
get_working_program_file_name,
open_program_file_with_line_number,
cat_program_file,
save_program_data,
save_as_woring_program_data,
save_woring_program_data,
insert_program_data,
delete_program_data,
update_program_data,
append_program_data,
#load_program_file_line_number_append,
get_working_program_data_line_number_append,
]
