Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- """
- AWarpSharp2 - Pixel-accurate Python port of dubhater's original VapourSynth-AWarpSharp2.
- Original C++: https://github.com/dubhater/vapoursynth-awarpsharp2
- License: MIT
- """
- import vapoursynth as vs
- import numpy as np
- __version__ = '5.3.0'
- # =====================================================================
- # SOBEL (pixel-accurate to C++ sobel_c)
- # =====================================================================
- def _sobel_c(src: np.ndarray, thresh: int, bits_per_sample: int) -> np.ndarray:
- """Port of dubhater's sobel_c() function."""
- height, width = src.shape
- dst = np.empty_like(src)
- pixel_max = (1 << bits_per_sample) - 1
- src_int = src.astype(np.int32)
- for y in range(1, height - 1):
- for x in range(1, width - 1):
- a11 = src_int[y - 1, x - 1]
- a21 = src_int[y - 1, x]
- a31 = src_int[y - 1, x + 1]
- a12 = src_int[y, x - 1]
- a32 = src_int[y, x + 1]
- a13 = src_int[y + 1, x - 1]
- a23 = src_int[y + 1, x]
- a33 = src_int[y + 1, x + 1]
- avg_up = (a21 + ((a11 + a31 + 1) >> 1) + 1) >> 1
- avg_down = (a23 + ((a13 + a33 + 1) >> 1) + 1) >> 1
- avg_left = (a12 + ((a13 + a11 + 1) >> 1) + 1) >> 1
- avg_right = (a32 + ((a33 + a31 + 1) >> 1) + 1) >> 1
- abs_v = abs(avg_up - avg_down)
- abs_h = abs(avg_left - avg_right)
- absolute = min(abs_v + abs_h, pixel_max)
- abs_max = max(abs_h, abs_v)
- absolute = min(absolute + abs_max, pixel_max)
- absolute = min(min(absolute * 2, pixel_max) + absolute, pixel_max)
- absolute = min(absolute * 2, pixel_max)
- dst[y, x] = min(absolute, thresh)
- for y in range(1, height - 1):
- dst[y, 0] = dst[y, 1]
- dst[y, width - 1] = dst[y, width - 2]
- dst[0, :] = dst[1, :]
- dst[height - 1, :] = dst[height - 2, :]
- return dst
- # =====================================================================
- # BLUR R6 (pixel-accurate to C++ blur_r6_c)
- # =====================================================================
- def _blur_r6_c(mask: np.ndarray, temp: np.ndarray) -> None:
- """Port of dubhater's blur_r6_c() function."""
- height, width = mask.shape
- mask_int = mask.astype(np.int32)
- temp_int = temp.astype(np.int32)
- for y in range(height):
- for x in range(6):
- avg12 = (mask_int[y, x + 1] + mask_int[y, x + 2] + 1) >> 1
- avg34 = (mask_int[y, x + 3] + mask_int[y, x + 4] + 1) >> 1
- avg56 = (mask_int[y, x + 5] + mask_int[y, x + 6] + 1) >> 1
- avg012 = (mask_int[y, x] + avg12 + 1) >> 1
- avg3456 = (avg34 + avg56 + 1) >> 1
- avg0123456 = (avg012 + avg3456 + 1) >> 1
- avg = (avg012 + avg0123456 + 1) >> 1
- temp_int[y, x] = avg
- for x in range(6, width - 6):
- avg11 = (mask_int[y, x - 1] + mask_int[y, x + 1] + 1) >> 1
- avg22 = (mask_int[y, x - 2] + mask_int[y, x + 2] + 1) >> 1
- avg33 = (mask_int[y, x - 3] + mask_int[y, x + 3] + 1) >> 1
- avg44 = (mask_int[y, x - 4] + mask_int[y, x + 4] + 1) >> 1
- avg55 = (mask_int[y, x - 5] + mask_int[y, x + 5] + 1) >> 1
- avg66 = (mask_int[y, x - 6] + mask_int[y, x + 6] + 1) >> 1
- avg12 = (avg11 + avg22 + 1) >> 1
- avg34 = (avg33 + avg44 + 1) >> 1
- avg56 = (avg55 + avg66 + 1) >> 1
- avg012 = (mask_int[y, x] + avg12 + 1) >> 1
- avg3456 = (avg34 + avg56 + 1) >> 1
- avg0123456 = (avg012 + avg3456 + 1) >> 1
- avg = (avg012 + avg0123456 + 1) >> 1
- temp_int[y, x] = avg
- for x in range(width - 6, width):
- avg12 = (mask_int[y, x - 1] + mask_int[y, x - 2] + 1) >> 1
- avg34 = (mask_int[y, x - 3] + mask_int[y, x - 4] + 1) >> 1
- avg56 = (mask_int[y, x - 5] + mask_int[y, x - 6] + 1) >> 1
- avg012 = (mask_int[y, x] + avg12 + 1) >> 1
- avg3456 = (avg34 + avg56 + 1) >> 1
- avg0123456 = (avg012 + avg3456 + 1) >> 1
- avg = (avg012 + avg0123456 + 1) >> 1
- temp_int[y, x] = avg
- for y in range(6):
- for x in range(width):
- l0 = temp_int[y, x]
- l1 = temp_int[y + 1, x] if y + 1 < height else temp_int[y, x]
- l2 = temp_int[y + 2, x] if y + 2 < height else temp_int[y, x]
- l3 = temp_int[y + 3, x] if y + 3 < height else temp_int[y, x]
- l4 = temp_int[y + 4, x] if y + 4 < height else temp_int[y, x]
- l5 = temp_int[y + 5, x] if y + 5 < height else temp_int[y, x]
- l6 = temp_int[y + 6, x] if y + 6 < height else temp_int[y, x]
- avg12 = (l1 + l2 + 1) >> 1
- avg34 = (l3 + l4 + 1) >> 1
- avg56 = (l5 + l6 + 1) >> 1
- avg3456 = (avg34 + avg56 + 1) >> 1
- avg012 = (l0 + avg12 + 1) >> 1
- avg0123456 = (avg012 + avg3456 + 1) >> 1
- avg = (avg012 + avg0123456 + 1) >> 1
- mask[y, x] = avg
- for y in range(6, height - 6):
- for x in range(width):
- m6 = temp_int[y - 6, x]
- m5 = temp_int[y - 5, x]
- m4 = temp_int[y - 4, x]
- m3 = temp_int[y - 3, x]
- m2 = temp_int[y - 2, x]
- m1 = temp_int[y - 1, x]
- l0 = temp_int[y, x]
- l1 = temp_int[y + 1, x]
- l2 = temp_int[y + 2, x]
- l3 = temp_int[y + 3, x]
- l4 = temp_int[y + 4, x]
- l5 = temp_int[y + 5, x]
- l6 = temp_int[y + 6, x]
- avg11 = (m1 + l1 + 1) >> 1
- avg22 = (m2 + l2 + 1) >> 1
- avg33 = (m3 + l3 + 1) >> 1
- avg44 = (m4 + l4 + 1) >> 1
- avg55 = (m5 + l5 + 1) >> 1
- avg66 = (m6 + l6 + 1) >> 1
- avg12 = (avg11 + avg22 + 1) >> 1
- avg34 = (avg33 + avg44 + 1) >> 1
- avg56 = (avg55 + avg66 + 1) >> 1
- avg012 = (l0 + avg12 + 1) >> 1
- avg3456 = (avg34 + avg56 + 1) >> 1
- avg0123456 = (avg012 + avg3456 + 1) >> 1
- avg = (avg012 + avg0123456 + 1) >> 1
- mask[y, x] = avg
- for y in range(height - 6, height):
- for x in range(width):
- m6 = temp_int[y - 6, x] if y - 6 >= 0 else temp_int[0, x]
- m5 = temp_int[y - 5, x] if y - 5 >= 0 else temp_int[0, x]
- m4 = temp_int[y - 4, x] if y - 4 >= 0 else temp_int[0, x]
- m3 = temp_int[y - 3, x] if y - 3 >= 0 else temp_int[0, x]
- m2 = temp_int[y - 2, x] if y - 2 >= 0 else temp_int[0, x]
- m1 = temp_int[y - 1, x] if y - 1 >= 0 else temp_int[0, x]
- l0 = temp_int[y, x]
- avg12 = (m1 + m2 + 1) >> 1
- avg34 = (m3 + m4 + 1) >> 1
- avg56 = (m5 + m6 + 1) >> 1
- avg012 = (l0 + avg12 + 1) >> 1
- avg3456 = (avg34 + avg56 + 1) >> 1
- avg0123456 = (avg012 + avg3456 + 1) >> 1
- avg = (avg012 + avg0123456 + 1) >> 1
- mask[y, x] = avg
- # =====================================================================
- # BLUR R2 (pixel-accurate to C++ blur_r2_c)
- # =====================================================================
- def _blur_r2_c(mask: np.ndarray, temp: np.ndarray) -> None:
- """Port of dubhater's blur_r2_c() function."""
- height, width = mask.shape
- mask_int = mask.astype(np.int32)
- temp_int = temp.astype(np.int32)
- for y in range(height):
- avg1 = (mask_int[y, 0] + mask_int[y, 1] + 1) >> 1
- avg2 = (mask_int[y, 0] + mask_int[y, 2] + 1) >> 1 if width > 2 else (mask_int[y, 0] + mask_int[y, 1] + 1) >> 1
- avg = (avg2 + mask_int[y, 0] + 1) >> 1
- avg = (avg + mask_int[y, 0] + 1) >> 1
- avg = (avg + avg1 + 1) >> 1
- temp_int[y, 0] = avg
- if width > 1:
- avg1 = (mask_int[y, 0] + mask_int[y, 2] + 1) >> 1 if width > 2 else (mask_int[y, 0] + mask_int[y, 1] + 1) >> 1
- avg2 = (mask_int[y, 0] + mask_int[y, 3] + 1) >> 1 if width > 3 else (mask_int[y, 0] + mask_int[y, 1] + 1) >> 1
- avg = (avg2 + mask_int[y, 1] + 1) >> 1
- avg = (avg + mask_int[y, 1] + 1) >> 1
- avg = (avg + avg1 + 1) >> 1
- temp_int[y, 1] = avg
- for x in range(2, width - 2):
- avg1 = (mask_int[y, x - 1] + mask_int[y, x + 1] + 1) >> 1
- avg2 = (mask_int[y, x - 2] + mask_int[y, x + 2] + 1) >> 1
- avg = (avg2 + mask_int[y, x] + 1) >> 1
- avg = (avg + mask_int[y, x] + 1) >> 1
- avg = (avg + avg1 + 1) >> 1
- temp_int[y, x] = avg
- if width > 2:
- x = width - 2
- avg1 = (mask_int[y, x - 1] + mask_int[y, x + 1] + 1) >> 1 if x + 1 < width else (mask_int[y, x - 1] + mask_int[y, x] + 1) >> 1
- avg2 = (mask_int[y, x - 2] + mask_int[y, x + 1] + 1) >> 1 if x + 1 < width else (mask_int[y, x - 2] + mask_int[y, x] + 1) >> 1
- avg = (avg2 + mask_int[y, x] + 1) >> 1
- avg = (avg + mask_int[y, x] + 1) >> 1
- avg = (avg + avg1 + 1) >> 1
- temp_int[y, x] = avg
- x = width - 1
- avg1 = (mask_int[y, x - 1] + mask_int[y, x] + 1) >> 1
- avg2 = (mask_int[y, x - 2] + mask_int[y, x] + 1) >> 1 if x - 2 >= 0 else (mask_int[y, x - 1] + mask_int[y, x] + 1) >> 1
- avg = (avg2 + mask_int[y, x] + 1) >> 1
- avg = (avg + mask_int[y, x] + 1) >> 1
- avg = (avg + avg1 + 1) >> 1
- temp_int[y, x] = avg
- for y in range(height):
- stride_p1 = -1 if y > 0 else 0
- stride_p2 = stride_p1 * 2 if y > 1 else stride_p1
- stride_n1 = 1 if y < height - 1 else 0
- stride_n2 = stride_n1 * 2 if y < height - 2 else stride_n1
- for x in range(width):
- m2 = temp_int[max(0, y + stride_p2), x]
- m1 = temp_int[max(0, y + stride_p1), x]
- l0 = temp_int[y, x]
- l1 = temp_int[min(height - 1, y + stride_n1), x]
- l2 = temp_int[min(height - 1, y + stride_n2), x]
- avg1 = (m1 + l1 + 1) >> 1
- avg2 = (m2 + l2 + 1) >> 1
- avg = (avg2 + l0 + 1) >> 1
- avg = (avg + l0 + 1) >> 1
- avg = (avg + avg1 + 1) >> 1
- mask[y, x] = avg
- # =====================================================================
- # BILINEAR DOWNSCALE
- # =====================================================================
- def _bilinear_downscale_h_c(src: np.ndarray, cplace: str) -> None:
- """Port of dubhater's bilinear_downscale_h_c() function."""
- height, width = src.shape
- if cplace == 'mpeg1':
- for y in range(height):
- for x in range(width // 2):
- src[y, x] = (src[y, x * 2] + src[y, x * 2 + 1] + 1) >> 1
- else: # mpeg2
- for y in range(height):
- c2 = src[y, 0]
- for x in range(width // 2):
- c0 = c2
- c1 = src[y, x * 2]
- c2 = src[y, x * 2 + 1]
- src[y, x] = (c0 + 2 * c1 + c2 + 2) >> 2
- def _bilinear_downscale_v_c(src: np.ndarray) -> None:
- """Port of dubhater's bilinear_downscale_v_c() function."""
- height, width = src.shape
- for y in range(height // 2):
- for x in range(width):
- src[y, x] = (src[y, x] + src[y + 1, x] + 1) >> 1
- def _bilinear_downscale_hv_c(src: np.ndarray, cplace: str) -> None:
- """Port of dubhater's bilinear_downscale_hv_c() function."""
- height, width = src.shape
- if cplace == 'mpeg1':
- for y in range(height // 2):
- for x in range(width // 2):
- avg1 = (src[y, x * 2] + src[y, x * 2 + 1] + 1) >> 1
- avg2 = (src[y + 1, x * 2] + src[y + 1, x * 2 + 1] + 1) >> 1
- src[y, x] = (avg1 + avg2 + 1) >> 1
- else: # mpeg2
- for y in range(height // 2):
- c2 = src[y, 0] + src[y + 1, 0]
- for x in range(width // 2):
- c0 = c2
- c1 = src[y, x * 2] + src[y + 1, x * 2]
- c2 = src[y, x * 2 + 1] + src[y + 1, x * 2 + 1]
- src[y, x] = (c0 + 2 * c1 + c2 + 4) >> 3
- # =====================================================================
- # WARP (pixel-accurate to C++ warp_c)
- # =====================================================================
- def _warp_c(src: np.ndarray, edge: np.ndarray, depth: int, bits_per_sample: int, smagl: int = 0) -> np.ndarray:
- """Port of dubhater's warp_c() function."""
- height, width = edge.shape
- dst = np.empty_like(edge)
- extra_bits = bits_per_sample - 8
- pixel_max = (1 << bits_per_sample) - 1
- SMAG = 1 << smagl
- depth_shifted = depth << 8
- x_limit_min = 0 * SMAG
- x_limit_max = (width - 1) * SMAG
- for y in range(height):
- y_limit_min = -y * 128
- y_limit_max = (height - y) * 128 - 129
- for x in range(width):
- if y == 0:
- above = edge[y, x]
- else:
- above = edge[y - 1, x]
- if y == height - 1:
- below = edge[y, x]
- else:
- below = edge[y + 1, x]
- if x == 0:
- left = edge[y, x]
- else:
- left = edge[y, x - 1]
- if x == width - 1:
- right = edge[y, x]
- else:
- right = edge[y, x + 1]
- h = int(left) - int(right)
- v = int(above) - int(below)
- if extra_bits > 0:
- h >>= extra_bits
- v >>= extra_bits
- h <<= 7
- v <<= 7
- h *= depth_shifted
- h >>= 16
- v *= depth_shifted
- v >>= 16
- v = max(v, y_limit_min)
- v = min(v, y_limit_max)
- remainder_h = h
- remainder_v = v
- if smagl:
- remainder_h <<= smagl
- remainder_v <<= smagl
- remainder_h &= 127
- remainder_v &= 127
- h >>= 7 - smagl
- v >>= 7 - smagl
- h += x << smagl
- # CRITICAL FIX: v must be converted to an absolute row coordinate
- v += y << smagl
- remainder_needed = (x_limit_max > h) and not (x_limit_min > h)
- if not remainder_needed:
- remainder_h = 0
- h = min(h, x_limit_max)
- h = max(h, x_limit_min)
- h = min(h, width - 2)
- v = min(v, height - 2)
- s00 = int(src[v, h])
- s01 = int(src[v, h + 1])
- s10 = int(src[v + 1, h])
- s11 = int(src[v + 1, h + 1])
- s0 = s00 * (128 - remainder_h)
- s1 = s10 * (128 - remainder_h)
- s0 += s01 * remainder_h
- s1 += s11 * remainder_h
- s0 += 64
- s1 += 64
- s0 >>= 7
- s1 >>= 7
- s0 *= 128 - remainder_v
- s1 *= remainder_v
- s = s0 + s1
- s += 64
- s >>= 7
- dst[y, x] = min(max(s, 0), pixel_max)
- return dst
- # =====================================================================
- # Helpers
- # =====================================================================
- def _copy_to_plane(frame: vs.VideoFrame, plane: int, data: np.ndarray) -> None:
- """Copy processed data back into a frame plane, handling R60+ API."""
- try:
- arr = frame.get_write_array(plane)
- except AttributeError:
- arr = np.asarray(frame[plane])
- np.copyto(arr, data)
- # =====================================================================
- # VAPOURSYNTH INTERFACE
- # =====================================================================
- def AWarpSharp2(
- clip: vs.VideoNode,
- thresh: int = 128,
- blur: int = None,
- type: int = 0,
- depth: int | list[int] = None,
- chroma: int = 0,
- planes: list[int] | None = None,
- cplace: str = 'mpeg1'
- ) -> vs.VideoNode:
- """
- AWarpSharp2 - pixel-accurate port of dubhater's original.
- """
- core = vs.core
- fmt = clip.format
- if not fmt:
- raise vs.Error("AWarpSharp2: clip must have constant format")
- if fmt.sample_type != vs.INTEGER or fmt.bits_per_sample > 16 or fmt.color_family == vs.RGB:
- raise vs.Error("AWarpSharp2: only 8-16 bit integer, non-RGB clips supported")
- is_gray = fmt.color_family == vs.GRAY
- if planes is None:
- planes = [0] if is_gray else [0, 1, 2]
- if depth is None:
- depth_list = [16] if is_gray else [16, 8, 8]
- elif isinstance(depth, int):
- depth_list = [depth]
- else:
- depth_list = list(depth)
- while len(depth_list) < fmt.num_planes:
- depth_list.append(depth_list[-1])
- if blur is None:
- blur = 3 if type == 1 else 2
- if thresh < 0 or thresh > 255:
- raise vs.Error("AWarpSharp2: thresh must be between 0 and 255")
- if blur < 0:
- raise vs.Error("AWarpSharp2: blur must be at least 0")
- if type < 0 or type > 1:
- raise vs.Error("AWarpSharp2: type must be 0 or 1")
- for d in depth_list:
- if d < -128 or d > 127:
- raise vs.Error("AWarpSharp2: depth must be between -128 and 127")
- if chroma < 0 or chroma > 1:
- raise vs.Error("AWarpSharp2: chroma must be 0 or 1")
- thresh_scaled = thresh << (fmt.bits_per_sample - 8)
- def process_frame(n, f):
- fout = f.copy()
- mask_y = None
- if 0 in planes or (chroma == 0 and not is_gray and (1 in planes or 2 in planes)):
- src_y = np.array(f[0])
- mask_y = _sobel_c(src_y.copy(), thresh_scaled, fmt.bits_per_sample)
- temp_y = np.empty_like(mask_y)
- bl = blur
- for _ in range(bl):
- if type == 0:
- _blur_r6_c(mask_y, temp_y)
- else:
- _blur_r2_c(mask_y, temp_y)
- if 0 in planes:
- dst_y = _warp_c(src_y, mask_y, depth_list[0], fmt.bits_per_sample)
- _copy_to_plane(fout, 0, dst_y)
- if not is_gray and (1 in planes or 2 in planes):
- if chroma == 0 and mask_y is not None:
- # WarpAlongLuma: use luma mask (downscaled if needed)
- if fmt.subsampling_w or fmt.subsampling_h:
- mask_y_copy = mask_y.copy()
- if fmt.subsampling_w and fmt.subsampling_h:
- _bilinear_downscale_hv_c(mask_y_copy, cplace)
- mask_y_copy = mask_y_copy[:mask_y.shape[0]//2, :mask_y.shape[1]//2]
- elif fmt.subsampling_w:
- _bilinear_downscale_h_c(mask_y_copy, cplace)
- mask_y_copy = mask_y_copy[:, :mask_y.shape[1]//2]
- elif fmt.subsampling_h:
- _bilinear_downscale_v_c(mask_y_copy)
- mask_y_copy = mask_y_copy[:mask_y.shape[0]//2, :]
- else:
- mask_y_copy = mask_y
- for plane_idx in [1, 2]:
- if plane_idx not in planes:
- continue
- src_uv = np.array(f[plane_idx])
- dst_uv = _warp_c(src_uv, mask_y_copy, depth_list[plane_idx], fmt.bits_per_sample)
- _copy_to_plane(fout, plane_idx, dst_uv)
- else:
- # WarpAlongChroma: each plane has its own mask
- for plane_idx in [1, 2]:
- if plane_idx not in planes:
- continue
- src_uv = np.array(f[plane_idx])
- mask_uv = _sobel_c(src_uv.copy(), thresh_scaled, fmt.bits_per_sample)
- temp_uv = np.empty_like(mask_uv)
- bl = (blur + 1) // 2
- for _ in range(bl):
- if type == 0:
- _blur_r6_c(mask_uv, temp_uv)
- else:
- _blur_r2_c(mask_uv, temp_uv)
- dst_uv = _warp_c(src_uv, mask_uv, depth_list[plane_idx], fmt.bits_per_sample)
- _copy_to_plane(fout, plane_idx, dst_uv)
- return fout
- return core.std.ModifyFrame(clip=clip, clips=clip, selector=process_frame)
- # VapourSynth R60+ plugin registration
- try:
- if hasattr(vs, 'register_function'):
- @vs.register_function()
- def AWarpSharp2_VS(clip: vs.VideoNode, thresh: int = 128, blur: int = None,
- type: int = 0, depth: int = None,
- chroma: int = 0, planes: list[int] = None,
- cplace: str = 'mpeg1') -> vs.VideoNode:
- return AWarpSharp2(clip, thresh, blur, type, depth, chroma, planes, cplace)
- except Exception:
- pass
Advertisement
Add Comment
Please, Sign In to add comment