| 1 | from FilmAgent_root.FilmAgent.util import * |
| 2 | from FilmAgent_root.FilmAgent.LLMCaller import * |
| 3 | from typing import Dict, List, Union |
| 4 | import random |
| 5 | import copy |
| 6 | |
| 7 | # TO DO |
| 8 | ROOT_PATH = "/path/to/FilmAgent" |
| 9 | ID = 13 |
| 10 | model = "gpt-4o" |
| 11 | # TO DO |
| 12 | |
| 13 | topics=["Reconcilation in a friend reunion", "A quarrel and breakup scene", "Casual meet-up with an old friend", "Emergency meeting after a security breach", "Late night brainstorming for a startup", "Family argument during dinner", "Emotional farewell at the roadside", "Heated debate over investments in the office", "Heated family discussion ending in a heartfelt apology", "Office gossip turning into a major understanding", "Celebratory end of project cheers with team members", "Planning a secret escape from a mundane routine", "Unexpected guest crashes a small house party", "An employee's emotional breakdown after being terminated", "Confession of a long-held secret among close friends"] |
| 14 | |
| 15 | class FilmCrafter: |
| 16 | |
| 17 | def __init__(self, topic: str) -> None: |
| 18 | self.topic = topic |
| 19 | self.store_path = os.path.join(ROOT_PATH, f"store\\no_interation\{ID}") |
| 20 | self.log_path = os.path.join(self.store_path, "prompt.txt") |
| 21 | self.profile_path = os.path.join(self.store_path, "actors_profile.json") |
| 22 | self.action_description_path = os.path.join(ROOT_PATH, "Locations\\actions.txt") |
| 23 | self.shot_description_path = os.path.join(ROOT_PATH, "Locations\\shots.txt") |
| 24 | # scenes |
| 25 | self.scene_path = os.path.join(self.store_path, "scenes_1.json") |
| 26 | # + lines |
| 27 | self.scene_path_1 = os.path.join(self.store_path, "scenes_2.json") |
| 28 | # + positions |
| 29 | self.scene_path_2 = os.path.join(self.store_path, "scenes_3.json") |
| 30 | # + actions |
| 31 | self.scene_path_3 = os.path.join(self.store_path, "scenes_4.json") |
| 32 | # + movement |
| 33 | self.scene_path_4 = os.path.join(self.store_path, "scenes_5.json") |
| 34 | # + shot (stage3_verify) |
| 35 | self.scene_path_5 = os.path.join(self.store_path, "scenes_6.json") |
| 36 | # The final script |
| 37 | self.script_path = os.path.join(self.store_path, "script.json") |
| 38 | |
| 39 | # The maximum number of characters in a film |
| 40 | self.character_limit = 4 |
| 41 | # The maximum number of scenes in a film |
| 42 | self.scene_limit = 3 |
| 43 | # The maximum number of discussions between director and screenwriter |
| 44 | self.stage1_verify_limit = 3 |
| 45 | # The maximum number of discussions between director, actor and screenwriter |
| 46 | self.stage2_verify_limit = 3 |
| 47 | # The maximum number of discussions between director and cinematographer |
| 48 | self.stage3_verify_limit = 3 |
| 49 | |
| 50 | if not os.path.exists(self.store_path): |
| 51 | os.makedirs(self.store_path) |
| 52 | |
| 53 | |
| 54 | def call(self, identity: str, params: Dict, trans2json: bool = True) -> Union[str, dict, list]: |
| 55 | prompt = read_prompt(os.path.join(ROOT_PATH, f"Prompt\{identity}.txt") ) |
| 56 | prompt = prompt_format(prompt, params) |
| 57 | log_prompt(self.log_path, prompt) |
| 58 | result = LLMCall(prompt, model) |
| 59 | if trans2json: |
| 60 | result = clean_text(result) |
| 61 | result = GPTResponse2JSON(result) |
| 62 | log_prompt(self.log_path, result) |
| 63 | return result |
| 64 | |
| 65 | |
| 66 | def casting(self): |
| 67 | ''' |
| 68 | Role: Director |
| 69 | |
| 70 | Behavior: Create the main characters and their bios for the film script. |
| 71 | ''' |
| 72 | params = {"{topic}": self.topic, "{character_limit}": self.character_limit} |
| 73 | result = self.call("director_1", params) |
| 74 | write_json(self.profile_path, result) |
| 75 | |
| 76 | |
| 77 | def scenes_plan(self): |
| 78 | ''' |
| 79 | Role: Director |
| 80 | |
| 81 | Behavior: |
| 82 | Plan the outline of script, include: |
| 83 | 1. The number of scenes. |
| 84 | 2. The characters, location and main plot of each scene. |
| 85 | ''' |
| 86 | profile = read_json(self.profile_path) |
| 87 | male_characters = ", ".join(list(map(lambda x: x['name'], |
| 88 | filter(lambda x: x['gender'].lower() == 'male', profile)))) |
| 89 | female_characters = ", ".join(list(map(lambda x: x['name'], |
| 90 | filter(lambda x: x['gender'].lower() == 'female', profile)))) |
| 91 | |
| 92 | params = {"{topic}": self.topic, |
| 93 | "{male_characters}": male_characters, |
| 94 | "{female_characters}": female_characters, |
| 95 | "{scene_limit}": self.scene_limit} |
| 96 | result = self.call("director_2", params) |
| 97 | write_json(self.scene_path, result) |
| 98 | |
| 99 | |
| 100 | def lines_generate(self): |
| 101 | ''' |
| 102 | Role: Screenwriter |
| 103 | |
| 104 | Behavior: Write lines for the script. |
| 105 | ''' |
| 106 | scenes = read_json(self.scene_path) |
| 107 | script_outline = "" |
| 108 | who = [] |
| 109 | where = [] |
| 110 | what = [] |
| 111 | for id,scene in enumerate(scenes): |
| 112 | selected_roles = scene[return_most_similar("selected-characters", list(scene.keys()))] |
| 113 | selected_location = scene[return_most_similar("selected-location", list(scene.keys()))] |
| 114 | story_plot = scene[return_most_similar("story-plot", list(scene.keys()))] |
| 115 | who.append(selected_roles) |
| 116 | where.append(selected_location) |
| 117 | what.append(story_plot) |
| 118 | |
| 119 | topic = scene[return_most_similar("sub-topic", list(scene.keys()))] |
| 120 | characters = ", ".join(selected_roles) |
| 121 | plot = story_plot |
| 122 | location = selected_location |
| 123 | goal = scene[return_most_similar("dialogue-goal", list(scene.keys()))] |
| 124 | |
| 125 | script_outline = script_outline + f"{id + 1}. **Scene {id + 1}**:\n - topic: {topic}\n - involved characters: {characters}\n - plot: {plot}\n - location: {location}\n - dialogue goal: {goal}\n\n" |
| 126 | |
| 127 | params = {"{script_outline}": script_outline.strip()} |
| 128 | result = self.call("screenwriter_1", params) |
| 129 | |
| 130 | lines = [] |
| 131 | assert len(result) == len(scenes) |
| 132 | for j in range(len(scenes)): |
| 133 | line = {} |
| 134 | line['scene_information'] = {} |
| 135 | line['scene_information']['who'] = who[j] |
| 136 | line['scene_information']['where'] = where[j] |
| 137 | line['scene_information']['what'] = what[j] |
| 138 | line['dialogues'] = result[j][return_most_similar("scene-dialogue", list(result[j].keys()))] |
| 139 | lines.append(line) |
| 140 | write_json(self.scene_path_1, lines) |
| 141 | |
| 142 | |
| 143 | |
| 144 | def position_mark(self): |
| 145 | ''' |
| 146 | Role: Screenwriter |
| 147 | |
| 148 | Behavior: Choose an appropriate initial position for each character in each scene of the script. |
| 149 | ''' |
| 150 | scenes = read_json(self.scene_path_1) |
| 151 | script_information = "" |
| 152 | optional_positions = "" |
| 153 | for id,scene in enumerate(scenes): |
| 154 | i = id + 1 |
| 155 | who = scene['scene_information']['who'] |
| 156 | where = scene['scene_information']['where'] |
| 157 | what = scene['scene_information']['what'] |
| 158 | |
| 159 | script_information = script_information + f"{i}. **Scene {i}**:\n - characters: {who}\n - location: {where}\n - plot: {what}\n\n" |
| 160 | |
| 161 | position_path = os.path.join(ROOT_PATH, f"Locations\{where}\position.json") |
| 162 | positions = read_json(position_path) |
| 163 | normal_position = [item for item in positions if item['fixed_angle'] == False] |
| 164 | # This "if judgment" is related to the position, and camera settings in Unity. |
| 165 | if len(who) >= len(positions) - len(normal_position) + 2: |
| 166 | p = "" |
| 167 | for it,position in enumerate(positions): |
| 168 | j = it + 1 |
| 169 | p = p + f" - Position {j}: " + position['description'] + '\n' |
| 170 | else: |
| 171 | p = "" |
| 172 | for it,position in enumerate(normal_position): |
| 173 | j = it + 1 |
| 174 | p = p + f" - Position {j}: " + position['description'] + '\n' |
| 175 | optional_positions = optional_positions + f"{i}. **Positions in {where}**:\n{p}\n" |
| 176 | |
| 177 | params = {"{script_information}": script_information.strip(), |
| 178 | "{optional_positions}": optional_positions.strip()} |
| 179 | result = self.call("screenwriter_2", params) |
| 180 | |
| 181 | assert len(result) == len(scenes) |
| 182 | for j in range(len(scenes)): |
| 183 | scenes[j]["initial position"] = result[j][return_most_similar("scene-position", list(result[j].keys()))] |
| 184 | write_json(self.scene_path_2, scenes) |
| 185 | |
| 186 | |
| 187 | |
| 188 | def action_mark(self): |
| 189 | ''' |
| 190 | Role: Screenwriter |
| 191 | |
| 192 | Behavior: Choose appropriate actions for the characters engaged in the dialogue. |
| 193 | ''' |
| 194 | scenes = read_json(self.scene_path_2) |
| 195 | all_actions = read_prompt(self.action_description_path) |
| 196 | data = [] |
| 197 | for scene in scenes: |
| 198 | position_path = os.path.join(ROOT_PATH, f"Locations\{scene['scene_information']['where']}\position.json") |
| 199 | positions = read_json(position_path) |
| 200 | |
| 201 | ini = "" |
| 202 | for id,item in enumerate(scene['initial position']): |
| 203 | if [it['sittable'] for it in positions if get_number(it['id']) == get_number(item['position'])][0]: |
| 204 | sit = "sittable" |
| 205 | else: |
| 206 | sit = "unsittable" |
| 207 | ini = ini + f" - {item['character']}: " + f"{sit} {item['position']}, standing\n" |
| 208 | ini = " " + ini.strip() |
| 209 | params = {"{initial}": ini, |
| 210 | "{plot}": scene['scene_information']['what'], |
| 211 | "{dialogues}": scene['dialogues'], |
| 212 | "{all_actions}": all_actions} |
| 213 | result = self.call("screenwriter_3", params) |
| 214 | |
| 215 | assert len(result) == len(scene['dialogues']) |
| 216 | scene['dialogues'] = result |
| 217 | data.append(scene) |
| 218 | |
| 219 | write_json(self.scene_path_3, data) |
| 220 | |
| 221 | |
| 222 | |
| 223 | def is_keep_standing(self, lines, character): |
| 224 | ''' |
| 225 | Description: Check if the character remains standing throughout the script. |
| 226 | |
| 227 | Input: The complete script, a character's name |
| 228 | |
| 229 | Output: True or False |
| 230 | ''' |
| 231 | for line in lines: |
| 232 | actions = line['actions'] |
| 233 | for action in actions: |
| 234 | if action['character'] == character and action['state'] == 'sitting': |
| 235 | return False |
| 236 | return True |
| 237 | |
| 238 | |
| 239 | def moveable_options(self, scene): |
| 240 | ''' |
| 241 | Input: The complete script |
| 242 | |
| 243 | Output: |
| 244 | 1. All movable characters (i.e., the characters that remain standing throughout the script) |
| 245 | 2. All positions that character can move to (i.e., the unoccupied positions) |
| 246 | ''' |
| 247 | position_path = os.path.join(ROOT_PATH, f"Locations\{scene[return_most_similar('scene_information', list(scene.keys()))]['where']}\position.json") |
| 248 | positions = read_json(position_path) |
| 249 | occupied_positions = [get_number(item['position']) for item in scene[return_most_similar('initial position', list(scene.keys()))]] |
| 250 | unoccupied_positions = [f"{item['id']}: {item['description']}" for item in positions if get_number(item['id']) not in occupied_positions] |
| 251 | if len(unoccupied_positions) == 0: |
| 252 | return None, None |
| 253 | |
| 254 | who = scene[return_most_similar('scene_information', list(scene.keys()))]['who'] |
| 255 | moveable_characters = [] |
| 256 | for character in who: |
| 257 | if self.is_keep_standing(scene['dialogues'], character): |
| 258 | moveable_characters.append(character) |
| 259 | if len(moveable_characters) == 0: |
| 260 | return None, None |
| 261 | |
| 262 | return moveable_characters, unoccupied_positions |
| 263 | |
| 264 | |
| 265 | def move_mark(self): |
| 266 | ''' |
| 267 | Role: Director |
| 268 | |
| 269 | Behavior: The director adds appropriate character movements into the script. |
| 270 | ''' |
| 271 | scenes = read_json(self.scene_path_3) |
| 272 | data = [] |
| 273 | for scene in scenes: |
| 274 | moveable_characters, unoccupied_positions = self.moveable_options(scene) |
| 275 | moved_charatcter, moveto_position = None, None |
| 276 | if moveable_characters: |
| 277 | move2destination = "" |
| 278 | for pn in unoccupied_positions: |
| 279 | move2destination = move2destination + f" - {pn}\n" |
| 280 | move2destination = " " + move2destination.strip() |
| 281 | lines = [] |
| 282 | for id in range(len(scene['dialogues'])): |
| 283 | lines.append(f"<Insertion Position {id}>") |
| 284 | item = {} |
| 285 | item['speaker'] = scene['dialogues'][id]['speaker'] |
| 286 | item['content'] = scene['dialogues'][id]['content'] |
| 287 | lines.append(item) |
| 288 | params = {"{moveable_characters}": moveable_characters, |
| 289 | "{story}": scene[return_most_similar('scene_information', list(scene.keys()))]['what'], |
| 290 | "{lines}": lines, |
| 291 | "{destinations}": move2destination, |
| 292 | "{current_positions}": scene[return_most_similar('initial position', list(scene.keys()))]} |
| 293 | result = self.call("director_7", params) |
| 294 | |
| 295 | if 'insertion' in result.keys(): |
| 296 | moved_charatcter = result['move']['character'] |
| 297 | moveto_position = result['move']['destination'] |
| 298 | scene['dialogues'].insert(get_number(result['insertion'][return_most_similar('insertion position', list(result['insertion'].keys()))]), result) |
| 299 | |
| 300 | # Update the characters' position in real time |
| 301 | position_change = False |
| 302 | for line in scene['dialogues']: |
| 303 | if "move" not in line.keys(): |
| 304 | line['current position'] = scene[return_most_similar('initial position', list(scene.keys()))] |
| 305 | else: |
| 306 | position_change = True |
| 307 | line['current position'] = scene[return_most_similar('initial position', list(scene.keys()))] |
| 308 | continue |
| 309 | if position_change: |
| 310 | line['current position'] = [item if item['character'] != moved_charatcter else {'character': moved_charatcter, 'position': moveto_position} for item in line['current position']] |
| 311 | |
| 312 | data.append(scene) |
| 313 | |
| 314 | write_json(self.scene_path_4, data) |
| 315 | |
| 316 | |
| 317 | def shot_mark(self): |
| 318 | ''' |
| 319 | Input: None |
| 320 | |
| 321 | Output: |
| 322 | 1. Director's shot annotations |
| 323 | 2. Cinematographer's shot annotations |
| 324 | 3. The complete script before adding shot annotations |
| 325 | 4. The script after inserting the shot annotation points. |
| 326 | ''' |
| 327 | scenes = read_json(self.scene_path_4) |
| 328 | script = {} |
| 329 | for ID,scene in enumerate(scenes): |
| 330 | I = ID + 1 |
| 331 | script[f'scene {I}'] = [] |
| 332 | for id,item in enumerate(scene['dialogues']): |
| 333 | line = {} |
| 334 | i = id + 1 |
| 335 | if "speaker" in item.keys(): |
| 336 | line['dialogue'] = f"{item['speaker']}: {item['content']}" |
| 337 | line['actions'] = item['actions'] |
| 338 | # line['plot'] = ' '.join([it['reason'] for it in item['actions']]) |
| 339 | line[f'selected-shot-{i}'] = "..." |
| 340 | else: |
| 341 | line['move'] = {} |
| 342 | line['move']['character'] = item['move']['character'] |
| 343 | line['move']['destination'] = item['move']['destination'] |
| 344 | line[f'selected-shot-{i}'] = "..." |
| 345 | script[f'scene {I}'].append(line) |
| 346 | |
| 347 | all_shots = read_prompt(self.shot_description_path) |
| 348 | params = {"{script}": script, "{all_shots}": all_shots} |
| 349 | result1 = self.call("cinematographer", params) |
| 350 | for scene_id, scene in result1.items(): |
| 351 | for shot_id, shot in scene.items(): |
| 352 | shot.pop("reasoning") |
| 353 | |
| 354 | assert len(list(result1.keys())) == len(scenes) |
| 355 | for id in range(len(list(result1.keys()))): |
| 356 | i = id + 1 |
| 357 | assert len(list(result1[f'scene {i}'].keys())) == len(scenes[id]['dialogues']) |
| 358 | for key,value in result1[f'scene {i}'].items(): |
| 359 | scenes[id]['dialogues'][get_number(key)-1]['selected shot'] = value[return_most_similar('shot', list(value.keys()))] |
| 360 | |
| 361 | write_json(self.scene_path_5, scenes) |
| 362 | |
| 363 | |
| 364 | |
| 365 | # Used for clean_script() |
| 366 | def process_action(self, actions, v_characters, v_actions): |
| 367 | new_actions = [] |
| 368 | for item in actions: |
| 369 | new_item = {} |
| 370 | new_item['character'] = return_most_similar(item['character'], v_characters) |
| 371 | new_item['state'] = return_most_similar(item['state'], ['standing', 'sitting']) |
| 372 | |
| 373 | new_item['action'] = return_most_similar(item['action'], v_actions) |
| 374 | if new_item['state'] == "standing" and ("Standing" in new_item['action'] or new_item['action'] == "Joyful Jump" or new_item['action'] == "Sit Down"): |
| 375 | if new_item['action'] == "Standing Talking": |
| 376 | new_item['action'] = "Standing Talking " + str(random.randint(1, 6)) |
| 377 | if new_item['action'] == "Standing Angry": |
| 378 | new_item['action'] = "Standing Angry " + str(random.randint(1, 4)) |
| 379 | if new_item['action'] == "Standing Arguing": |
| 380 | new_item['action'] = "Standing Arguing " + str(random.randint(1, 2)) |
| 381 | if new_item['action'] == "Standing Agree": |
| 382 | new_item['action'] = "Standing Agree " + str(random.randint(1, 2)) |
| 383 | new_actions.append(new_item) |
| 384 | elif new_item['state'] == "sitting" and ("Sitting" in new_item['action'] or new_item['action'] == "Stand Up"): |
| 385 | if new_item['action'] == "Sitting Talking": |
| 386 | new_item['action'] = "Sitting Talking " + str(random.randint(1, 2)) |
| 387 | new_actions.append(new_item) |
| 388 | else: |
| 389 | pass |
| 390 | return new_actions |
| 391 | |
| 392 | |
| 393 | # Used for clean_script() |
| 394 | def process_shot(self, info, location, shot, v_shots): |
| 395 | shot = return_most_similar(shot, v_shots) |
| 396 | if shot == "Pan Shot": |
| 397 | return "Pan Shot 1" |
| 398 | elif shot == "Track Shot": |
| 399 | return "Track Shot " + str(random.randint(1, int(info[location]['track']))) |
| 400 | elif shot == "Long Shot": |
| 401 | return "Long Shot " + str(random.randint(1, int(info[location]['long']))) |
| 402 | else: |
| 403 | return shot |
| 404 | |
| 405 | |
| 406 | def clean_script(self): |
| 407 | ''' |
| 408 | Description: Only keep the necessary information in the script and perform certain checks to avoid errors when executing the script in Unity. |
| 409 | ''' |
| 410 | scenes = read_json(self.scene_path_5) |
| 411 | profiles = read_json(self.profile_path) |
| 412 | v_characters = [item['name'] for item in profiles] |
| 413 | info = read_json(os.path.join(ROOT_PATH, "Locations\\rotateandtrack.json")) |
| 414 | v_locations = [location for location in info.keys()] |
| 415 | info_1 = read_json(os.path.join(ROOT_PATH, "Locations\\actions.json")) |
| 416 | v_actions = [action for action in info_1.keys()] |
| 417 | info_2 = read_json(os.path.join(ROOT_PATH, "Locations\\shots.json")) |
| 418 | v_shots = [shot for shot in info_2.keys()] |
| 419 | |
| 420 | data = [] |
| 421 | for scene in scenes: |
| 422 | new_scene = {} |
| 423 | scene_information_key = return_most_similar('scene_information', list(scene.keys())) |
| 424 | new_scene['scene information'] = scene[scene_information_key] |
| 425 | |
| 426 | # verify |
| 427 | for role in new_scene['scene information']['who']: |
| 428 | role = return_most_similar(role, v_characters) |
| 429 | new_scene['scene information']['where'] = return_most_similar(new_scene['scene information']['where'], v_locations) |
| 430 | # verify |
| 431 | |
| 432 | new_scene['scene'] = [] |
| 433 | for line in scene['dialogues']: |
| 434 | new_line = {} |
| 435 | selected_shot_key = return_most_similar('selected shot', list(line.keys())) |
| 436 | current_position_key = return_most_similar('current position', list(line.keys())) |
| 437 | # verify |
| 438 | if 'speaker' in line.keys(): |
| 439 | new_line['speaker'] = return_most_similar(line['speaker'], v_characters) |
| 440 | new_line['content'] = line['content'] |
| 441 | new_line['actions'] = self.process_action(line['actions'], v_characters, v_actions) |
| 442 | new_line['shot'] = self.process_shot(info, scene[scene_information_key]['where'], line[selected_shot_key], v_shots) |
| 443 | new_line['current position'] = line[current_position_key] |
| 444 | else: |
| 445 | new_line['move'] = {} |
| 446 | new_line['move']['character'] = return_most_similar(line['move']['character'], v_characters) |
| 447 | new_line['move']['destination'] = "Position " + str(get_number(line['move']['destination'])) |
| 448 | new_line['shot'] = self.process_shot(info, scene[scene_information_key]['where'], line[selected_shot_key], v_shots) |
| 449 | new_line['current position'] = line[current_position_key] |
| 450 | # verify |
| 451 | |
| 452 | # verify |
| 453 | for item in new_line['current position']: |
| 454 | item['character'] = return_most_similar(item['character'], v_characters) |
| 455 | item['position'] = "Position " + str(get_number(item['position'])) |
| 456 | # verify |
| 457 | |
| 458 | new_scene['scene'].append(new_line) |
| 459 | |
| 460 | initial_position_key = return_most_similar('initial position', list(scene.keys())) |
| 461 | new_scene['initial position'] = scene[initial_position_key] |
| 462 | for ini in new_scene['initial position']: |
| 463 | ini['character'] = return_most_similar(ini['character'], v_characters) |
| 464 | ini['position'] = "Position " + str(get_number(ini['position'])) |
| 465 | data.append(new_scene) |
| 466 | |
| 467 | write_json(self.script_path, data) |
| 468 | |
| 469 | |
| 470 | |
| 471 | if __name__ == '__main__': |
| 472 | f = FilmCrafter(topic = topics[ID-1]) |
| 473 | print("Characters selecting >>>") |
| 474 | f.casting() |
| 475 | print("Scenes planning >>>") |
| 476 | f.scenes_plan() |
| 477 | print("Lines generating >>>") |
| 478 | f.lines_generate() |
| 479 | print("Positions marking >>>") |
| 480 | f.position_mark() |
| 481 | print("Actions marking >>>") |
| 482 | f.action_mark() |
| 483 | print("Movement marking >>>") |
| 484 | f.move_mark() |
| 485 | print("Shots marking >>>") |
| 486 | f.shot_mark() |
| 487 | print("Script cleaning >>>") |
| 488 | f.clean_script() |
| 489 |