from gpt4all import GPT4All
from contextlib import contextmanager
from pythonosc import dispatcher
from pythonosc import osc_server
from pythonosc import udp_client
import concurrent.futures
from functools import lru_cache
# Initialize the model once and reuse it
model = GPT4All(model_name='Meta-Llama-3-8B-Instruct.Q4_0.gguf', n_threads=8, device="gpu")
system_template = 'You are a system that receives a sentence and returns values for a primary, secondary, tertiary and quaternary category, as inferred by the meaning of the sentence. The four categories are: hero, angel, feminist, rebel. You cannot pick the same word for different categories. Primary and secondary category cannot be 0, and primary category must be bigger than the secondary. The sum of the four categories should always be 100. If the sentence doesnt make sense, try your best to derive some meaning. If no meaning can be derived or if the sentence is in any way offensive, just say "Ambiguous". Always format your answer like this example: PRIMARY: "angel" = calculated value, SECONDARY: "hero" = calculated value, TERTIARY: "feminist" = calculated value, QUATERNARY: "rebel" = calculated value. From now on, you will receive sentences, and you will reply only in the above format. PRIMARY and SECONDARY strictly cannot be the same. If the sentence is ambiguous, do not use the PRIMARY, SECONDARY format, but just say AMBIGUOUS".\n'
class FlorenceAttributes:
LABELS = ["angel", "hero","feminist", "rebel"]
def __init__(self, primary=None, secondary=None, tertiary=None, quaternary=None, primary_value=None, secondary_value=None, tertiary_value=None, quaternary_value=None, ambiguous=False):
self.secondary = secondary
self.quaternary = quaternary
self.primary_value = primary_value
self.secondary_value = secondary_value
self.tertiary_value = tertiary_value
self.quaternary_value = quaternary_value
self.ambiguous = ambiguous
return f'PRIMARY: "{self.primary}" = {self.primary_value}, SECONDARY: "{self.secondary}" = {self.secondary_value}, TERTIARY: "{self.tertiary}" = {self.tertiary_value}, QUATERNARY: "{self.quaternary}" = {self.quaternary_value}'
def has_null_labels(self):
return any((label == "" or label == "None") for label in [self.primary, self.secondary, self.tertiary, self.quaternary])
def has_null_values(self):
return any((value == 0 or value == None)for value in [self.primary_value, self.secondary_value, self.tertiary_value, self.quaternary_value])
def fill_missing_labels(self):
used_labels = [label for label in [self.primary, self.secondary, self.tertiary, self.quaternary] if label != ""]
free_labels = [label for label in self.LABELS if label not in used_labels]
if len(free_labels) == 1:
self.primary = free_labels[0]
elif len(free_labels) > 1:
self.primary = random.choice(free_labels)
used_labels = [label for label in [self.primary, self.secondary, self.tertiary, self.quaternary] if label != ""]
free_labels = [label for label in self.LABELS if label not in used_labels]
if len(free_labels) == 1:
self.secondary = free_labels[0]
elif len(free_labels) > 1:
self.secondary = random.choice(free_labels)
used_labels = [label for label in [self.primary, self.secondary, self.tertiary, self.quaternary] if label != ""]
free_labels = [label for label in self.LABELS if label not in used_labels]
if len(free_labels) == 1:
self.tertiary = free_labels[0]
elif len(free_labels) > 1:
self.tertiary = random.choice(free_labels)
used_labels = [label for label in [self.primary, self.secondary, self.tertiary, self.quaternary] if label != ""]
free_labels = [label for label in self.LABELS if label not in used_labels]
if self.quaternary == "":
if len(free_labels) == 1:
self.quaternary = free_labels[0]
elif len(free_labels) > 1:
self.quaternary = random.choice(free_labels)
used_labels = [label for label in [self.primary, self.secondary, self.tertiary, self.quaternary] if label != ""]
free_labels = [label for label in self.LABELS if label not in used_labels]
def reorganize_data(self):
global new_message # Declare the global variable inside the function
new_message += 1 # Update the global counter variable
labels = [self.primary, self.secondary, self.tertiary, self.quaternary]
values = [self.primary_value, self.secondary_value, self.tertiary_value, self.quaternary_value]
# Pair each label with its corresponding value
label_value_pairs = zip(labels, values)
# Rearrange labels and values according to the specified order
for label in self.LABELS:
value = next((v for l, v in label_value_pairs if l == label), None)
result.append(f"{label}: {value}")
result.append(f"/new_message {new_message}")
def finalGeneratedValues(self, wordResponse):
global new_message # Declare the global variable inside the function
new_message += 1 # Update the global counter variable
return (f"{self.primary} = {self.primary_value}, "
f"{self.secondary} = {self.secondary_value}, "
f"{self.tertiary} = {self.tertiary_value}, "
f"{self.quaternary} = {self.quaternary_value}, "
f"newMessage = {new_message}, "
f"wordResponse = {wordResponse}")
def generate_random_attributes(cls):
labels = cls.LABELS.copy()
quaternary = labels.pop()
first_value = random.randint(0, 100)
second_value = random.randint(0, 100 - first_value)
sum_value = first_value + second_value
third_value = random.randint(0, 100 - sum_value)
sum_value1 = sum_value + third_value
fourth_value = 100 - sum_value1
primary_value = first_value
secondary_value = second_value
tertiary_value = third_value
quaternary_value = fourth_value
return cls(primary, secondary, tertiary, quaternary, primary_value, secondary_value, tertiary_value, quaternary_value)
def update_with_random_attributes(self):
random_attributes = FlorenceAttributes.generate_random_attributes()
self.primary = random_attributes.primary
self.secondary = random_attributes.secondary
self.tertiary = random_attributes.tertiary
self.quaternary = random_attributes.quaternary
self.primary_value = random_attributes.primary_value
self.secondary_value = random_attributes.secondary_value
self.tertiary_value = random_attributes.tertiary_value
self.quaternary_value = random_attributes.quaternary_value
self.ambiguous = random_attributes.ambiguous
def __init__(self, model_name):
self.model_name = model_name
self.model = GPT4All(model_name, n_threads=8, device="gpu")
self.session_active = False
if not self.session_active:
self.session = self.model.chat_session().__enter__()
self.session_active = True
print("Chat session started.")
print("Chat session is already open.")
self.session.__exit__(None, None, None)
print("Chat session ended.")
print(f"Error closing the session: {e}")
self.session_active = False
print("No chat session to close.")
def is_session_open(self):
return self.session_active
def generate_response(self, prompt):
if not self.session_active:
raise RuntimeError("No active session. Please start a session first.")
return self.session.generate(prompt)
print(f"Error generating response: {e}")
def store_attributes(input_string):
if input_string.strip().lower() == "ambiguous":
return FlorenceAttributes(ambiguous=True), False, False
primary = secondary = tertiary = quaternary = None
primary_value = secondary_value = tertiary_value = quaternary_value = None
input_string = normalize_response(input_string)
lines = input_string.split('\n')
if line.startswith('PRIMARY: "'):
primary, primary_value = extract_attribute(line)
elif line.startswith('SECONDARY: "'):
secondary, secondary_value = extract_attribute(line)
elif line.startswith('TERTIARY: "'):
tertiary, tertiary_value = extract_attribute(line)
elif line.startswith('QUATERNARY: "'):
quaternary, quaternary_value = extract_attribute(line)
except (IndexError, ValueError):
raise ValueError("Invalid format")
attributes = FlorenceAttributes(primary=primary, secondary=secondary, tertiary=tertiary, quaternary=quaternary, primary_value=primary_value, secondary_value=secondary_value, tertiary_value=tertiary_value, quaternary_value=quaternary_value)
return attributes, attributes.has_null_labels(), attributes.has_null_values()
def extract_attribute(line):
prefix, rest = line.split(':', 1)
label_part, value_part = rest.split(' = ', 1)
label = label_part.split('"')[1]
value = int(value_part.strip())
except (IndexError, ValueError):
raise ValueError("Invalid attribute format")
def normalize_response(input_string):
cleaned_input = re.sub(r'[,.]', '', input_string).strip()
lines = cleaned_input.splitlines()
pattern = re.compile(r'(PRIMARY|SECONDARY|TERTIARY|QUATERNARY): ".*?" = \d+')
normalized_output = "\n".join(pattern.findall(cleaned_input))
normalized_lines = [line.strip() for line in lines]
normalized_output = "\n".join(normalized_lines)
def handle_null_labels(attributes):
if attributes.has_null_labels():
raise ValueError("One or more labels are null: " + str(attributes))
print("All labels are set: " + str(attributes))
def handle_null_values(attributes):
if attributes.has_null_values():
raise ValueError("One or more values are null: " + str(attributes))
print("All values are set: " + str(attributes))
async def classify_sentence(sentence):
print("1a---- I am in classify")
async with model.chat_session(system_template) as session:
print("1b---- I am in chat session")
response = await session.generate(prompt)
print("1d---- I generated response")
attr1 = FlorenceAttributes()
attr1, has_null_labels1, has_null_values1 = store_attributes(response)
print("!!! generated stored values: ")
print("3--- i am ambiguous ", str(attr1))
print("4--- iwill generate random ")
print("5--- Updating with random attributes...")
attr1.update_with_random_attributes()
if attr1.has_null_labels():
print("6--- -----NULL VALUE----")
print("6a--- ----FILL LABELS----")
attr1.fill_missing_labels()
print("2A1--------rensonse for WORD ")
response1 = await session.generate("Which is the most distincive adjective or noun that you can detect in the received sentence? The word must appear in the received sentence. You cannot pick silly or offensive words. Reply by telling me only this word. If there is no word simply return a blank, for example \" \".")
print("2A2--------rensonse for WORD ")
print("2A3--------rensonse for WORD ")
response1 = await session.generate("Which is the most distincive adjective or noun that you can detect in the received sentence? The word must appear in the received sentence. You cannot pick silly or offensive words. Reply by telling me only this word. If there is no word simply return a blank, for example \" \".")
print("2A4--------rensonse for WORD ")
print(f"7--- Modified Document {attr1}")
return str(attr1.finalGeneratedValues(response1)).strip(), attr1, new_message, response1
async def handle_osc_message(client, client1, unused_addr, sentence):
print(f"1---received FROM UNITY message {sentence}")
response, attribute, newMessage, wordResponse = await classify_sentence(sentence)
print(f"8--- send TO UNITY message {response}")
print(f"9--- send TO TD message {attribute}" + f"{newMessage}" + f"{wordResponse}")
client.send_message("/response", response)
client1.send_message(f"/{attribute.primary}", attribute.primary_value)
client1.send_message(f"/{attribute.secondary}", attribute.secondary_value)
client1.send_message(f"/{attribute.tertiary}", attribute.tertiary_value)
client1.send_message(f"/{attribute.quaternary}", attribute.quaternary_value)
client1.send_message("/newMessage", newMessage)
client1.send_message("/wordResponse", wordResponse)
def main(ip, port, unity_ip, unity_port):
parser = argparse.ArgumentParser()
parser.add_argument("--ip", default=ip, help="The IP to listen on")
parser.add_argument("--port", type=int, default=port, help="The port to listen on")
parser.add_argument("--unity_ip", default=unity_ip, help="The IP of the Unity client")
parser.add_argument("--unity_port", type=int, default=unity_port, help="The port of the Unity client")
args = parser.parse_args()
client = udp_client.SimpleUDPClient(args.unity_ip, args.unity_port)
client1 = udp_client.SimpleUDPClient("127.0.0.1", 9000)
disp = dispatcher.Dispatcher()
disp.map("/florence_sentence", lambda unused_addr, *args: asyncio.run(handle_osc_message(client, client1, unused_addr, *args)))
server = osc_server.ThreadingOSCUDPServer((args.ip, args.port), disp)
print(f"Serving on {server.server_address}")
if __name__ == "__main__":
main(ip, port, unity_ip, unity_port)