mirror of
https://github.com/vladmandic/automatic
synced 2026-09-19 09:14:35 +02:00
process add video info and metadata
Signed-off-by: Vladimir Mandic <mandic00@live.com>
This commit is contained in:
Vladimir Mandic
+4
-3
@@ -1,6 +1,6 @@
|
||||
# Change Log for SD.Next
|
||||
|
||||
## Update for 2026-07-18
|
||||
## Update for 2026-07-19
|
||||
|
||||
- **Compute**
|
||||
- torch: update to `2.13.0` for CUDA, ROCm, IPEX
|
||||
@@ -11,8 +11,9 @@
|
||||
- sdnq attention optimizations
|
||||
- sdnq separate dit/te settings
|
||||
- **Features**
|
||||
- SeedVR enhanced support
|
||||
- Propagate server tracebacks to client
|
||||
- seedvr: enhanced upscaler support
|
||||
- process: read video properties metadata
|
||||
- logs: propagate server tracebacks to client
|
||||
- **Fixes**
|
||||
- upscaler auto-refresh to catch chainner upscalers that are not loaded on first attempt
|
||||
- lora support diffusers trainer
|
||||
|
||||
Submodule extensions-builtin/sdnext-modernui updated: 937554e88f...9868ac66d0
@@ -292,7 +292,7 @@ class APIProcess:
|
||||
reqDict, script_args = self.set_upscalers(req)
|
||||
reqDict['image'] = helpers.decode_base64_to_image(reqDict['image'])
|
||||
with self.queue_lock:
|
||||
result = postprocessing.run_extras(extras_mode=0, image_folder="", input_dir="", output_dir="", save_output=False, script_args=script_args, **reqDict)
|
||||
result = postprocessing.run_extras(extras_mode=0, image_folder="", input_dir="", output_dir="", video="", save_output=False, script_args=script_args, **reqDict)
|
||||
return models.ResProcessImage(image=helpers.encode_pil_to_base64(result[0][0]), html_info=result[1])
|
||||
|
||||
def extras_batch_images_api(self, req: models.ReqProcessBatch):
|
||||
@@ -301,5 +301,5 @@ class APIProcess:
|
||||
image_list = reqDict.pop('imageList', [])
|
||||
image_folder = [helpers.decode_base64_to_image(x.data) for x in image_list]
|
||||
with self.queue_lock:
|
||||
result = postprocessing.run_extras(extras_mode=1, image_folder=image_folder, image="", input_dir="", output_dir="", save_output=False, script_args=script_args, **reqDict)
|
||||
result = postprocessing.run_extras(extras_mode=1, image_folder=image_folder, image="", input_dir="", output_dir="", video="", save_output=False, script_args=script_args, **reqDict)
|
||||
return models.ResProcessBatch(images=list(map(helpers.encode_pil_to_base64, result[0])), html_info=result[1])
|
||||
|
||||
@@ -161,6 +161,8 @@ def blend(images):
|
||||
|
||||
|
||||
def decode_fourcc(cc):
|
||||
if cc is None:
|
||||
return None
|
||||
cc_bytes = int(cc).to_bytes(4, byteorder=sys.byteorder) # convert code to a bytearray
|
||||
cc_str = cc_bytes.decode() # decode byteaarray to a string
|
||||
return cc_str
|
||||
|
||||
@@ -9,12 +9,19 @@ from modules.shared import opts
|
||||
from modules.paths import resolve_output_path
|
||||
|
||||
|
||||
def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemporaryFile], input_dir, output_dir, show_extras_results, *args, save_output: bool = True):
|
||||
def run_postprocessing(extras_mode,
|
||||
image,
|
||||
image_folder: list[tempfile.NamedTemporaryFile],
|
||||
input_dir,
|
||||
output_dir,
|
||||
extras_video,
|
||||
show_extras_results,
|
||||
*args,
|
||||
save_output: bool = True):
|
||||
devices.torch_gc()
|
||||
shared.state.begin('Extras')
|
||||
image_data = []
|
||||
image_names = []
|
||||
image_fullnames = []
|
||||
image_ext = []
|
||||
outputs = []
|
||||
params = {}
|
||||
@@ -32,7 +39,6 @@ def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemp
|
||||
log.error(f'Failed to open image: file="{img.name}" {e}')
|
||||
continue
|
||||
fn, ext = os.path.splitext(img.orig_name)
|
||||
image_fullnames.append(img.name)
|
||||
image_data.append(image)
|
||||
image_names.append(fn)
|
||||
image_ext.append(ext)
|
||||
@@ -47,11 +53,12 @@ def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemp
|
||||
except Exception as e:
|
||||
log.error(f'Failed to open image: file="{fn}" {e}')
|
||||
continue
|
||||
image_fullnames.append(fn)
|
||||
image_data.append(image)
|
||||
image_names.append(fn)
|
||||
image_ext.append(None)
|
||||
log.debug(f'Process: mode=folder inputs={input_dir} files={len(image_list)} images={len(image_data)}')
|
||||
elif extras_mode == 3:
|
||||
log.error(f'Process: mode=video file="{extras_video}" not implemented yet')
|
||||
else:
|
||||
image_data.append(image)
|
||||
image_names.append(None)
|
||||
@@ -60,6 +67,7 @@ def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemp
|
||||
outpath = output_dir
|
||||
else:
|
||||
outpath = resolve_output_path(opts.outdir_samples, opts.outdir_extras_samples)
|
||||
|
||||
processed_images = []
|
||||
for image, name, ext in zip(image_data, image_names, image_ext, strict=False): # pylint: disable=redefined-argument-from-local
|
||||
log.debug(f'Process: image={image} {args}')
|
||||
@@ -67,6 +75,12 @@ def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemp
|
||||
if shared.state.interrupted:
|
||||
log.debug('Postprocess interrupted')
|
||||
break
|
||||
if isinstance(image, str):
|
||||
try:
|
||||
image = Image.open(image)
|
||||
except Exception as e:
|
||||
log.error(f'Failed to open image: file="{image}" {e}')
|
||||
continue
|
||||
if image is None:
|
||||
continue
|
||||
shared.state.textinfo = name
|
||||
@@ -100,7 +114,7 @@ def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemp
|
||||
return outputs, info, params
|
||||
|
||||
|
||||
def run_extras(extras_mode, resize_mode, image, image_folder, input_dir, output_dir, show_extras_results, upscaling_resize, upscaling_resize_w, upscaling_resize_h, upscaling_crop, extras_upscaler_1, extras_upscaler_2, extras_upscaler_2_visibility, save_output: bool = True, script_args: dict | None = None):
|
||||
def run_extras(extras_mode, resize_mode, image, image_folder, input_dir, output_dir, video, show_extras_results, upscaling_resize, upscaling_resize_w, upscaling_resize_h, upscaling_crop, extras_upscaler_1, extras_upscaler_2, extras_upscaler_2_visibility, save_output: bool = True, script_args: dict | None = None):
|
||||
"""old handler for API"""
|
||||
|
||||
merged = {
|
||||
@@ -120,4 +134,4 @@ def run_extras(extras_mode, resize_mode, image, image_folder, input_dir, output_
|
||||
merged.setdefault(name, {}).update(kvs or {})
|
||||
args = scripts_manager.scripts_postproc.create_args_for_run(merged)
|
||||
|
||||
return run_postprocessing(extras_mode, image, image_folder, input_dir, output_dir, show_extras_results, *args, save_output=save_output)
|
||||
return run_postprocessing(extras_mode, image, image_folder, input_dir, output_dir, video, show_extras_results, *args, save_output=save_output)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import os
|
||||
import gradio as gr
|
||||
from modules import scripts_manager, shared, ui_common, postprocessing, call_queue, generation_parameters_copypaste
|
||||
from modules.logger import log
|
||||
@@ -12,9 +13,26 @@ def submit_info(image):
|
||||
return infotext_to_html(geninfo), info, geninfo
|
||||
|
||||
|
||||
def submit_process(tab_index, extras_image, image_batch, extras_batch_input_dir, extras_batch_output_dir, show_extras_results, save_output, *script_inputs):
|
||||
def submit_video(video):
|
||||
if not video or not isinstance(video, str) or not os.path.isfile(video):
|
||||
return '', '', ''
|
||||
from modules.video import get_video_info, get_video_metadata
|
||||
info = get_video_info(video)
|
||||
metadata = get_video_metadata(video)
|
||||
if metadata:
|
||||
info['metadata'] = metadata
|
||||
text = ''
|
||||
html = ''
|
||||
html = [f'<b>{k}</b>: {v}' for k, v in info.items()]
|
||||
html = '<br>'.join(html)
|
||||
text = [f'{k}: {v}' for k, v in info.items()]
|
||||
text = ', '.join(text)
|
||||
return html, '', text
|
||||
|
||||
|
||||
def submit_process(tab_index, extras_image, image_batch, extras_batch_input_dir, extras_batch_output_dir, extras_video, show_extras_results, save_output, *script_inputs):
|
||||
from modules.ui_common import infotext_to_html
|
||||
result_images, geninfo, _js_info = postprocessing.run_postprocessing(tab_index, extras_image, image_batch, extras_batch_input_dir, extras_batch_output_dir, show_extras_results, *script_inputs, save_output=save_output)
|
||||
result_images, geninfo, _js_info = postprocessing.run_postprocessing(tab_index, extras_image, image_batch, extras_batch_input_dir, extras_batch_output_dir, extras_video, show_extras_results, *script_inputs, save_output=save_output)
|
||||
return result_images, geninfo, infotext_to_html(geninfo)
|
||||
|
||||
|
||||
@@ -25,14 +43,15 @@ def create_ui():
|
||||
with gr.Column(variant='compact'):
|
||||
with gr.Tabs(elem_id="mode_extras"):
|
||||
with gr.Tab('Process Image', id="single_image", elem_id="extras_single_tab") as tab_single:
|
||||
with gr.Row():
|
||||
extras_image = gr.Image(label="Source", interactive=True, type="pil", elem_id="extras_image")
|
||||
extras_image = gr.Image(label="Source", interactive=True, type="pil", elem_id="extras_image")
|
||||
with gr.Tab('Process Batch', id="batch_process", elem_id="extras_batch_process_tab") as tab_batch:
|
||||
image_batch = gr.Files(label="Batch process", interactive=True, elem_id="extras_image_batch")
|
||||
with gr.Tab('Process Folder', id="batch_from_directory", elem_id="extras_batch_directory_tab") as tab_batch_dir:
|
||||
extras_batch_input_dir = gr.Textbox(label="Input directory", **shared.hide_dirs, placeholder="A directory on the same machine where the server is running.", elem_id="extras_batch_input_dir")
|
||||
extras_batch_output_dir = gr.Textbox(label="Output directory", **shared.hide_dirs, placeholder="Leave blank to save images to the default path.", elem_id="extras_batch_output_dir")
|
||||
show_extras_results = gr.Checkbox(label='Show result images', value=True, elem_id="extras_show_extras_results")
|
||||
with gr.Tab('Process Video', id="process_video", elem_id="extras_process_video_tab") as tab_process_video:
|
||||
extras_video = gr.Video(label="Input Video", show_label=False, interactive=True, elem_id="extras_video")
|
||||
with gr.Row():
|
||||
save_output = gr.Checkbox(label='Save output', value=True, elem_id="extras_save_output")
|
||||
|
||||
@@ -60,7 +79,11 @@ def create_ui():
|
||||
tab_single.select(fn=lambda: 0, inputs=[], outputs=[tab_index])
|
||||
tab_batch.select(fn=lambda: 1, inputs=[], outputs=[tab_index])
|
||||
tab_batch_dir.select(fn=lambda: 2, inputs=[], outputs=[tab_index])
|
||||
tab_process_video.select(fn=lambda: 3, inputs=[], outputs=[tab_index])
|
||||
|
||||
extras_image.change(fn=submit_info, inputs=[extras_image], outputs=[html_info_formatted, exif_info, generation_info])
|
||||
extras_video.change(fn=submit_video, inputs=[extras_video], outputs=[html_info_formatted, exif_info, generation_info])
|
||||
|
||||
submit.click(
|
||||
_js="submit_postprocessing",
|
||||
fn=call_queue.wrap_gradio_gpu_call(submit_process, extra_outputs=[None, ''], name='Postprocess'),
|
||||
@@ -70,6 +93,7 @@ def create_ui():
|
||||
image_batch,
|
||||
extras_batch_input_dir,
|
||||
extras_batch_output_dir,
|
||||
extras_video,
|
||||
show_extras_results,
|
||||
save_output,
|
||||
*script_inputs,
|
||||
|
||||
+93
-1
@@ -105,7 +105,10 @@ def get_video_params(filepath: str, capture: bool = False):
|
||||
raise RuntimeError(msg)
|
||||
frames = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
|
||||
fps = round(video.get(cv2.CAP_PROP_FPS), 2)
|
||||
duration = round(float(frames) / fps, 2)
|
||||
if fps > 0:
|
||||
duration = round(float(frames) / fps, 2)
|
||||
else:
|
||||
duration = 0
|
||||
w, h = int(video.get(cv2.CAP_PROP_FRAME_WIDTH)), int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||||
codec = decode_fourcc(video.get(cv2.CAP_PROP_FOURCC))
|
||||
frame = None
|
||||
@@ -115,3 +118,92 @@ def get_video_params(filepath: str, capture: bool = False):
|
||||
frame = Image.fromarray(frame)
|
||||
video.release()
|
||||
return frames, fps, duration, w, h, codec, frame
|
||||
|
||||
|
||||
def get_video_info(filepath: str):
|
||||
import cv2
|
||||
from modules.control.util import decode_fourcc
|
||||
|
||||
try:
|
||||
info = {
|
||||
'file': os.path.basename(filepath),
|
||||
'size': os.path.getsize(filepath),
|
||||
'container': os.path.splitext(filepath)[1].lower().lstrip('.'),
|
||||
}
|
||||
except Exception as e:
|
||||
log.error(f'Video probe failed: path="{filepath}" {e}')
|
||||
return {}
|
||||
|
||||
def get_prop(video, name: str, label: str | None, cast: type = float):
|
||||
prop_id = getattr(cv2, name, None)
|
||||
if prop_id is None:
|
||||
return None
|
||||
try:
|
||||
value = video.get(prop_id)
|
||||
if value is None:
|
||||
return None
|
||||
value = cast(value)
|
||||
if value == 0:
|
||||
return None
|
||||
if label is not None:
|
||||
info[label] = value
|
||||
return value
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
video = cv2.VideoCapture(filepath)
|
||||
try:
|
||||
if not video.isOpened():
|
||||
msg = f'Video open failed: path="{filepath}"'
|
||||
info['error'] = msg
|
||||
log.error(msg)
|
||||
return info
|
||||
try:
|
||||
backend = video.getBackendName()
|
||||
if backend:
|
||||
info['backend'] = backend
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
frames = get_prop(video, 'CAP_PROP_FRAME_COUNT', 'frames', int) or 0
|
||||
fps = get_prop(video, 'CAP_PROP_FPS', 'fps', float) or 0
|
||||
_width = get_prop(video, 'CAP_PROP_FRAME_WIDTH', 'width', int) or 0
|
||||
_height = get_prop(video, 'CAP_PROP_FRAME_HEIGHT', 'height', int) or 0
|
||||
_bitrate = get_prop(video, 'CAP_PROP_BITRATE', 'bitrate_kbps', float)
|
||||
_orientation = get_prop(video, 'CAP_PROP_ORIENTATION_META', 'rotation', float)
|
||||
|
||||
if frames > 0 and fps > 0:
|
||||
info['duration'] = round(float(frames) / fps, 3)
|
||||
|
||||
fourcc = get_prop(video, 'CAP_PROP_FOURCC', None, int) or 0
|
||||
info['codec'] = decode_fourcc(fourcc)
|
||||
|
||||
pix = get_prop(video, 'CAP_PROP_CODEC_PIXEL_FORMAT', None, int) or 0
|
||||
info['format'] = decode_fourcc(pix)
|
||||
|
||||
sar_num = get_prop(video, 'CAP_PROP_SAR_NUM', None, int) or 1
|
||||
sar_den = get_prop(video, 'CAP_PROP_SAR_DEN', None, int) or 1
|
||||
if (sar_num > 0 and sar_den > 0) and (sar_num != 1 or sar_den != 1):
|
||||
info['sar'] = f'{sar_num}/{sar_den}'
|
||||
|
||||
audio_streams = get_prop(video, 'CAP_PROP_AUDIO_TOTAL_STREAMS', 'audio_streams', int) or 0
|
||||
audio_channels = get_prop(video, 'CAP_PROP_AUDIO_TOTAL_CHANNELS', 'audio_channels', int) or 0
|
||||
audio_sample_rate = get_prop(video, 'CAP_PROP_AUDIO_SAMPLES_PER_SECOND', 'audio_sample_rate', int) or 0
|
||||
if audio_streams > 0 or audio_channels > 0 or audio_sample_rate > 0:
|
||||
info['audio'] = f'{audio_streams}x{audio_channels}x{audio_sample_rate}'
|
||||
|
||||
return info
|
||||
except Exception as e:
|
||||
log.error(f'Video probe failed: path="{filepath}" {e}')
|
||||
return info
|
||||
finally:
|
||||
video.release()
|
||||
|
||||
|
||||
def get_video_metadata(video):
|
||||
if not video or not isinstance(video, str) or not os.path.isfile(video):
|
||||
return {}
|
||||
from modules.video_models.video_utils import check_av
|
||||
av = check_av()
|
||||
with av.open(video, mode="r") as container:
|
||||
return dict(container.metadata)
|
||||
|
||||
Reference in New Issue
Block a user