Как переместить playhead через Python API

"""S04 author helpers: exact timecode math and explicitly scoped Resolve reads.

Python 3.8+. Standard library only. Not a Blackmagic SDK, transaction or sandbox.
String timecode policy: HH:MM:SS:FF (NDF) / HH:MM:SS;FF (DF), within one 24h day.
Supported timebases: 24/25/30/48/50/60 and 24000/30000/48000/60000 over 1001.
DF is deliberately limited to 30000/1001 and 60000/1001. No silent rate aliases.
Only move_playhead(..., apply=True) calls a setter, once, after reviewed guards.
"""
from dataclasses import dataclass
from fractions import Fraction
import hashlib
import json
import re
from typing import Any, Dict


class ClockError(ValueError):
    """Input or observed state does not meet this example's contract."""


class OutcomeUnknown(RuntimeError):
    """A setter was dispatched but its result was not fully verified. Do not retry."""


def integer(value: Any, name: str) -> int:
    # bool is a subclass of int; never accept it as a frame index.
    if type(value) is not int:
        raise ClockError(name + " must be an integer, not a float, bool or string")
    return value


def exact_number(value: Any, name: str) -> Fraction:
    if isinstance(value, Fraction):
        return value
    if type(value) is int:
        return Fraction(value)
    if not isinstance(value, str) or len(value) > 80:
        raise ClockError(name + " must be exact decimal/rational text (not float)")
    # Avoid scientific exponents/unbounded integers and locale-dependent commas.
    if not re.fullmatch(r"[+-]?[0-9]{1,18}(?:\.[0-9]{1,18}|/[0-9]{1,18})?", value):
        raise ClockError(name + " requires ASCII decimal or numerator/denominator text")
    try:
        return Fraction(value)
    except (ValueError, ZeroDivisionError) as exc:
        raise ClockError(name + " is invalid") from exc


SUPPORTED = {Fraction(x): x for x in (24, 25, 30, 48, 50, 60)}
SUPPORTED.update({Fraction(x * 1000, 1001): x for x in (24, 30, 48, 60)})


@dataclass(frozen=True)
class Clock:
    rate: Fraction
    mode: str

    def __post_init__(self):
        rate = exact_number(self.rate, "rate")
        object.__setattr__(self, "rate", rate)
        if rate not in SUPPORTED:
            raise ClockError("Unsupported/ambiguous rate. Use e.g. 24 or 30000/1001, not 29.97")
        if self.mode not in ("ndf", "df"):
            raise ClockError("mode must explicitly be ndf or df")
        if self.mode == "df" and rate not in (Fraction(30000, 1001), Fraction(60000, 1001)):
            raise ClockError("DF example supports only 30000/1001 and 60000/1001")

    @property
    def nominal(self) -> int:
        return SUPPORTED[self.rate]

    @property
    def drop(self) -> int:
        return self.nominal // 15 if self.mode == "df" else 0

    @property
    def frames_per_day(self) -> int:
        return self.nominal * 86400 - self.drop * (1440 - 144)


def frames_to_timecode(frames: int, clock: Clock) -> str:
    """Actual whole frames since TC zero -> label. No wrapping or rounding."""
    frames = integer(frames, "frames")
    if not 0 <= frames < clock.frames_per_day:
        raise ClockError("Frame count must be within one timecode day; no negative/wrap")
    labelled = frames
    if clock.drop:
        # Independent implementation of the ten-minute/minute numbering rule.
        per_ten = clock.nominal * 600 - clock.drop * 9
        per_short_minute = clock.nominal * 60 - clock.drop
        tens, rest = divmod(frames, per_ten)
        skipped_minutes = max(0, (rest - clock.drop) // per_short_minute)
        labelled += clock.drop * (tens * 9 + skipped_minutes)
    seconds, ff = divmod(labelled, clock.nominal)
    hours, seconds = divmod(seconds, 3600)
    minutes, seconds = divmod(seconds, 60)
    separator = ";" if clock.mode == "df" else ":"
    return "{:02d}:{:02d}:{:02d}{}{:02d}".format(hours, minutes, seconds, separator, ff)


def timecode_to_frames(timecode: str, clock: Clock) -> int:
    """Label -> actual whole frames since TC zero. Invalid labels never become 0."""
    if not isinstance(timecode, str):
        raise ClockError("timecode must be text")
    match = re.fullmatch(r"([0-9]{2}):([0-9]{2}):([0-9]{2})([:;])([0-9]{2})", timecode)
    if not match:
        raise ClockError("Expected HH:MM:SS:FF (NDF) or HH:MM:SS;FF (DF)")
    h, m, s, sep, f = match.groups()
    h, m, s, f = map(int, (h, m, s, f))
    if h >= 24 or m >= 60 or s >= 60 or f >= clock.nominal:
        raise ClockError("Timecode field outside this clock's range")
    if sep != (";" if clock.mode == "df" else ":"):
        raise ClockError("Timecode separator disagrees with explicit DF/NDF mode")
    if clock.drop and m % 10 != 0 and s == 0 and f < clock.drop:
        raise ClockError("This label is skipped by drop-frame numbering")
    minutes = h * 60 + m
    count = (h * 3600 + m * 60 + s) * clock.nominal + f
    return count - clock.drop * (minutes - minutes // 10)


def offset_to_timecode(offset: int, start_timecode: str, clock: Clock) -> str:
    offset = integer(offset, "offset")
    if offset < 0:
        raise ClockError("Negative offsets are outside this tutorial's scope")
    return frames_to_timecode(timecode_to_frames(start_timecode, clock) + offset, clock)


def timecode_to_offset(timecode: str, start_timecode: str, clock: Clock) -> int:
    offset = timecode_to_frames(timecode, clock) - timecode_to_frames(start_timecode, clock)
    if offset < 0:
        raise ClockError("Timecode precedes start; no implicit midnight wrap")
    return offset


def seconds_to_frames(seconds: Any, clock: Clock, rounding: str) -> Dict[str, Any]:
    """Duration -> offset, with exact arithmetic and an explicit rounding policy."""
    value = exact_number(seconds, "seconds")
    if value < 0:
        raise ClockError("Negative durations are rejected, not clamped to zero")
    exact = value * clock.rate
    floor = exact.numerator // exact.denominator
    if rounding == "exact":
        if exact.denominator != 1:
            raise ClockError("Duration is not a whole frame; choose a rounding policy")
        result = floor
    elif rounding == "floor":
        result = floor
    elif rounding == "ceil":
        result = floor + (exact.denominator != 1)
    elif rounding == "nearest":
        # Documented half-up for non-negative durations, not Python's ties-to-even.
        result = floor + ((exact - floor) >= Fraction(1, 2))
    else:
        raise ClockError("rounding must be exact, floor, ceil or nearest")
    return {"rate": str(clock.rate), "seconds_exact": str(value), "frames_exact": str(exact),
            "offset_frames": result, "rounding": rounding,
            "error_seconds_exact": str(Fraction(result, 1) / clock.rate - value)}


def call(obj: Any, method: str, *args: Any) -> Any:
    try:
        fn = getattr(obj, method, None)
        if not callable(fn):
            raise ClockError(method + " is not callable")
        return fn(*args)
    except ClockError:
        raise
    except Exception as exc:
        raise ClockError(method + " failed; original exception retained") from exc


def text(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ClockError(field + " must be non-empty text")
    return value


def context(resolve):
    manager = call(resolve, "GetProjectManager")
    project = call(manager, "GetCurrentProject")
    if project is None:
        raise ClockError("No open project")
    timeline = call(project, "GetCurrentTimeline")
    if timeline is None:
        raise ClockError("No active timeline")
    identity = {"project_id": text(call(project, "GetUniqueId"), "project ID"),
                "timeline_id": text(call(timeline, "GetUniqueId"), "timeline ID")}
    return project, timeline, identity


def observation(resolve, include_playhead=False):
    _, timeline, identity = context(resolve)
    result = dict(identity)
    result.update({"start_frame": integer(call(timeline, "GetStartFrame"), "start_frame"),
                   "end_frame": integer(call(timeline, "GetEndFrame"), "end_frame"),
                   "start_timecode_raw": text(call(timeline, "GetStartTimecode"), "start timecode"),
                   "fps_raw": text(call(timeline, "GetSetting", "timelineFrameRate"), "timeline FPS")})
    if include_playhead:
        page = call(resolve, "GetCurrentPage")
        if page not in ("cut", "edit", "color", "fairlight", "deliver"):
            raise ClockError("Open Cut/Edit/Color/Fairlight/Deliver manually; no page switch is made")
        result["page"] = page
        result["current_timecode_raw"] = text(call(timeline, "GetCurrentTimecode"), "current timecode")
    _, _, after = context(resolve)
    if identity != after:
        raise ClockError("Active target changed while reading")
    return result


def read_clock(resolve, include_playhead=False):
    """Two matching observations, NOT an atomic snapshot or a process timeout."""
    a = observation(resolve, include_playhead)
    b = observation(resolve, include_playhead)
    if a != b:
        raise ClockError("Clock observations differ; stop playback/editing and inspect")
    return dict(a, observation_policy="two_matching_reads_not_atomic")


# These labels are accepted only AFTER the caller explicitly selects the exact rate.
# The report retains the raw label and marks that rate origin as user-declared.
DISPLAY_LABELS = {Fraction(24000, 1001): "23.976", Fraction(30000, 1001): "29.97",
                  Fraction(48000, 1001): "47.952", Fraction(60000, 1001): "59.94"}


def validate_clock(state, clock: Clock):
    raw = state["fps_raw"].strip()
    has_df = raw.upper().endswith(" DF")
    numeric = raw[:-3].strip() if has_df else raw
    actual = exact_number(numeric, "observed FPS label")
    alias = DISPLAY_LABELS.get(clock.rate)
    if actual != clock.rate and (alias is None or actual != Fraction(alias)):
        raise ClockError("Explicit rate conflicts with the observed timeline FPS label")
    if has_df and clock.mode != "df":
        raise ClockError("Observed FPS label says DF but NDF was requested")
    # Absence of DF in the FPS string does not itself prove NDF; inspect start label.
    timecode_to_frames(state["start_timecode_raw"], clock)
    if "current_timecode_raw" in state:
        timecode_to_frames(state["current_timecode_raw"], clock)


def seconds_at_timeline(resolve, seconds, rate, mode, rounding):
    clock = Clock(rate, mode)
    state = read_clock(resolve)
    validate_clock(state, clock)
    result = seconds_to_frames(seconds, clock, rounding)
    result.update({"state": state, "rate_origin": "explicit_user_selection_checked_against_label",
                   "absolute_timeline_frame": state["start_frame"] + result["offset_frames"],
                   "timecode": offset_to_timecode(result["offset_frames"], state["start_timecode_raw"], clock),
                   "inside_conservative_bounds": state["start_frame"] + result["offset_frames"] < state["end_frame"],
                   "operation": "calculation_only_no_seek_or_insert"})
    return result


def make_plan(state, offset, clock):
    validate_clock(state, clock)
    offset = integer(offset, "offset")
    absolute = state["start_frame"] + offset
    if offset < 0 or not state["start_frame"] <= absolute < state["end_frame"]:
        raise ClockError("Target outside conservative [start, end) guard; end endpoint is excluded")
    target_tc = offset_to_timecode(offset, state["start_timecode_raw"], clock)
    plan = {"state": state, "offset_frames": offset, "absolute_timeline_frame": absolute,
            "target_timecode": target_tc, "rate": str(clock.rate), "mode": clock.mode,
            "rate_origin": "explicit_user_selection_checked_against_label"}
    payload = json.dumps(plan, sort_keys=True, ensure_ascii=True, separators=(",", ":"))
    plan["plan_digest"] = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    return plan


def move_playhead(resolve, offset, rate, mode, apply=False, expected_project_id=None,
                  expected_timeline_id=None, expected_plan_digest=None):
    """Plan only unless explicit guards match. Setter dispatched at most once."""
    if type(apply) is not bool:
        raise ClockError("apply must be bool")
    clock = Clock(rate, mode)
    before = read_clock(resolve, include_playhead=True)
    plan = make_plan(before, offset, clock)
    if not apply:
        return dict(plan, status="plan_only", setter_called=False)
    if (not expected_project_id or not expected_timeline_id or not expected_plan_digest or
        expected_project_id != before["project_id"] or expected_timeline_id != before["timeline_id"] or
        expected_plan_digest != plan["plan_digest"]):
        raise ClockError("--apply requires matching reviewed project, timeline and plan digest")
    if read_clock(resolve, include_playhead=True) != before:
        raise ClockError("Target or playhead changed since the plan; no setter called")
    _, timeline, identity = context(resolve)
    if any(identity[k] != before[k] for k in identity):
        raise ClockError("Target changed before dispatch")
    if before["current_timecode_raw"] == plan["target_timecode"]:
        return dict(plan, status="already_at_target", setter_called=False)
    try:
        # The only setter in this file. No fallback, polling, save, playback or retry.
        accepted = call(timeline, "SetCurrentTimecode", plan["target_timecode"])
        after = read_clock(resolve, include_playhead=True)
        static_before = {k: v for k, v in before.items() if k != "current_timecode_raw"}
        static_after = {k: v for k, v in after.items() if k != "current_timecode_raw"}
        if accepted is not True or static_before != static_after:
            raise ClockError("Setter not confirmed or target/clock changed")
        if after["current_timecode_raw"] != plan["target_timecode"]:
            raise ClockError("Read-back does not match requested timecode exactly")
    except Exception as exc:
        raise OutcomeUnknown("Playhead setter dispatched; inspect Resolve before any retry") from exc
    return dict(plan, status="read_back_matched_check_ui", setter_called=True, after=after,
                ui_verified=False)

"""Load the installed Blackmagic module; do not install a similarly named package.

This helper uses standard desktop installation paths or RESOLVE_SCRIPT_API.
It does not enable scripting, start Resolve, open projects or retry API calls.
Validate Python/native-library compatibility against your installed README.txt.
"""
from __future__ import annotations

import importlib.util
import os
from pathlib import Path
import struct
import sys


class ResolveConnectionError(RuntimeError):
    """An explicit environment, module, or application-connection failure."""


def scripting_root() -> Path:
    configured = os.environ.get("RESOLVE_SCRIPT_API")
    if configured is not None:
        if not configured.strip() or configured != configured.strip():
            raise ResolveConnectionError("RESOLVE_SCRIPT_API is empty or has outer whitespace")
        if configured.startswith(('"', "'")) or configured.endswith(('"', "'")):
            raise ResolveConnectionError("Do not store literal outer quotes in RESOLVE_SCRIPT_API")
        root = Path(os.path.expandvars(configured)).expanduser()
    elif sys.platform == "win32":
        root = Path(os.environ.get("PROGRAMDATA", r"C:\ProgramData")) / "Blackmagic Design/DaVinci Resolve/Support/Developer/Scripting"
    elif sys.platform == "darwin":
        root = Path("/Library/Application Support/Blackmagic Design/DaVinci Resolve/Developer/Scripting")
    elif sys.platform.startswith("linux"):
        root = Path("/opt/resolve/Developer/Scripting")
    else:
        raise ResolveConnectionError("Set RESOLVE_SCRIPT_API for this platform/installation")
    if not root.is_absolute():
        raise ResolveConnectionError("RESOLVE_SCRIPT_API must be an absolute Scripting directory")
    return root


def module_file() -> Path:
    # RESOLVE_SCRIPT_API points to Scripting, not its Modules subdirectory.
    return scripting_root() / "Modules" / "DaVinciResolveScript.py"


def load_resolve_module():
    if struct.calcsize("P") * 8 != 64:
        raise ResolveConnectionError("Use a compatible 64-bit Python interpreter")
    source = module_file()
    if not source.is_file():
        raise ResolveConnectionError("Official DaVinciResolveScript.py not found at: " + str(source))
    configured_lib = os.environ.get("RESOLVE_SCRIPT_LIB")
    if configured_lib is not None:
        lib = Path(os.path.expandvars(configured_lib)).expanduser()
        if not lib.is_absolute() or not lib.is_file():
            raise ResolveConnectionError("RESOLVE_SCRIPT_LIB must name the installed native library file")
    name = "DaVinciResolveScript"
    spec = importlib.util.spec_from_file_location(name, source)
    if spec is None or spec.loader is None:
        raise ResolveConnectionError("Cannot create a loader for the official module")
    previous = sys.modules.get(name)
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    try:
        spec.loader.exec_module(module)
        # Blackmagic's wrapper can replace its sys.modules entry with a native module.
        loaded = sys.modules[name]
        if not callable(getattr(loaded, "scriptapp", None)):
            raise ResolveConnectionError("Loaded module does not expose scriptapp")
        return loaded
    except Exception as exc:
        if previous is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = previous
        raise ResolveConnectionError("Official module was found, but loading it failed") from exc


def connect():
    module = load_resolve_module()
    try:
        resolve = module.scriptapp("Resolve")
    except Exception as exc:
        raise ResolveConnectionError("scriptapp failed; inspect the original exception") from exc
    if not resolve:
        raise ResolveConnectionError("No Resolve object: check running Studio, Local access and Python environment")
    return resolve


def project_manager(resolve):
    manager = resolve.GetProjectManager()
    if not manager:
        raise ResolveConnectionError("Resolve connected, but Project Manager is unavailable")
    return manager