You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
101 lines
4.4 KiB
Python
101 lines
4.4 KiB
Python
import os
|
|
import json
|
|
import asyncio
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
|
|
# Set PYTHONPATH to include the current directory so we can import src
|
|
sys.path.append(os.getcwd())
|
|
|
|
# Set Environment Variables if not present (matches your .env)
|
|
os.environ.setdefault("PFI_API_URL", "http://192.168.1.82:8000/pfi/api/v1")
|
|
os.environ.setdefault("RP_API_URL", "http://192.168.1.82:8000/reliability")
|
|
os.environ.setdefault("PLANT_STATUS_URL", "http://192.168.1.82:8000/monitoring/plant-status")
|
|
|
|
from src.ircfa_interaction.tools import (
|
|
CurrentReliabilityTool,
|
|
CurrentSensorTool,
|
|
FailureModeTool,
|
|
GetLocationTagEquipmentName,
|
|
HistorySensorTool,
|
|
NextFailureRPTool,
|
|
PlantStatusTool,
|
|
RCFATool,
|
|
HistoryReliabilityTool,
|
|
SearchEquipmentNameTool,
|
|
GetRPLocationTag,
|
|
PerEquipmentRCFATool,
|
|
IntentionRecognitionTool
|
|
)
|
|
|
|
# Token provided by USER
|
|
TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJmcmVzaCI6ZmFsc2UsImlhdCI6MTc3NTEwMzUzNi43NTc3OTMsImp0aSI6Imd6d0lPSFJISzJ6VFFSUm8taUxBamciLCJ0eXBlIjoiYWNjZXNzIiwic3ViIjoiMTUzZjdlMmUtZjBlYS00NDBjLTk5Y2QtN2E3ODJhZThhNjMxIiwibmJmIjoxNzc1MTAzNTM2LCJjc3JmIjoiNTRhOGJkMGEtMWUyNS00YTViLWFlNTMtMTkwZmZiYzFkY2EyIiwiZXhwIjoyMDkwNDYzNTM2LCJub25jZSI6IjJiZGZlNjM5YTExMzU3MjYxYzliMzVjZGFkMTNmYjI4IiwidXNlcm5hbWUiOiJkZXZhZG1pbiJ9.BU83kdKkpBuM_Cg_BzQGvrUQcNcqabMHCG_YTuhzMbwQTvgXD9FzYlafm2GmJuO7mYsWNCJbFejD41M5hSflmvIUgzZgVs6ye8kOgWYiMlfty5vnXfeEgnHTPzGpK0lBXZoXEtEgAEmECMt4g2IQM5-2ghVFsJ5cUszK2VLvWj-4QmxFfDLTdDX0zkNBZnT3dcu6CLLjrdyYBXma0kJFijeGlFHQEN1a6_oei4OvN0xi46iu35Jq0pEBJDUjmISAVblywkGbWFwA13E38I-Ze39jShB9BagDFZ3IhuxsdbTAZeVL6S55YMxpzUeeCEkmEvDGc6oJWTFc9wERlwc7Qw"
|
|
TEST_TAG = "3AH-H501A"
|
|
|
|
async def test_tool(tool_class, name, params, init_tag=None):
|
|
print(f"\n--- Testing {name} ---")
|
|
try:
|
|
# Introspect constructor arguments
|
|
import inspect
|
|
sig = inspect.signature(tool_class.__init__)
|
|
init_args = {}
|
|
if "token" in sig.parameters:
|
|
init_args["token"] = TOKEN
|
|
if "location_tag" in sig.parameters:
|
|
init_args["location_tag"] = init_tag
|
|
|
|
tool = tool_class(**init_args)
|
|
|
|
# Call the tool
|
|
result = tool.call(params)
|
|
|
|
# Simple output verification
|
|
output_str = str(result)
|
|
print(f"Result (first 500 chars): {output_str[:500]}...")
|
|
|
|
if "success" in output_str.lower() or "data" in output_str.lower() or "📊" in output_str or "[" in output_str:
|
|
print(f"✅ {name} returned data/structure.")
|
|
else:
|
|
print(f"⚠️ {name} result might be an error or empty.")
|
|
|
|
except Exception as e:
|
|
print(f"❌ {name} failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
async def main():
|
|
print(f"Starting Tool Capability Tests with token provided.")
|
|
|
|
# 1. Info / List Tools
|
|
await test_tool(GetLocationTagEquipmentName, "GetLocationTagEquipmentName", {"limit": 3})
|
|
await test_tool(SearchEquipmentNameTool, "SearchEquipmentNameTool", {"equipment_name": "Pump"})
|
|
await test_tool(GetRPLocationTag, "GetRPLocationTag", {})
|
|
await test_tool(PlantStatusTool, "PlantStatusTool", {})
|
|
|
|
# 2. Real-time Reliability & Sensor Tools
|
|
# Test fallback vs param priority for CurrentReliabilityTool
|
|
await test_tool(CurrentReliabilityTool, "CurrentReliabilityTool (Fallback)", {}, init_tag=TEST_TAG)
|
|
await test_tool(CurrentReliabilityTool, "CurrentReliabilityTool (Params Priority)", {"location_tag": "3FW-H050"}, init_tag=TEST_TAG)
|
|
# await test_tool(CurrentSensorTool, "CurrentSensorTool", {"location_tag": TEST_TAG})
|
|
|
|
# 3. History Tools
|
|
start = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d")
|
|
end = datetime.utcnow().strftime("%Y-%m-%d")
|
|
await test_tool(HistorySensorTool, "HistorySensorTool", {"location_tag": TEST_TAG, "start_date": start, "end_date": end})
|
|
await test_tool(HistoryReliabilityTool, "HistoryReliabilityTool", {"location_tag": TEST_TAG})
|
|
|
|
# 4. Analysis / Alert Tools
|
|
await test_tool(FailureModeTool, "FailureModeTool", {"location_tag": TEST_TAG})
|
|
await test_tool(RCFATool, "RCFATool", {})
|
|
await test_tool(NextFailureRPTool, "NextFailureRPTool", {})
|
|
await test_tool(PerEquipmentRCFATool, "PerEquipmentRCFATool", {"location_tag": TEST_TAG})
|
|
|
|
# 5. Classifier Tool
|
|
await test_tool(IntentionRecognitionTool, "IntentionRecognitionTool", {"question": "Apa penyebab boiler fail?"})
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1:
|
|
TOKEN = sys.argv[1]
|
|
|
|
asyncio.run(main())
|