from gpt4all import GPT4All
from contextlib import contextmanager
from pythonosc import dispatcher
from pythonosc import osc_server
from pythonosc import udp_client
# unity_ip = "192.168.1.72"
PRIMARY: "primaryCategoryName" = primary_value,
SECONDARY: "secondaryCategoryName" = secondary_value,
TERTIARY: "tertiaryCategoryName" = tertiary_value,
QUATERNARY: "quaternaryCategoryName" = quaternary_value.
model = GPT4All(model_name='Meta-Llama-3-8B-Instruct.Q4_0.gguf', n_threads=8, device="gpu")
system_template = f'You are a system that receives a sentence about Florence Nightingale and returns her primary, secondary, tertiary and quaternary characteristics, as extracted by the meaning of the sentence. You can only pick from these four words: 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 someone says absolute nonsense, try your best to derive some meaning. if not, just say "Ambiguous". Always format your answer like that example: {output_format}. Always put the category name in quotation marks, as in the example. From now on, you will receive sentences about Florence, 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'
#region ########################### FLORENCE ATTRIBUTES #####################
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 is 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() == "AMBIGUOUS":
return FlorenceAttributes(ambiguous=True), False, False
if input_string.strip() == "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)
print("Parsed attributes:", 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+')
matches = pattern.findall(cleaned_input)
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))
# Function to classify the sentence
def classify_sentence(sentence):
print("Classifying sentence...")
with model.chat_session(system_template):
response = model.generate(prompt)
print("Response from GPT-4 All:", response)
attr1 = FlorenceAttributes()
prompt1 = "which is the most distincive adjective or noun that has to do with Florence Nightingale that you can detect in the received sentence? You cannot pick the words Florence and Nightingale. Reply by telling me only this word"
attr1, has_null_labels1, has_null_values1 = store_attributes(response)
if attr1.ambiguous is True:
print("Ambiguous response detected, generating random attributes.")
attr1.update_with_random_attributes()
if attr1.has_null_labels() is True:
attr1.fill_missing_labels()
response1 = model.generate(prompt1)
response1 = model.generate(prompt1)
print(f"Modified Document: {attr1}")
return str(attr1.finalGeneratedValues(response1)).strip(), attr1, new_message, response1
def handle_osc_message(client, client1, unused_addr, sentence):
print(f"Received message from Unity: {sentence}")
response, attribute, newMessage, wordResponse = classify_sentence(sentence)
print(f"Response: {response}")
print(f"Sending to Unity: {response}")
print(f"Sending to TouchDesigner: {attribute}, {newMessage}, {wordResponse}")
client.send_message("/response", response)
client1.send_message(f"/{attribute.primary}", f"{attribute.primary_value}")
client1.send_message(f"/{attribute.secondary}", f"{attribute.secondary_value}")
client1.send_message(f"/{attribute.tertiary}", f"{attribute.tertiary_value}")
client1.send_message(f"/{attribute.quaternary}", f"{attribute.quaternary_value}")
client1.send_message("/newMessage", f"{newMessage}")
client1.send_message("/wordResponse", f"{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: 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)