import threading
import requests
import json
# Assume DEEPSEEK_API_KEY is replaced with VENICE_API_KEY
VENICE_API_KEY = "your_venice_api_key_here"
class DaVeniceRun(ShorniSplash):
def __init__(self):
super().__init__()
self.input_text = "" # Temporary storage for input text
self.context = "" # Store context and preferences here
# Venice.ai api key (place it in DLC/api_keys/venice_api_key.txt)
self.apikey: str = VENICE_API_KEY
def trigger(self, ear: str, skin: str, eye: str) -> bool:
# Check if the ear string ends with the word "run"
return ear.strip().endswith("run")
@staticmethod
def _async_func(this_cls):
# Use the stored input text and context
input_text = this_cls.input_text
context = this_cls.context
data = {
"prompt": f"{context}\n{input_text}",
"model": "venice-uncensored-1.1", # Specify the model you want to use
"max_tokens": 150 # Adjust the number of tokens as needed
}
# Call the Venice.ai API (replace with actual API endpoint and logic)
response = this_cls.call_venice_api(data, this_cls.apikey)
if response.status_code == 200:
this_cls._result = response.json().get("choices", [{}])[0].get("text", "No response from API")
else:
this_cls._result = f"Error calling Venice.ai API: Status code {response.status_code}"
def input(self, ear: str, skin: str, eye: str):
# Check if the skill should trigger
if self.trigger(ear, skin, eye):
# Remove the last word "run" from the ear string
self.input_text = ear.rsplit(" ", 1)[0].strip()
# Start the async operation in a daemon thread
my_thread = threading.Thread(
target=self._async_func,
args=(self,) # Pass self as the only argument
)
my_thread.daemon = True
my_thread.start()
# Output the result if available
if len(self._result) > 0:
self.output_result()
self._result = ""
@staticmethod
def call_venice_api(input_text: str, venice_api_key: str) -> str:
# Replace this with the actual Venice.ai API call logic
# Example:
api_url = "https://api.venice.ai/v1/generate"
payload = input_text
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {venice_api_key}"
}
response = requests.post(api_url, json=payload, headers=headers)
return response
def skillNotes(self, param: str) -> str:
if param == "notes":
return "Venice.ai rest API"
elif param == "triggers":
return "end your input with run."
return "note unavailable"
def set_context(self, context: str):
"""Set the context and preferences to be maintained across API calls."""
self.context = context
def get_context(self) -> str:
"""Get the current context and preferences."""
return self.context