generate_sre.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Lang2LTL version 2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prerequisite: Import libraries "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import json\n",
    "import random\n",
    "import os\n",
    "\n",
    "from srer import srer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Generate natural language commands/utterances"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load samples for utterances"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# -- load samples of LTL utterances from a CSV file using Pandas library:\n",
    "ltl_samples = pd.read_csv(open('data/symbolic_batch12_noperm_pun.csv', 'r'))\n",
    "display(ltl_samples[:10])\n",
    "\n",
    "city_jsons = os.listdir('osm/')\n",
    "selected_city = random.choice(city_jsons)\n",
    "display(selected_city)\n",
    "\n",
    "# -- load a city's landmarks from a JSON file (obtained from OpenStreetMap):\n",
    "landmarks = json.load(open(f'osm/{selected_city}', 'r', encoding='utf8'))\n",
    "display(list(landmarks.keys()))\n",
    "\n",
    "# NOTE: several prepositions are taken from Kaiyu's paper (https://h2r.cs.brown.edu/wp-content/uploads/zheng2021spatial.pdf)\n",
    "# -- each key is the phrase describing the spatial relation\n",
    "# -- a key's corresponding value indicates the number of arguments it would take\n",
    "spatial_relations = [\n",
    "    {'between': 2},\n",
    "    {'in front of': 1},\n",
    "    {'behind': 1},\n",
    "    {'next to': 1},\n",
    "    {'adjacent to': 1},\n",
    "    {'opposite': 1},\n",
    "    {'near': 1},\n",
    "    {'left of': 1},\n",
    "    {'right of': 1},\n",
    "    {'at': 1},\n",
    "    {'north of': 1},\n",
    "    {'south of': 1},\n",
    "    {'east of': 1},\n",
    "    {'west of': 1},\n",
    "]\n",
    "\n",
    "# -- list of landmarks in the city\n",
    "landmarks = list(landmarks.keys())\n",
    "\n",
    "# -- list of objects to use as reference objects:\n",
    "generic_city_objects = [\n",
    "    'bench', 'statue', 'stairs', 'store', 'door', 'bus stop', 'gas station', 'coffee shop',\n",
    "    'bicycle rack', 'parking meter', 'parking spot', 'parking space', 'stop sign',\n",
    "]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# NOTE: this is Spot experiment specific content:\n",
    "\n",
    "selected_city = 'blackstone.json'\n",
    "\n",
    "# -- load a city's landmarks from a JSON file (obtained from OpenStreetMap):\n",
    "landmarks = json.load(open(f'osm/{selected_city}', 'r', encoding='utf8'))\n",
    "all_landmarks = list(landmarks.keys())\n",
    "\n",
    "landmarks_to_use = ['Wildflour', 'Garden Grille Cafe']\n",
    "for L in all_landmarks:\n",
    "    if L not in landmarks_to_use:\n",
    "        landmarks.pop(L)\n",
    "\n",
    "# -- transforming specific names of large landmarks into generic categorical names:\n",
    "landmarks['bakery'] = landmarks.pop('Wildflour')\n",
    "landmarks['restaurant'] = landmarks.pop('Garden Grille Cafe')\n",
    "\n",
    "# NOTE: below are the objects found in experiment location:\n",
    "generic_city_objects = ['bicycle rack', 'chair', 'car']\n",
    "\n",
    "landmarks = list(landmarks.keys())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define function for generating utterances"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_synthetic_command(params, num_utterances=10, min_props=2, force_sre=False):\n",
    "    examples, landmarks, city_objects, spatial_relations = params['samples'], params[\n",
    "        'landmarks'], params['city_objects'], params['spatial_relations']\n",
    "\n",
    "    list_utterances = []\n",
    "    used_templates = []\n",
    "\n",
    "    count = 0\n",
    "\n",
    "    while count < num_utterances:\n",
    "        # -- use the eval function to get the list of props for a randomly selected row from the set of LTL blueprints:\n",
    "        random_sample = random.randint(1, len(examples))\n",
    "\n",
    "        ltl_sample = examples.iloc[random_sample-1]\n",
    "        ltl_propositions = eval(ltl_sample['props'])\n",
    "\n",
    "        if len(ltl_propositions) < min_props:\n",
    "            continue\n",
    "\n",
    "        count += 1\n",
    "\n",
    "        # display(ltl_sample)\n",
    "\n",
    "        ltl_blueprint = ltl_sample['utterance']\n",
    "\n",
    "        if not ltl_blueprint.endswith('.'):\n",
    "            ltl_blueprint += '.'\n",
    "\n",
    "        # -- save all original templates for matching them to reverse-engineered result later:\n",
    "        used_templates.append(str(ltl_blueprint))\n",
    "\n",
    "        # -- add a full-stop at the beginning and end of the sentence for easier tokenization:\n",
    "        if not ltl_blueprint.startswith('.'):\n",
    "            ltl_blueprint = '.' + ltl_blueprint\n",
    "\n",
    "        for x in range(len(ltl_propositions)):\n",
    "\n",
    "            new_entity = None\n",
    "\n",
    "            # -- flip a coin to see if we will use a landmark or an object (potentially) near to the landmark:\n",
    "            use_spatial_rel = random.randint(1, 2)\n",
    "\n",
    "            # NOTE: if force_sre is switched to true, then it will generate propositions with all spatial referring expressions\n",
    "\n",
    "            if use_spatial_rel > 1 or force_sre:\n",
    "                # -- we will randomly select a spatial relation, breaking it down to its phrase and number of args:\n",
    "                spatial_rel = random.choice(spatial_relations)\n",
    "\n",
    "                # -- this is the spatial relation phrase/expression\n",
    "                spatial_key = list(spatial_rel.keys())[-1]\n",
    "                # -- this is the number of arguments it accepts\n",
    "                spatial_args = spatial_rel[spatial_key]\n",
    "\n",
    "                # NOTE: so far, we only have spatial relations with either 1 or 2 arguments.\n",
    "                # -- are there possibly some with more?\n",
    "                if spatial_args == 2:\n",
    "                    # -- this is if we have a spatial relation with 2 arguments (e.g., between):\n",
    "\n",
    "                    # -- randomly select 2 landmarks without replacement with which we will make a SRE:\n",
    "                    landmark_samples = random.sample(landmarks, 2)\n",
    "\n",
    "                    new_entity = f' the {random.choice(city_objects)} {spatial_key} {landmark_samples[0]} and {landmark_samples[1]}'\n",
    "                elif spatial_args == 1:\n",
    "                    # -- this is if we have a regular spatial relation with a single argument:\n",
    "\n",
    "                    # -- randomly select only a single landmark:\n",
    "                    new_entity = f' the {random.choice(city_objects)} {spatial_key} {random.choice(landmarks)}'\n",
    "            else:\n",
    "                # -- in this case, we will just select a landmark from the entire set in the city:\n",
    "                new_entity = f' {random.choice(landmarks)}'\n",
    "\n",
    "            # NOTE: to do replacement of the lifted proposition with the generated one, we need to account for\n",
    "            # different ways it would be written preceded by a whitespace character, i.e., ' a ', ' a,', ' a.'\n",
    "\n",
    "            # -- we will replace the proposition in the lifted expression with the grounded entity:\n",
    "            props_to_replace = [(f' {ltl_propositions[x]}' + y) for y in [' ', '.', ',']] + [\n",
    "                ('.' + f'{ltl_propositions[x]} '), ('.' + f'{ltl_propositions[x]},')]\n",
    "            for prop in props_to_replace:\n",
    "                # -- only replace if it was found:\n",
    "                if prop in ltl_blueprint:\n",
    "                    ltl_blueprint = ltl_blueprint.replace(\n",
    "                        prop, new_entity + prop[-1])\n",
    "\n",
    "            # NOTE: some utterances will be missing some propositions\n",
    "\n",
    "        list_utterances.append(ltl_blueprint[1:])\n",
    "\n",
    "    return used_templates, list_utterances\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "generation_params = {\n",
    "\t'samples': ltl_samples,\n",
    "\t'landmarks': landmarks,\n",
    "\t'city_objects': generic_city_objects,\n",
    "\t'spatial_relations': spatial_relations\n",
    "}\n",
    "\n",
    "num_commands = 4\n",
    "\n",
    "# -- forcing minimum number of propositions to be 5, forcing all SREs for propositions:\n",
    "used_templates, utterances = generate_synthetic_command(generation_params, num_utterances=num_commands, min_props=2, force_sre=True)\n",
    "\n",
    "for N in range(num_commands):\n",
    "\tprint(f'{utterances[N]}\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Prompt LLM for spatial referring expressions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "count = 0\n",
    "\n",
    "for command in utterances:\n",
    "    raw_output, parsed_output = srer(command)\n",
    "\n",
    "    parsed_output['lifted_gtr'] = used_templates[count]\n",
    "\n",
    "    count += 1\n",
    "\n",
    "    print(f'Command: {command}\\n{json.dumps(parsed_output,indent=2)}')\n",
    "    # display(parsed_output['command'])\n",
    "    print(f\"Groundtruth matched?: {parsed_output['lifted_llm'] == parsed_output['lifted_gtr']}\" )\n",
    "    print(f\"-> LLM Lifted Translation: {parsed_output['lifted_llm']}\")\n",
    "    print(f\"-> Groundtruth Lifted Translation: {parsed_output['lifted_gtr']}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "utterances = list(filter(None, [X.strip() for X in open('blackstone_commands.txt', 'r').readlines()]))\n",
    "\n",
    "print(utterances)\n",
    "\n",
    "count = 0\n",
    "\n",
    "all_llm_output = []\n",
    "\n",
    "for command in utterances:\n",
    "    raw_output, parsed_output = referring_exp_recognition(command)\n",
    "\n",
    "    # parsed_output['lifted_gtr'] = used_templates[count]\n",
    "\n",
    "    count += 1\n",
    "\n",
    "    print(f'Command: {command}\\n{json.dumps(parsed_output,indent=2)}')\n",
    "    # display(parsed_output['command'])\n",
    "\n",
    "    all_llm_output.append(parsed_output)\n",
    "\n",
    "json.dump(all_llm_output, open('blackstone_srer_output.json', 'w'))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}