Move PULS code

neus_v/puls/llm.py

@@ -0,0 +1,58 @@
+import datetime
+import json
+import os
+
+
+class LLM:
+    def __init__(self, client, save_dir="outputs"):
+        self.client = client
+        self.history = []
+        self.save_dir = save_dir
+        if save_dir:
+            os.makedirs(save_dir, exist_ok=True)
+
+    def prompt(self, p):
+        # Add user message to history
+        user_message = {"role": "user", "content": [{"type": "text", "text": p}]}
+        self.history.append(user_message)
+
+        # Create messages list with history
+        messages = self.history.copy()
+
+        response = self.client.chat.completions.create(
+            model="o1-mini-2024-09-12",
+            messages=self.history,
+            store=False,
+        )
+
+        # Extract assistant response
+        assistant_response = response.choices[0].message.content
+
+        # Add assistant response to history
+        assistant_message = {"role": "assistant", "content": [{"type": "text", "text": assistant_response}]}
+        self.history.append(assistant_message)
+
+        return assistant_response
+
+    def save_history(self, filename="conversation_history.json"):
+        """Save conversation history to a JSON file"""
+        if not self.save_dir:
+            return
+
+        # Add timestamp to filename
+        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+        base_name, extension = os.path.splitext(filename)
+        timestamped_filename = f"{base_name}_{timestamp}{extension}"
+
+        save_path = os.path.join(self.save_dir, timestamped_filename)
+        try:
+            with open(save_path, "w", encoding="utf-8") as f:
+                json.dump(self.history, f, indent=4, ensure_ascii=False)
+            print(f"Conversation history saved to: {save_path}")
+        except Exception as e:
+            print(f"Failed to save conversation history: {e}")
+
+    def __del__(self):
+        """Destructor: Saves conversation history when the object is destroyed"""
+        if self.save_dir:
+            self.save_history()

neus_v/puls/main.py

@@ -0,0 +1,28 @@
+from puls import *
+
+import argparse
+import json
+import time
+
+def main():
+    prompt = "a dog sits on a mat while a ball rolls past the dog towards a couch, then the dog picks up the ball and places it beside the couch"
+    modes = [Mode.OBJECT_ACTION_ALIGNMENT, Mode.OVERALL_CONSISTENCY, Mode.OBJECT_EXISTENCE, Mode.SPATIAL_RELATIONSHIP]
+
+    parser = argparse.ArgumentParser(description='Set OpenAI API Key.')
+    parser.add_argument('--openai_key', type=str, help='Your OpenAI API key')
+    args = parser.parse_args()
+    
+    start_time = time.time()
+    if args.openai_key:
+        data = PULS(prompt, modes, args.openai_key)
+    else:
+        data = PULS(prompt, modes)
+    end_time = time.time()
+
+    print(prompt)
+    print(json.dumps(data, indent=2))
+    print(f"Elapsed Time: {end_time - start_time}")
+    
+if __name__ == "__main__":
+    main()
+

neus_v/puls/prompts.py

@@ -0,0 +1,85 @@
+from enum import Enum
+
+class Mode(Enum):
+    OBJECT_ACTION_ALIGNMENT = 1
+    OBJECT_EXISTENCE = 2
+    OVERALL_CONSISTENCY = 3
+    SPATIAL_RELATIONSHIP = 4
+
+mode_prompts = {
+    Mode.OBJECT_ACTION_ALIGNMENT: (
+        "\"object_action_alignment\":\n"
+        "Extract actions and their participating objects. Each proposition must describe an action and its related objects.\n"
+        "Example:\n"
+        "\"object_action_alignment\": [\"person holds hotdog\", \"person walks\"]"
+    ),
+    Mode.OBJECT_EXISTENCE: (
+        "\"object_existence\":\n"
+        "Extract only the tangible objects mentioned in the prompt.\n"
+        "Example:\n"
+        "\"object_existence\": [\"person\", \"hotdog\", \"car\", \"truck\"]"
+    ),
+    Mode.OVERALL_CONSISTENCY: (
+        "\"overall_consistency\":\n"
+        "Extract all meaningful event propositions that describe the combined semantics of objects, actions, and spatial relationships — "
+        "but avoid TL keywords such as 'and', 'or', 'not', 'until', 'eventually'.\n"
+        "Example:\n"
+        "\"overall_consistency\": [\"person holds hotdog\", \"person walks\", \"car next to truck\"]"
+    ),
+    Mode.SPATIAL_RELATIONSHIP: (
+        "\"spatial_relationships\":\n"
+        "Extract only spatial relationships between tangible objects (e.g., \"object A next to object B\"). Do not infer or hallucinate spatial relationships.\n"
+        "Example:\n"
+        "\"spatial_relationships\": [\"car next to truck\"]"
+    )
+}
+
+mode_outputs = {
+    Mode.OBJECT_ACTION_ALIGNMENT: (
+        "  \"object_action_alignment\": [...],\n"
+        "  \"object_action_alignment_spec\": \"...\","
+    ),
+    Mode.OBJECT_EXISTENCE: (
+        "  \"object_existence\": [...],\n"
+        "  \"object_existence_spec\": \"...\","
+    ),
+    Mode.OVERALL_CONSISTENCY: (
+        "  \"overall_consistency\": [...],\n"
+        "  \"overall_consistency_spec\": \"...\","
+    ),
+    Mode.SPATIAL_RELATIONSHIP: (
+        "  \"spatial_relationships\": [...],\n"
+        "  \"spatial_relationships_spec\": \"...\""
+    )
+}
+
+header = (
+    "You are an intelligent agent designed to extract structured representations from video description prompts. "
+    "You will operate in two stages: (1) proposition extraction and (2) TL specification generation.\n\n"
+)
+
+stage1_intro = (
+    "Stage 1: Proposition Extraction\n\n"
+    "Given an input prompt summarizing a video, extract atomic propositions in the following four modes. "
+    "Return all outputs in JSON format.\n\n"
+)
+
+stage2_intro = "Stage 2: TL Specification Generation\n\n"
+
+spec_gen_intro = (
+    "For each of the {n} list(s) of propositions extracted in Stage 1, generate a separate Temporal Logic (TL) specification "
+    "describing the structure or sequence of events in that list.\n\n"
+)
+
+tl_instructions = (
+    "Rules for TL specification:\n"
+    "- The input is a single list of propositions from one of the extraction modes.\n"
+    "- The output is a single TL formula using **only** the propositions from that list and the allowed TL symbols: "
+    "['AND', 'OR', 'NOT', 'UNTIL', 'ALWAYS', 'EVENTUALLY']\n"
+    "- Do not introduce any new propositions.\n"
+    "- Each formula should reflect the temporal or logical relationships between the propositions in a way that makes semantic sense.\n\n"
+)
+
+input_template = "Input:\n{{\n  \"prompt\": \"{}\"\n}}\n\n"
+expected_output_header = "Expected Output:\n{\n"
+expected_output_footer = "\n}"

neus_v/puls/puls.py

@@ -0,0 +1,80 @@
+from llm import *
+from openai import OpenAI
+from prompts import *
+import json
+import os
+import re
+
+def clean_and_parse_json(raw_str):
+    start = raw_str.find('{')
+    end = raw_str.rfind('}') + 1
+    json_str = raw_str[start:end]
+    return json.loads(json_str)
+
+def process_specification(specification, propositions):
+    new_propositions = []
+    for prop in propositions:
+        prop_cleaned = re.sub(r"^[^a-zA-Z]+|[^a-zA-Z]+$", "", prop)
+        prop_cleaned = re.sub(r"\s+", "_", prop_cleaned)
+        new_propositions.append(prop_cleaned)
+
+    for original, new in zip(propositions, new_propositions):
+        specification = specification.replace(original, f'"{new}"')
+
+    replacements = {
+        "AND": "&",
+        "OR": "|",
+        "UNTIL": "U",
+        "ALWAYS": "G",
+        "EVENTUALLY": "F",
+        "NOT": "!"
+    }
+    for word, symbol in replacements.items():
+        specification = specification.replace(word, symbol)
+
+    return new_propositions, specification
+
+def create_prompt(prompt, modes):
+    full_prompt = header + stage1_intro
+
+    for i, m in enumerate(modes, start=1):
+        full_prompt += f"{i}. {mode_prompts[m]}\n\n"
+    
+    full_prompt += stage2_intro
+    full_prompt += spec_gen_intro.format(n=len(modes))
+    full_prompt += tl_instructions
+    full_prompt += input_template.format(prompt)
+    
+    full_prompt += expected_output_header
+    for i, m in enumerate(modes):
+        full_prompt += mode_outputs[m]
+        if i != len(modes) - 1:
+            full_prompt += "\n\n"
+    full_prompt += expected_output_footer
+
+    return full_prompt
+
+def PULS(prompt, modes, openai_key=None):
+    if openai_key:
+        os.environ["OPENAI_API_KEY"] = openai_key
+    
+    client = OpenAI()
+    llm = LLM(client)
+
+    full_prompt = create_prompt(prompt, modes)
+    llm_output = llm.prompt(full_prompt)
+    parsed = clean_and_parse_json(llm_output)
+
+    final_output = {}
+
+    for key, value in parsed.items():
+        if key.endswith("_spec"):
+            base_key = key.replace("_spec", "")
+            propositions = parsed.get(base_key, [])
+            cleaned_props, processed_spec = process_specification(value, propositions)
+            final_output[base_key] = cleaned_props
+            final_output[key] = processed_spec
+        else:
+            final_output[key] = value
+
+    return final_output