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