| 119 | return gpt_response(prompt, "gpt-4", 1)[0]["message"]["content"] |
| 120 | |
| 121 | class Vision_Perception_Experts: |
| 122 | def __init__(self, ram_path, current_rgb_path="current_view.jpg", view_record_path="cache_files/view_cache.json"): |
| 123 | self.current_rgb_path = current_rgb_path |
| 124 | self.view_record_path = view_record_path |
| 125 | if os.path.exists(self.view_record_path): |
| 126 | with open(self.view_record_path, "r", encoding="utf-8") as file: |
| 127 | self.view_record = json.load(file) |
| 128 | else: |
| 129 | self.view_record = {} |
| 130 | |
| 131 | while True: |
| 132 | try: |
| 133 | self.instructblip_model, self.instructblip_vis_processors, _ = load_model_and_preprocess(name="blip2_t5_instruct", model_type="flant5xl", is_eval=True, device=device) |
| 134 | self.ram_transform = get_transform(image_size=384) |
| 135 | self.ram_model = ram(pretrained=ram_path, image_size=384, vit='swin_l').eval().to(device) |
| 136 | break |
| 137 | except: |
| 138 | continue |
| 139 | |
| 140 | def ram_img_tagging(self, image): |
| 141 | ram_img = self.ram_transform(image).unsqueeze(0).to(device) |
| 142 | img_tags = inference_ram(ram_img, self.ram_model)[0] |
| 143 | |
| 144 | return img_tags |
| 145 | |
| 146 | def instructblip_description(self, image, scene_questions, img_tags): |
| 147 | instruct_blip_img = self.instructblip_vis_processors["eval"](image).unsqueeze(0).to(device) |
| 148 | prompt_list = ["Describe this indoor scene in details"] |
| 149 | |
| 150 | response_list = [] |
| 151 | for start_idx in range(0, 999, 3): |
| 152 | end_idx = len(prompt_list) if start_idx + 3 > len(prompt_list) else start_idx + 3 |
| 153 | prompts = prompt_list[start_idx:end_idx] |
| 154 | if prompts == []: |
| 155 | break |
| 156 | batch_images = torch.cat(tuple([instruct_blip_img]*len(prompts))) |
| 157 | response_list.extend(self.instructblip_model.generate({"image": batch_images, "prompt": prompts})) |
| 158 | |
| 159 | return " ".join(response_list) |
| 160 | |
| 161 | def observe_view(self, direction_idx, instruction, landmarks, navigable_vps, navigable_vps_elevation, scene_questions, history_vp): |
| 162 | raw_img = open_image(self.current_rgb_path) |
| 163 | current_vp = sim.getState()[0].location.viewpointId |
| 164 | navigable_vps_str = ", ".join(navigable_vps) |
| 165 | view_observe_idx = f"{current_vp} -> {navigable_vps_str} (Elevation 0)" |
| 166 | |
| 167 | if view_observe_idx not in self.view_record.keys(): |
| 168 | img_tags = self.ram_img_tagging(raw_img) |
| 169 | instructblip_img_info = self.instructblip_description(raw_img, scene_questions, img_tags) |
| 170 | view_observation = f"Scene Description: {instructblip_img_info} Scene Objects: {img_tags}; " |
| 171 | self.view_record[view_observe_idx] = view_observation |
| 172 | else: |
| 173 | view_observation = self.view_record[view_observe_idx] |
| 174 | |
| 175 | relative_elevation = navigable_vps_elevation[0] |
| 176 | if relative_elevation < -0.1: |
| 177 | elevation_flag = "(lower position indicates down stairs)" |
| 178 | elif relative_elevation > 0.1: |