Upload 5 files

jobs/job_helper.py

@@ -0,0 +1,20 @@
+import os
+from datetime import datetime
+from typing import Optional
+
+from facefusion.filesystem import get_file_extension, get_file_name
+
+
+def get_step_output_path(job_id : str, step_index : int, output_path : str) -> Optional[str]:
+	if output_path:
+		output_directory_path, output_file_path = os.path.split(output_path)
+		output_file_name = get_file_name(output_file_path)
+		output_file_extension = get_file_extension(output_file_path)
+
+		if output_file_name and output_file_extension:
+			return os.path.join(output_directory_path, output_file_name + '-' + job_id + '-' + str(step_index) + output_file_extension)
+	return None
+
+
+def suggest_job_id(job_prefix : str = 'job') -> str:
+	return job_prefix + '-' + datetime.now().strftime('%Y-%m-%d-%H-%M-%S')

jobs/job_list.py

@@ -0,0 +1,34 @@
+from datetime import datetime
+from typing import List, Optional, Tuple
+
+from facefusion.jobs import job_manager
+from facefusion.time_helper import describe_time_ago
+from facefusion.types import JobStatus, TableContent, TableHeader
+
+
+def compose_job_list(job_status : JobStatus) -> Tuple[List[TableHeader], List[List[TableContent]]]:
+	jobs = job_manager.find_jobs(job_status)
+	job_headers : List[TableHeader] = [ 'job id', 'steps', 'date created', 'date updated', 'job status' ]
+	job_contents : List[List[TableContent]] = []
+
+	for index, job_id in enumerate(jobs):
+		if job_manager.validate_job(job_id):
+			job = jobs[job_id]
+			step_total = job_manager.count_step_total(job_id)
+			date_created = prepare_describe_datetime(job.get('date_created'))
+			date_updated = prepare_describe_datetime(job.get('date_updated'))
+			job_contents.append(
+			[
+				job_id,
+				step_total,
+				date_created,
+				date_updated,
+				job_status
+			])
+	return job_headers, job_contents
+
+
+def prepare_describe_datetime(date_time : Optional[str]) -> Optional[str]:
+	if date_time:
+		return describe_time_ago(datetime.fromisoformat(date_time))
+	return None

jobs/job_manager.py

@@ -0,0 +1,267 @@
+import os
+from copy import copy
+from typing import List, Optional
+
+import facefusion.choices
+from facefusion.filesystem import create_directory, get_file_name, is_directory, is_file, move_file, remove_directory, remove_file, resolve_file_pattern
+from facefusion.jobs.job_helper import get_step_output_path
+from facefusion.json import read_json, write_json
+from facefusion.sanitizer import sanitize_job_id
+from facefusion.time_helper import get_current_date_time
+from facefusion.types import Args, Job, JobSet, JobStatus, JobStep, JobStepStatus
+
+JOBS_PATH : Optional[str] = None
+
+
+def init_jobs(jobs_path : str) -> bool:
+	global JOBS_PATH
+
+	JOBS_PATH = jobs_path
+	job_status_paths = [ os.path.join(JOBS_PATH, job_status) for job_status in facefusion.choices.job_statuses ]
+
+	for job_status_path in job_status_paths:
+		create_directory(job_status_path)
+	return all(is_directory(status_path) for status_path in job_status_paths)
+
+
+def clear_jobs(jobs_path : str) -> bool:
+	return remove_directory(jobs_path)
+
+
+def create_job(job_id : str) -> bool:
+	job : Job =\
+	{
+		'version': '1',
+		'date_created': get_current_date_time().isoformat(),
+		'date_updated': None,
+		'steps': []
+	}
+
+	return create_job_file(job_id, job)
+
+
+def submit_job(job_id : str) -> bool:
+	drafted_job_ids = find_job_ids('drafted')
+	steps = get_steps(job_id)
+
+	if job_id in drafted_job_ids and steps:
+		return set_steps_status(job_id, 'queued') and move_job_file(job_id, 'queued')
+	return False
+
+
+def submit_jobs(halt_on_error : bool) -> bool:
+	drafted_job_ids = find_job_ids('drafted')
+	has_error = False
+
+	if drafted_job_ids:
+		for job_id in drafted_job_ids:
+			if not submit_job(job_id):
+				has_error = True
+				if halt_on_error:
+					return False
+		return not has_error
+	return False
+
+
+def delete_job(job_id : str) -> bool:
+	return delete_job_file(job_id)
+
+
+def delete_jobs(halt_on_error : bool) -> bool:
+	job_ids = find_job_ids('drafted') + find_job_ids('queued') + find_job_ids('failed') + find_job_ids('completed')
+	has_error = False
+
+	if job_ids:
+		for job_id in job_ids:
+			if not delete_job(job_id):
+				has_error = True
+				if halt_on_error:
+					return False
+		return not has_error
+	return False
+
+
+def find_jobs(job_status : JobStatus) -> JobSet:
+	job_ids = find_job_ids(job_status)
+	job_set : JobSet = {}
+
+	for job_id in job_ids:
+		job_set[job_id] = read_job_file(job_id)
+	return job_set
+
+
+def find_job_ids(job_status : JobStatus) -> List[str]:
+	job_pattern = os.path.join(JOBS_PATH, job_status, '*.json')
+	job_paths = resolve_file_pattern(job_pattern)
+	job_paths.sort(key = os.path.getmtime)
+	job_ids = []
+
+	for job_path in job_paths:
+		job_id = get_file_name(job_path)
+		job_ids.append(job_id)
+	return job_ids
+
+
+def validate_job(job_id : str) -> bool:
+	job = read_job_file(job_id)
+	return bool(job and 'version' in job and 'date_created' in job and 'date_updated' in job and 'steps' in job)
+
+
+def has_step(job_id : str, step_index : int) -> bool:
+	step_total = count_step_total(job_id)
+	return step_index in range(step_total)
+
+
+def add_step(job_id : str, step_args : Args) -> bool:
+	job = read_job_file(job_id)
+
+	if job:
+		job.get('steps').append(
+		{
+			'args': step_args,
+			'status': 'drafted'
+		})
+		return update_job_file(job_id, job)
+	return False
+
+
+def remix_step(job_id : str, step_index : int, step_args : Args) -> bool:
+	steps = get_steps(job_id)
+	step_args = copy(step_args)
+
+	if step_index and step_index < 0:
+		step_index = count_step_total(job_id) - 1
+
+	if has_step(job_id, step_index):
+		output_path = steps[step_index].get('args').get('output_path')
+		step_args['target_path'] = get_step_output_path(job_id, step_index, output_path)
+		return add_step(job_id, step_args)
+	return False
+
+
+def insert_step(job_id : str, step_index : int, step_args : Args) -> bool:
+	job = read_job_file(job_id)
+	step_args = copy(step_args)
+
+	if step_index and step_index < 0:
+		step_index = count_step_total(job_id) - 1
+
+	if job and has_step(job_id, step_index):
+		job.get('steps').insert(step_index,
+		{
+			'args': step_args,
+			'status': 'drafted'
+		})
+		return update_job_file(job_id, job)
+	return False
+
+
+def remove_step(job_id : str, step_index : int) -> bool:
+	job = read_job_file(job_id)
+
+	if step_index and step_index < 0:
+		step_index = count_step_total(job_id) - 1
+
+	if job and has_step(job_id, step_index):
+		job.get('steps').pop(step_index)
+		return update_job_file(job_id, job)
+	return False
+
+
+def get_steps(job_id : str) -> List[JobStep]:
+	job = read_job_file(job_id)
+
+	if job:
+		return job.get('steps')
+	return []
+
+
+def count_step_total(job_id : str) -> int:
+	steps = get_steps(job_id)
+
+	if steps:
+		return len(steps)
+	return 0
+
+
+def set_step_status(job_id : str, step_index : int, step_status : JobStepStatus) -> bool:
+	job = read_job_file(job_id)
+
+	if job:
+		steps = job.get('steps')
+		if has_step(job_id, step_index):
+			steps[step_index]['status'] = step_status
+			return update_job_file(job_id, job)
+	return False
+
+
+def set_steps_status(job_id : str, step_status : JobStepStatus) -> bool:
+	job = read_job_file(job_id)
+
+	if job:
+		for step in job.get('steps'):
+			step['status'] = step_status
+		return update_job_file(job_id, job)
+	return False
+
+
+def read_job_file(job_id : str) -> Optional[Job]:
+	job_path = find_job_path(job_id)
+	return read_json(job_path) #type:ignore[return-value]
+
+
+def create_job_file(job_id : str, job : Job) -> bool:
+	job_path = find_job_path(job_id)
+
+	if not is_file(job_path):
+		job_create_path = suggest_job_path(job_id, 'drafted')
+		return write_json(job_create_path, job) #type:ignore[arg-type]
+	return False
+
+
+def update_job_file(job_id : str, job : Job) -> bool:
+	job_path = find_job_path(job_id)
+
+	if is_file(job_path):
+		job['date_updated'] = get_current_date_time().isoformat()
+		return write_json(job_path, job) #type:ignore[arg-type]
+	return False
+
+
+def move_job_file(job_id : str, job_status : JobStatus) -> bool:
+	job_path = find_job_path(job_id)
+	job_move_path = suggest_job_path(job_id, job_status)
+	return move_file(job_path, job_move_path)
+
+
+def delete_job_file(job_id : str) -> bool:
+	job_path = find_job_path(job_id)
+	return remove_file(job_path)
+
+
+def suggest_job_path(job_id : str, job_status : JobStatus) -> Optional[str]:
+	job_file_name = get_job_file_name(job_id)
+
+	if job_file_name:
+		return os.path.join(JOBS_PATH, job_status, job_file_name)
+	return None
+
+
+def find_job_path(job_id : str) -> Optional[str]:
+	job_file_name = get_job_file_name(job_id)
+
+	if job_file_name:
+		for job_status in facefusion.choices.job_statuses:
+			job_pattern = os.path.join(JOBS_PATH, job_status, job_file_name)
+			job_paths = resolve_file_pattern(job_pattern)
+
+			for job_path in job_paths:
+				return job_path
+	return None
+
+
+def get_job_file_name(job_id : str) -> Optional[str]:
+	if job_id:
+		job_id = sanitize_job_id(job_id)
+		return job_id + '.json'
+	return None

jobs/job_runner.py

@@ -0,0 +1,112 @@
+from facefusion.ffmpeg import concat_video
+from facefusion.filesystem import are_images, are_videos, move_file, remove_file
+from facefusion.jobs import job_helper, job_manager
+from facefusion.types import JobOutputSet, JobStep, ProcessStep
+
+
+def run_job(job_id : str, process_step : ProcessStep) -> bool:
+	queued_job_ids = job_manager.find_job_ids('queued')
+
+	if job_id in queued_job_ids:
+		if run_steps(job_id, process_step) and finalize_steps(job_id):
+			clean_steps(job_id)
+			return job_manager.move_job_file(job_id, 'completed')
+		clean_steps(job_id)
+		job_manager.move_job_file(job_id, 'failed')
+	return False
+
+
+def run_jobs(process_step : ProcessStep, halt_on_error : bool) -> bool:
+	queued_job_ids = job_manager.find_job_ids('queued')
+	has_error = False
+
+	if queued_job_ids:
+		for job_id in queued_job_ids:
+			if not run_job(job_id, process_step):
+				has_error = True
+				if halt_on_error:
+					return False
+		return not has_error
+	return False
+
+
+def retry_job(job_id : str, process_step : ProcessStep) -> bool:
+	failed_job_ids = job_manager.find_job_ids('failed')
+
+	if job_id in failed_job_ids:
+		return job_manager.set_steps_status(job_id, 'queued') and job_manager.move_job_file(job_id, 'queued') and run_job(job_id, process_step)
+	return False
+
+
+def retry_jobs(process_step : ProcessStep, halt_on_error : bool) -> bool:
+	failed_job_ids = job_manager.find_job_ids('failed')
+	has_error = False
+
+	if failed_job_ids:
+		for job_id in failed_job_ids:
+			if not retry_job(job_id, process_step):
+				has_error = True
+				if halt_on_error:
+					return False
+		return not has_error
+	return False
+
+
+def run_step(job_id : str, step_index : int, step : JobStep, process_step : ProcessStep) -> bool:
+	step_args = step.get('args')
+
+	if job_manager.set_step_status(job_id, step_index, 'started') and process_step(job_id, step_index, step_args):
+		output_path = step_args.get('output_path')
+		step_output_path = job_helper.get_step_output_path(job_id, step_index, output_path)
+
+		return move_file(output_path, step_output_path) and job_manager.set_step_status(job_id, step_index, 'completed')
+	job_manager.set_step_status(job_id, step_index, 'failed')
+	return False
+
+
+def run_steps(job_id : str, process_step : ProcessStep) -> bool:
+	steps = job_manager.get_steps(job_id)
+
+	if steps:
+		for index, step in enumerate(steps):
+			if not run_step(job_id, index, step, process_step):
+				return False
+		return True
+	return False
+
+
+def finalize_steps(job_id : str) -> bool:
+	output_set = collect_output_set(job_id)
+
+	for output_path, temp_output_paths in output_set.items():
+		if are_videos(temp_output_paths):
+			if not concat_video(output_path, temp_output_paths):
+				return False
+		if are_images(temp_output_paths):
+			for temp_output_path in temp_output_paths:
+				if not move_file(temp_output_path, output_path):
+					return False
+	return True
+
+
+def clean_steps(job_id: str) -> bool:
+	output_set = collect_output_set(job_id)
+
+	for temp_output_paths in output_set.values():
+		for temp_output_path in temp_output_paths:
+			if not remove_file(temp_output_path):
+				return False
+	return True
+
+
+def collect_output_set(job_id : str) -> JobOutputSet:
+	steps = job_manager.get_steps(job_id)
+	job_output_set : JobOutputSet = {}
+
+	for index, step in enumerate(steps):
+		output_path = step.get('args').get('output_path')
+
+		if output_path:
+			step_output_path = job_manager.get_step_output_path(job_id, index, output_path)
+			job_output_set.setdefault(output_path, []).append(step_output_path)
+	return job_output_set

jobs/job_store.py

@@ -0,0 +1,27 @@
+from typing import List
+
+from facefusion.types import JobStore
+
+JOB_STORE : JobStore =\
+{
+	'job_keys': [],
+	'step_keys': []
+}
+
+
+def get_job_keys() -> List[str]:
+	return JOB_STORE.get('job_keys')
+
+
+def get_step_keys() -> List[str]:
+	return JOB_STORE.get('step_keys')
+
+
+def register_job_keys(job_keys : List[str]) -> None:
+	for job_key in job_keys:
+		JOB_STORE['job_keys'].append(job_key)
+
+
+def register_step_keys(step_keys : List[str]) -> None:
+	for step_key in step_keys:
+		JOB_STORE['step_keys'].append(step_key)