| 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 | import argparse |
| 7 | |
| 8 | |
| 9 | # TO DO |
| 10 | ROOT_PATH = "/path/to/FilmAgent" |
| 11 | # TO DO |
| 12 | |
| 13 | class FilmCrafter: |
| 14 | |
| 15 | def __init__(self, topic: str) -> None: |
| 16 | self.topic = topic |
| 17 | self.log_path = cretae_new_path(os.path.join(ROOT_PATH, "Logs"), "txt") |
| 18 | self.profile_path = os.path.join(ROOT_PATH, "Script\\actors_profile.json") |
| 19 | self.action_description_path = os.path.join(ROOT_PATH, "Locations\\actions.txt") |
| 20 | self.shot_description_path = os.path.join(ROOT_PATH, "Locations\\shots.txt") |
| 21 | # scenes |
| 22 | self.scene_path = os.path.join(ROOT_PATH, "Script\scenes_1.json") |
| 23 | # + lines |
| 24 | self.scene_path_1 = os.path.join(ROOT_PATH, "Script\scenes_2.json") |
| 25 | # + positions |
| 26 | self.scene_path_2 = os.path.join(ROOT_PATH, "Script\scenes_3.json") |
| 27 | # + actions |
| 28 | self.scene_path_3 = os.path.join(ROOT_PATH, "Script\scenes_4.json") |
| 29 | # stage1_verify |
| 30 | self.scene_path_4 = os.path.join(ROOT_PATH, "Script\scenes_5.json") |
| 31 | # stage2_verify |
| 32 | self.scene_path_5 = os.path.join(ROOT_PATH, "Script\scenes_6.json") |
| 33 | # + movement |
| 34 | self.scene_path_6 = os.path.join(ROOT_PATH, "Script\scenes_7.json") |
| 35 | # + shot (stage3_verify) |
| 36 | self.scene_path_7 = os.path.join(ROOT_PATH, "Script\scenes_8.json") |
| 37 | # The final script |
| 38 | self.script_path = cretae_new_path(os.path.join(ROOT_PATH, "Script\script"), "json") |
| 39 | |
| 40 | # director's shot annotation |
| 41 | self.director_shot_path = os.path.join(ROOT_PATH, "Script\director_shot.json") |
| 42 | # cinematographer's shot annotation |
| 43 | self.cinematographer_shot_path = os.path.join(ROOT_PATH, "Script\cinematographer_shot.json") |
| 44 | |
| 45 | # The maximum number of characters in a film |
| 46 | self.character_limit = 4 |
| 47 | # The maximum number of scenes in a film |
| 48 | self.scene_limit = 3 |
| 49 | # The maximum number of discussions between director and screenwriter |
| 50 | self.stage1_verify_limit = 3 |
| 51 | # The maximum number of discussions between director, actor and screenwriter |
| 52 | self.stage2_verify_limit = 3 |
| 53 | # The maximum number of discussions between director and cinematographer |
| 54 | self.stage3_verify_limit = 4 |
| 55 | |
| 56 | |
| 57 | def call(self, identity: str, params: Dict, trans2json: bool = True) -> Union[str, dict, list]: |
| 58 | prompt = read_prompt(os.path.join(ROOT_PATH, f"Prompt\{identity}.txt") ) |
| 59 | prompt = prompt_format(prompt, params) |
| 60 | log_prompt(self.log_path, prompt) |
| 61 | result = LLMCall(prompt, model) |
| 62 | if trans2json: |
| 63 | result = clean_text(result) |
| 64 | result = GPTResponse2JSON(result) |
| 65 | log_prompt(self.log_path, result) |
| 66 | return result |
| 67 | |
| 68 | |
| 69 | def casting(self): |
| 70 | ''' |
| 71 | Role: Director |
| 72 | |
| 73 | Behavior: Create the main characters and their bios for the film script. |
| 74 | ''' |
| 75 | params = {"{topic}": self.topic, "{character_limit}": self.character_limit} |
| 76 | result = self.call("director_1", params) |
| 77 | write_json(self.profile_path, result) |
| 78 | |
| 79 | |
| 80 | def scenes_plan(self): |
| 81 | ''' |
| 82 | Role: Director |
| 83 | |
| 84 | Behavior: |
| 85 | Plan the outline of script, include: |
| 86 | 1. The number of scenes. |
| 87 | 2. The characters, location and main plot of each scene. |
| 88 | ''' |
| 89 | profile = read_json(self.profile_path) |
| 90 | male_characters = ", ".join(list(map(lambda x: x['name'], |
| 91 | filter(lambda x: x['gender'].lower() == 'male', profile)))) |
| 92 | female_characters = ", ".join(list(map(lambda x: x['name'], |
| 93 | filter(lambda x: x['gender'].lower() == 'female', profile)))) |
| 94 | |
| 95 | params = {"{topic}": self.topic, |
| 96 | "{male_characters}": male_characters, |
| 97 | "{female_characters}": female_characters, |
| 98 | "{scene_limit}": self.scene_limit} |
| 99 | result = self.call("director_2", params) |
| 100 | write_json(self.scene_path, result) |
| 101 | |
| 102 | |
| 103 | def lines_generate(self): |
| 104 | ''' |
| 105 | Role: Screenwriter |
| 106 | |
| 107 | Behavior: Write lines for the script. |
| 108 | ''' |
| 109 | scenes = read_json(self.scene_path) |
| 110 | script_outline = "" |
| 111 | who = [] |
| 112 | where = [] |
| 113 | what = [] |
| 114 | for id,scene in enumerate(scenes): |
| 115 | selected_roles = scene[return_most_similar("selected-characters", list(scene.keys()))] |
| 116 | selected_location = scene[return_most_similar("selected-location", list(scene.keys()))] |
| 117 | story_plot = scene[return_most_similar("story-plot", list(scene.keys()))] |
| 118 | who.append(selected_roles) |
| 119 | where.append(selected_location) |
| 120 | what.append(story_plot) |
| 121 | |
| 122 | topic = scene[return_most_similar("sub-topic", list(scene.keys()))] |
| 123 | characters = ", ".join(selected_roles) |
| 124 | plot = story_plot |
| 125 | location = selected_location |
| 126 | goal = scene[return_most_similar("dialogue-goal", list(scene.keys()))] |
| 127 | |
| 128 | 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" |
| 129 | |
| 130 | params = {"{script_outline}": script_outline.strip()} |
| 131 | result = self.call("screenwriter_1", params) |
| 132 | |
| 133 | lines = [] |
| 134 | assert len(result) == len(scenes) |
| 135 | for j in range(len(scenes)): |
| 136 | line = {} |
| 137 | line['scene_information'] = {} |
| 138 | line['scene_information']['who'] = who[j] |
| 139 | line['scene_information']['where'] = where[j] |
| 140 | line['scene_information']['what'] = what[j] |
| 141 | line['dialogues'] = result[j][return_most_similar("scene-dialogue", list(result[j].keys()))] |
| 142 | lines.append(line) |
| 143 | write_json(self.scene_path_1, lines) |
| 144 | |
| 145 | |
| 146 | |
| 147 | def position_mark(self): |
| 148 | ''' |
| 149 | Role: Screenwriter |
| 150 | |
| 151 | Behavior: Choose an appropriate initial position for each character in each scene of the script. |
| 152 | ''' |
| 153 | scenes = read_json(self.scene_path_1) |
| 154 | script_information = "" |
| 155 | optional_positions = "" |
| 156 | for id,scene in enumerate(scenes): |
| 157 | i = id + 1 |
| 158 | who = scene['scene_information']['who'] |
| 159 | where = scene['scene_information']['where'] |
| 160 | what = scene['scene_information']['what'] |
| 161 | |
| 162 | script_information = script_information + f"{i}. **Scene {i}**:\n - characters: {who}\n - location: {where}\n - plot: {what}\n\n" |
| 163 | |
| 164 | position_path = os.path.join(ROOT_PATH, f"Locations\{where}\position.json") |
| 165 | positions = read_json(position_path) |
| 166 | normal_position = [item for item in positions if item['fixed_angle'] == False] |
| 167 | # This "if judgment" is related to the position, and camera settings in Unity. |
| 168 | if len(who) >= len(positions) - len(normal_position) + 2: |
| 169 | p = "" |
| 170 | for it,position in enumerate(positions): |
| 171 | j = it + 1 |
| 172 | p = p + f" - Position {j}: " + position['description'] + '\n' |
| 173 | else: |
| 174 | p = "" |
| 175 | for it,position in enumerate(normal_position): |
| 176 | j = it + 1 |
| 177 | p = p + f" - Position {j}: " + position['description'] + '\n' |
| 178 | optional_positions = optional_positions + f"{i}. **Positions in {where}**:\n{p}\n" |
| 179 | |
| 180 | params = {"{script_information}": script_information.strip(), |
| 181 | "{optional_positions}": optional_positions.strip()} |
| 182 | result = self.call("screenwriter_2", params) |
| 183 | |
| 184 | assert len(result) == len(scenes) |
| 185 | for j in range(len(scenes)): |
| 186 | scenes[j]["initial position"] = result[j][return_most_similar("scene-position", list(result[j].keys()))] |
| 187 | write_json(self.scene_path_2, scenes) |
| 188 | |
| 189 | |
| 190 | |
| 191 | def action_mark(self): |
| 192 | ''' |
| 193 | Role: Screenwriter |
| 194 | |
| 195 | Behavior: Choose appropriate actions for the characters engaged in the dialogue. |
| 196 | ''' |
| 197 | scenes = read_json(self.scene_path_2) |
| 198 | all_actions = read_prompt(self.action_description_path) |
| 199 | data = [] |
| 200 | for scene in scenes: |
| 201 | position_path = os.path.join(ROOT_PATH, f"Locations\{scene['scene_information']['where']}\position.json") |
| 202 | positions = read_json(position_path) |
| 203 | |
| 204 | ini = "" |
| 205 | for id,item in enumerate(scene['initial position']): |
| 206 | if [it['sittable'] for it in positions if get_number(it['id']) == get_number(item['position'])][0]: |
| 207 | sit = "sittable" |
| 208 | else: |
| 209 | sit = "unsittable" |
| 210 | ini = ini + f" - {item['character']}: " + f"{sit} Position {str(get_number(item['position']))}, standing\n" |
| 211 | ini = " " + ini.strip() |
| 212 | params = {"{initial}": ini, |
| 213 | "{plot}": scene['scene_information']['what'], |
| 214 | "{dialogues}": scene['dialogues'], |
| 215 | "{all_actions}": all_actions} |
| 216 | result = self.call("screenwriter_3", params) |
| 217 | |
| 218 | assert len(result) == len(scene['dialogues']) |
| 219 | scene['dialogues'] = result |
| 220 | data.append(scene) |
| 221 | |
| 222 | write_json(self.scene_path_3, data) |
| 223 | |
| 224 | |
| 225 | |
| 226 | def find_unknown_actions(self, scenes: List) -> List: |
| 227 | ''' |
| 228 | Input: The complete script |
| 229 | |
| 230 | Output: The list of actions appearing in the script that are not set in Unity. |
| 231 | ''' |
| 232 | unknown_actions = [] |
| 233 | all_actions = read_json(os.path.join(ROOT_PATH, "Locations\\actions.json")) |
| 234 | for scene in scenes: |
| 235 | for line in scene['dialogues']: |
| 236 | for action in line['actions']: |
| 237 | if action['action'] not in all_actions.keys() and action['action'] not in unknown_actions: |
| 238 | unknown_actions.append(action['action']) |
| 239 | |
| 240 | return unknown_actions |
| 241 | |
| 242 | |
| 243 | |
| 244 | def stage1_verify(self): |
| 245 | ''' |
| 246 | Role: Director and Screenwriter |
| 247 | |
| 248 | Behavior: The director and screenwriter have a comprehensive discussion about the script to improve it, focusing on the following three aspects: Action Reasonableness, Theme Consistency, Script Fluency. |
| 249 | ''' |
| 250 | scenes = read_json(self.scene_path_3) |
| 251 | current_script = [] |
| 252 | unknown_actions = self.find_unknown_actions(scenes) |
| 253 | characters_position = "" |
| 254 | for id,scene in enumerate(scenes): |
| 255 | new_scene = {} |
| 256 | new_scene['scene_information'] = scene['scene_information'] |
| 257 | new_scene['initial position'] = scene['initial position'] |
| 258 | new_scene['dialogues'] = [] |
| 259 | for line in scene['dialogues']: |
| 260 | new_line = copy.deepcopy(line) |
| 261 | for action in new_line['actions']: |
| 262 | action.pop("reasoning") |
| 263 | new_scene['dialogues'].append(new_line) |
| 264 | current_script.append(new_scene) |
| 265 | |
| 266 | position_path = os.path.join(ROOT_PATH, f"Locations\{scene['scene_information']['where']}\position.json") |
| 267 | positions = read_json(position_path) |
| 268 | p = [] |
| 269 | for position in scene['initial position']: |
| 270 | position_id = get_number(position['position']) |
| 271 | sittable = "sittable" if positions[position_id-1]['sittable'] else "unsittable" |
| 272 | p.append(f"{position['character']}'s position: {sittable}") |
| 273 | characters_position = characters_position + f"{id+1}. **Scene {id+1}**:\n{', '.join(p)}\n\n" |
| 274 | |
| 275 | all_actions = read_prompt(self.action_description_path) |
| 276 | for i in range(self.stage1_verify_limit): |
| 277 | params = {"{theme}": self.topic, |
| 278 | "{original_script}": current_script, |
| 279 | "{all_actions}": all_actions, |
| 280 | "{unknown_actions}": ', '.join(unknown_actions)} |
| 281 | feedback = self.call("director_3", params, trans2json=False) |
| 282 | |
| 283 | params = {"{theme}": self.topic, |
| 284 | "{feedback}": feedback, |
| 285 | "{script}": current_script, |
| 286 | "{all_actions}": all_actions, |
| 287 | "{characters_position}": characters_position} |
| 288 | revised_script = self.call("screenwriter_4", params) |
| 289 | current_script = revised_script |
| 290 | unknown_actions = self.find_unknown_actions(revised_script) |
| 291 | # If there are still unknown actions, discuss again immediately. |
| 292 | if len(unknown_actions)>0: |
| 293 | continue |
| 294 | |
| 295 | params = {"{feedback}": feedback, |
| 296 | "{revised_script}": revised_script} |
| 297 | verify = self.call("director_4", params, trans2json=False) |
| 298 | if verify.find("finalize") == -1: |
| 299 | continue |
| 300 | verify = verify[verify.find("finalize"):] |
| 301 | if "True" in verify or "true" in verify: |
| 302 | break |
| 303 | |
| 304 | write_json(self.scene_path_4, revised_script) |
| 305 | |
| 306 | |
| 307 | |
| 308 | def stage2_verify(self): |
| 309 | ''' |
| 310 | Role: Actor, Director and Screenwriter |
| 311 | |
| 312 | Behavior: The actors suggest minor adjustments to the script lines, and the director and screenwriter discuss the revisions. |
| 313 | ''' |
| 314 | scenes = read_json(self.scene_path_4) |
| 315 | profiles = read_json(self.profile_path) |
| 316 | feedback = {} |
| 317 | scenes_for_actor = [] |
| 318 | for scene in scenes: |
| 319 | item = {} |
| 320 | item['scene_information'] = scene[return_most_similar('scene_information', list(scene.keys()))] |
| 321 | item["dialogues"] = [] |
| 322 | for line in scene['dialogues']: |
| 323 | new_line = {} |
| 324 | new_line['speaker'] = line['speaker'] |
| 325 | new_line['content'] = line['content'] |
| 326 | item["dialogues"].append(new_line) |
| 327 | scenes_for_actor.append(item) |
| 328 | |
| 329 | for profile in profiles: |
| 330 | params = {"{character}": profile['name'], |
| 331 | "{profile}": profile, |
| 332 | "{script}": scenes_for_actor} |
| 333 | result = self.call("actor", params, trans2json=False) |
| 334 | feedback[profile['name']] = result |
| 335 | |
| 336 | suggestions = "" |
| 337 | for name, suggestion in feedback.items(): |
| 338 | suggestions = suggestions + f" - **{name}**: {suggestion}\n" |
| 339 | params = {"{suggestions}": suggestions, |
| 340 | "{character_profiles}": profiles, |
| 341 | "{draft_script}": scenes} |
| 342 | result = self.call("director_5", params) |
| 343 | |
| 344 | if result[return_most_similar("adopted-suggestions", list(result.keys()))] == "None": |
| 345 | write_json(self.scene_path_5, scenes) |
| 346 | else: |
| 347 | current_script = scenes_for_actor |
| 348 | adopted_suggestions = result[return_most_similar("adopted-suggestions", list(result.keys()))] |
| 349 | for sugg in adopted_suggestions: |
| 350 | sugg.pop("reason") |
| 351 | for i in range(self.stage2_verify_limit): |
| 352 | params = {"{feedback}": adopted_suggestions, |
| 353 | "{script}": current_script} |
| 354 | revised_script = self.call("screenwriter_5", params) |
| 355 | current_script = revised_script |
| 356 | |
| 357 | params = {"{feedback}": adopted_suggestions, |
| 358 | "{revised_script}": revised_script} |
| 359 | verify = self.call("director_6", params, trans2json=False) |
| 360 | verify = verify[verify.find("finalize"):] |
| 361 | if "True" in verify or "true" in verify: |
| 362 | break |
| 363 | |
| 364 | assert len(revised_script) == len(scenes) |
| 365 | for id,scene in enumerate(scenes): |
| 366 | assert len(scene['dialogues']) == len(revised_script[id]['dialogues']) |
| 367 | for it,line in enumerate(scene['dialogues']): |
| 368 | assert line['speaker'] == revised_script[id]['dialogues'][it]['speaker'] |
| 369 | line['content'] = revised_script[id]['dialogues'][it]['content'] |
| 370 | |
| 371 | write_json(self.scene_path_5, scenes) |
| 372 | |
| 373 | |
| 374 | |
| 375 | def is_keep_standing(self, lines, character): |
| 376 | ''' |
| 377 | Description: Check if the character remains standing throughout the script. |
| 378 | |
| 379 | Input: The complete script, a character's name |
| 380 | |
| 381 | Output: True or False |
| 382 | ''' |
| 383 | for line in lines: |
| 384 | actions = line['actions'] |
| 385 | for action in actions: |
| 386 | if action['character'] == character and action['state'] == 'sitting': |
| 387 | return False |
| 388 | return True |
| 389 | |
| 390 | |
| 391 | def moveable_options(self, scene): |
| 392 | ''' |
| 393 | Input: The complete script |
| 394 | |
| 395 | Output: |
| 396 | 1. All movable characters (i.e., the characters that remain standing throughout the script) |
| 397 | 2. All positions that character can move to (i.e., the unoccupied positions) |
| 398 | ''' |
| 399 | position_path = os.path.join(ROOT_PATH, f"Locations\{scene[return_most_similar('scene_information', list(scene.keys()))]['where']}\position.json") |
| 400 | positions = read_json(position_path) |
| 401 | occupied_positions = [get_number(item['position']) for item in scene[return_most_similar('initial position', list(scene.keys()))]] |
| 402 | unoccupied_positions = [f"{item['id']}: {item['description']}" for item in positions if get_number(item['id']) not in occupied_positions] |
| 403 | if len(unoccupied_positions) == 0: |
| 404 | return None, None |
| 405 | |
| 406 | who = scene[return_most_similar('scene_information', list(scene.keys()))]['who'] |
| 407 | moveable_characters = [] |
| 408 | for character in who: |
| 409 | if self.is_keep_standing(scene['dialogues'], character): |
| 410 | moveable_characters.append(character) |
| 411 | if len(moveable_characters) == 0: |
| 412 | return None, None |
| 413 | |
| 414 | return moveable_characters, unoccupied_positions |
| 415 | |
| 416 | |
| 417 | def move_mark(self): |
| 418 | ''' |
| 419 | Role: Director |
| 420 | |
| 421 | Behavior: The director adds appropriate character movements into the script. |
| 422 | ''' |
| 423 | scenes = read_json(self.scene_path_5) |
| 424 | data = [] |
| 425 | for scene in scenes: |
| 426 | moveable_characters, unoccupied_positions = self.moveable_options(scene) |
| 427 | moved_charatcter, moveto_position = None, None |
| 428 | if moveable_characters: |
| 429 | move2destination = "" |
| 430 | for pn in unoccupied_positions: |
| 431 | move2destination = move2destination + f" - {pn}\n" |
| 432 | move2destination = " " + move2destination.strip() |
| 433 | lines = [] |
| 434 | for id in range(len(scene['dialogues'])): |
| 435 | lines.append(f"<Insertion Position {id}>") |
| 436 | item = {} |
| 437 | item['speaker'] = scene['dialogues'][id]['speaker'] |
| 438 | item['content'] = scene['dialogues'][id]['content'] |
| 439 | lines.append(item) |
| 440 | params = {"{moveable_characters}": moveable_characters, |
| 441 | "{story}": scene[return_most_similar('scene_information', list(scene.keys()))]['what'], |
| 442 | "{lines}": lines, |
| 443 | "{destinations}": move2destination, |
| 444 | "{current_positions}": scene[return_most_similar('initial position', list(scene.keys()))]} |
| 445 | result = self.call("director_7", params) |
| 446 | |
| 447 | if 'insertion' in result.keys(): |
| 448 | moved_charatcter = result['move']['character'] |
| 449 | moveto_position = result['move']['destination'] |
| 450 | scene['dialogues'].insert(get_number(result['insertion'][return_most_similar('insertion position', list(result['insertion'].keys()))]), result) |
| 451 | |
| 452 | # Update the characters' position in real time |
| 453 | position_change = False |
| 454 | for line in scene['dialogues']: |
| 455 | if "move" not in line.keys(): |
| 456 | line['current position'] = scene[return_most_similar('initial position', list(scene.keys()))] |
| 457 | else: |
| 458 | position_change = True |
| 459 | line['current position'] = scene[return_most_similar('initial position', list(scene.keys()))] |
| 460 | continue |
| 461 | if position_change: |
| 462 | line['current position'] = [item if item['character'] != moved_charatcter else {'character': moved_charatcter, 'position': moveto_position} for item in line['current position']] |
| 463 | |
| 464 | data.append(scene) |
| 465 | |
| 466 | write_json(self.scene_path_6, data) |
| 467 | |
| 468 | |
| 469 | def shot_mark(self): |
| 470 | ''' |
| 471 | Input: None |
| 472 | |
| 473 | Output: |
| 474 | 1. Director's shot annotations |
| 475 | 2. Cinematographer's shot annotations |
| 476 | 3. The complete script before adding shot annotations |
| 477 | 4. The script after inserting the shot annotation points. |
| 478 | ''' |
| 479 | scenes = read_json(self.scene_path_6) |
| 480 | script = {} |
| 481 | for ID,scene in enumerate(scenes): |
| 482 | I = ID + 1 |
| 483 | script[f'scene {I}'] = [] |
| 484 | for id,item in enumerate(scene['dialogues']): |
| 485 | line = {} |
| 486 | i = id + 1 |
| 487 | if "speaker" in item.keys(): |
| 488 | line['dialogue'] = f"{item['speaker']}: {item['content']}" |
| 489 | line['actions'] = item['actions'] |
| 490 | # line['plot'] = ' '.join([it['reason'] for it in item['actions']]) |
| 491 | line[f'selected-shot-{i}'] = "..." |
| 492 | else: |
| 493 | line['move'] = {} |
| 494 | line['move']['character'] = item['move']['character'] |
| 495 | line['move']['destination'] = item['move']['destination'] |
| 496 | line[f'selected-shot-{i}'] = "..." |
| 497 | script[f'scene {I}'].append(line) |
| 498 | |
| 499 | all_shots = read_prompt(self.shot_description_path) |
| 500 | params = {"{script}": script, "{all_shots}": all_shots} |
| 501 | result1 = self.call("cinematographer", params) |
| 502 | result2 = self.call("cinematographer", params) |
| 503 | # result2 = self.call("director_8", params) |
| 504 | for scene_id, scene in result1.items(): |
| 505 | for shot_id, shot in scene.items(): |
| 506 | shot.pop("reasoning") |
| 507 | |
| 508 | for scene_id, scene in result2.items(): |
| 509 | for shot_id, shot in scene.items(): |
| 510 | shot.pop("reasoning") |
| 511 | |
| 512 | write_json(self.director_shot_path, result2) |
| 513 | write_json(self.cinematographer_shot_path, result1) |
| 514 | |
| 515 | return script, result1, result2, scenes |
| 516 | |
| 517 | |
| 518 | def revise_shot_annotation(self, src, feedback): |
| 519 | ''' |
| 520 | Input: Original shot annotation, revision suggestion |
| 521 | |
| 522 | Output: Revised shot annotation |
| 523 | ''' |
| 524 | new = src |
| 525 | for scene_id, scene in new.items(): |
| 526 | for shot_id, shot in scene.items(): |
| 527 | if shot['shot'] != feedback[scene_id][shot_id]['shot']: |
| 528 | continue |
| 529 | |
| 530 | need_update = feedback[scene_id][shot_id][return_most_similar("need update", list(feedback[scene_id][shot_id].keys()))] |
| 531 | if isinstance(need_update, str) and need_update.lower() == "false": |
| 532 | continue |
| 533 | if isinstance(need_update, bool) and not need_update: |
| 534 | continue |
| 535 | shot['shot'] = feedback[scene_id][shot_id][return_most_similar("updated shot", list(feedback[scene_id][shot_id].keys()))] |
| 536 | |
| 537 | return new |
| 538 | |
| 539 | |
| 540 | |
| 541 | def stage3_verify(self): |
| 542 | ''' |
| 543 | Role: Director and Cinematographer |
| 544 | |
| 545 | Behavior: The director and the cinematographer provide their own shot annotations, then discuss and revise them together, and finally, the director selects the better one. |
| 546 | ''' |
| 547 | script, shot_mark_cinematographer, shot_mark_director, scenes = self.shot_mark() |
| 548 | current_shot_c, current_shot_d = shot_mark_cinematographer, shot_mark_director |
| 549 | all_shots = read_prompt(self.shot_description_path) |
| 550 | for i in range(self.stage3_verify_limit): |
| 551 | params = {"{identity_1}": "Director", |
| 552 | "{identity_2}": "Cinematographer", |
| 553 | "{script}": script, |
| 554 | "{shot_annotation}": current_shot_c, |
| 555 | "{all_shots}": all_shots |
| 556 | } |
| 557 | result1 = self.call("shot_review", params) |
| 558 | current_shot_c = self.revise_shot_annotation(current_shot_c, result1) |
| 559 | |
| 560 | params = {"{identity_1}": "Cinematographer", |
| 561 | "{identity_2}": "Director", |
| 562 | "{script}": script, |
| 563 | "{shot_annotation}": current_shot_d, |
| 564 | "{all_shots}": all_shots |
| 565 | } |
| 566 | result2 = self.call("shot_review", params) |
| 567 | current_shot_d = self.revise_shot_annotation(current_shot_d, result2) |
| 568 | |
| 569 | params = {"{shot_annotation_1}": current_shot_d, |
| 570 | "{shot_annotation_2}": current_shot_c, |
| 571 | "{script}": script, |
| 572 | "{all_shots}": all_shots |
| 573 | } |
| 574 | result = self.call("director_9", params) |
| 575 | last_shots = current_shot_d if result['better']=="1" else current_shot_c |
| 576 | # last_shots = current_shot_d if result['better'].lower()=="director" else current_shot_c |
| 577 | |
| 578 | assert len(list(last_shots.keys())) == len(scenes) |
| 579 | for id in range(len(list(last_shots.keys()))): |
| 580 | i = id + 1 |
| 581 | assert len(list(last_shots[f'scene {i}'].keys())) == len(scenes[id]['dialogues']) |
| 582 | for key,value in last_shots[f'scene {i}'].items(): |
| 583 | scenes[id]['dialogues'][get_number(key)-1]['selected shot'] = value[return_most_similar('shot', list(value.keys()))] |
| 584 | |
| 585 | |
| 586 | write_json(self.scene_path_7, scenes) |
| 587 | |
| 588 | |
| 589 | # Used for clean_script() |
| 590 | def process_action(self, actions, v_characters, v_actions): |
| 591 | new_actions = [] |
| 592 | for item in actions: |
| 593 | new_item = {} |
| 594 | new_item['character'] = return_most_similar(item['character'], v_characters) |
| 595 | new_item['state'] = return_most_similar(item['state'], ['standing', 'sitting']) |
| 596 | |
| 597 | new_item['action'] = return_most_similar(item['action'], v_actions) |
| 598 | if new_item['state'] == "standing" and ("Standing" in new_item['action'] or new_item['action'] == "Joyful Jump" or new_item['action'] == "Sit Down"): |
| 599 | if new_item['action'] == "Standing Talking": |
| 600 | new_item['action'] = "Standing Talking " + str(random.randint(1, 6)) |
| 601 | if new_item['action'] == "Standing Angry": |
| 602 | new_item['action'] = "Standing Angry " + str(random.randint(1, 4)) |
| 603 | if new_item['action'] == "Standing Arguing": |
| 604 | new_item['action'] = "Standing Arguing " + str(random.randint(1, 2)) |
| 605 | if new_item['action'] == "Standing Agree": |
| 606 | new_item['action'] = "Standing Agree " + str(random.randint(1, 2)) |
| 607 | new_actions.append(new_item) |
| 608 | elif new_item['state'] == "sitting" and ("Sitting" in new_item['action'] or new_item['action'] == "Stand Up"): |
| 609 | if new_item['action'] == "Sitting Talking": |
| 610 | new_item['action'] = "Sitting Talking " + str(random.randint(1, 2)) |
| 611 | new_actions.append(new_item) |
| 612 | else: |
| 613 | pass |
| 614 | return new_actions |
| 615 | |
| 616 | |
| 617 | # Used for clean_script() |
| 618 | def process_shot(self, info, location, shot, v_shots): |
| 619 | shot = return_most_similar(shot, v_shots) |
| 620 | if shot == "Pan Shot": |
| 621 | return "Pan Shot 1" |
| 622 | elif shot == "Track Shot": |
| 623 | return "Track Shot " + str(random.randint(1, int(info[location]['track']))) |
| 624 | elif shot == "Long Shot": |
| 625 | return "Long Shot " + str(random.randint(1, int(info[location]['long']))) |
| 626 | else: |
| 627 | return shot |
| 628 | |
| 629 | |
| 630 | def clean_script(self): |
| 631 | ''' |
| 632 | Description: Only keep the necessary information in the script and perform certain checks to avoid errors when executing the script in Unity. |
| 633 | ''' |
| 634 | scenes = read_json(self.scene_path_7) |
| 635 | profiles = read_json(self.profile_path) |
| 636 | v_characters = [item['name'] for item in profiles] |
| 637 | info = read_json(os.path.join(ROOT_PATH, "Locations\\rotateandtrack.json")) |
| 638 | v_locations = [location for location in info.keys()] |
| 639 | info_1 = read_json(os.path.join(ROOT_PATH, "Locations\\actions.json")) |
| 640 | v_actions = [action for action in info_1.keys()] |
| 641 | info_2 = read_json(os.path.join(ROOT_PATH, "Locations\\shots.json")) |
| 642 | v_shots = [shot for shot in info_2.keys()] |
| 643 | |
| 644 | data = [] |
| 645 | for scene in scenes: |
| 646 | new_scene = {} |
| 647 | scene_information_key = return_most_similar('scene_information', list(scene.keys())) |
| 648 | new_scene['scene information'] = scene[scene_information_key] |
| 649 | |
| 650 | # verify |
| 651 | for role in new_scene['scene information']['who']: |
| 652 | role = return_most_similar(role, v_characters) |
| 653 | new_scene['scene information']['where'] = return_most_similar(new_scene['scene information']['where'], v_locations) |
| 654 | # verify |
| 655 | |
| 656 | new_scene['scene'] = [] |
| 657 | for line in scene['dialogues']: |
| 658 | new_line = {} |
| 659 | selected_shot_key = return_most_similar('selected shot', list(line.keys())) |
| 660 | current_position_key = return_most_similar('current position', list(line.keys())) |
| 661 | # verify |
| 662 | if 'speaker' in line.keys(): |
| 663 | new_line['speaker'] = return_most_similar(line['speaker'], v_characters) |
| 664 | new_line['content'] = line['content'] |
| 665 | new_line['actions'] = self.process_action(line['actions'], v_characters, v_actions) |
| 666 | new_line['shot'] = self.process_shot(info, scene[scene_information_key]['where'], line[selected_shot_key], v_shots) |
| 667 | new_line['current position'] = line[current_position_key] |
| 668 | else: |
| 669 | new_line['move'] = {} |
| 670 | new_line['move']['character'] = return_most_similar(line['move']['character'], v_characters) |
| 671 | new_line['move']['destination'] = "Position " + str(get_number(line['move']['destination'])) |
| 672 | new_line['shot'] = self.process_shot(info, scene[scene_information_key]['where'], line[selected_shot_key], v_shots) |
| 673 | new_line['current position'] = line[current_position_key] |
| 674 | # verify |
| 675 | |
| 676 | # verify |
| 677 | for item in new_line['current position']: |
| 678 | item['character'] = return_most_similar(item['character'], v_characters) |
| 679 | item['position'] = "Position " + str(get_number(item['position'])) |
| 680 | # verify |
| 681 | |
| 682 | new_scene['scene'].append(new_line) |
| 683 | |
| 684 | initial_position_key = return_most_similar('initial position', list(scene.keys())) |
| 685 | new_scene['initial position'] = scene[initial_position_key] |
| 686 | for ini in new_scene['initial position']: |
| 687 | ini['character'] = return_most_similar(ini['character'], v_characters) |
| 688 | ini['position'] = "Position " + str(get_number(ini['position'])) |
| 689 | data.append(new_scene) |
| 690 | |
| 691 | write_json(self.script_path, data) |
| 692 | |
| 693 | |
| 694 | |
| 695 | if __name__ == '__main__': |
| 696 | parser = argparse.ArgumentParser() |
| 697 | parser.add_argument('--model', help="LLM to be called", type=str) |
| 698 | args = parser.parse_args() |
| 699 | model = args.model |
| 700 | |
| 701 | f = FilmCrafter(topic = "Reconcilation in a friend reunion") |
| 702 | print("Characters selecting >>>") |
| 703 | f.casting() |
| 704 | print("Scenes planning >>>") |
| 705 | f.scenes_plan() |
| 706 | print("Lines generating >>>") |
| 707 | f.lines_generate() |
| 708 | print("Positions marking >>>") |
| 709 | f.position_mark() |
| 710 | print("Actions marking >>>") |
| 711 | f.action_mark() |
| 712 | print("Director discusses with screenwriter about the script >>>") |
| 713 | f.stage1_verify() |
| 714 | print("Actors give comments on the lines >>>") |
| 715 | f.stage2_verify() |
| 716 | print("Movement marking >>>") |
| 717 | f.move_mark() |
| 718 | print("Director discusses with cinematographer about the shots >>>") |
| 719 | f.stage3_verify() |
| 720 | print("Script cleaning >>>") |
| 721 | f.clean_script() |
| 722 |