diff --git a/.github/workflows/build-installers.yml b/.github/workflows/build-installers.yml index fdfa062..e75d746 100644 --- a/.github/workflows/build-installers.yml +++ b/.github/workflows/build-installers.yml @@ -55,6 +55,9 @@ jobs: - name: Compile Python sources run: python -m compileall openbeat scripts tests + - name: Install Lua for Resolve behavior tests + run: sudo apt-get update && sudo apt-get install -y lua5.1 + - name: Run tests run: python -m unittest discover -s tests -v @@ -68,10 +71,12 @@ jobs: include: - platform: macos runner: macos-latest + runtime_path: dist/openbeat artifact_path: dist/installers/*.pkg artifact_name: OpenBeat-macos-installer - platform: windows runner: windows-latest + runtime_path: dist/openbeat/openbeat.exe artifact_path: dist/installers/*installer.exe artifact_name: OpenBeat-windows-installer @@ -94,6 +99,9 @@ jobs: - name: Build installer bundle run: python scripts/build_installers.py --platform ${{ matrix.platform }} + - name: Verify bundled runtime with real audio + run: python scripts/smoke_test_runtime.py ${{ matrix.runtime_path }} + - name: Upload installer artifact if: github.event_name == 'workflow_dispatch' || github.event_name == 'workflow_call' uses: actions/upload-artifact@v6 diff --git a/.gitignore b/.gitignore index aee88a3..d2f7fe6 100644 --- a/.gitignore +++ b/.gitignore @@ -19,6 +19,7 @@ __pycache__/ *.egg-info/ build/ dist/ +*.spec # Generated OpenBeat outputs *.openbeat-clicks.wav diff --git a/README.md b/README.md index 022bdb0..6593688 100644 --- a/README.md +++ b/README.md @@ -20,9 +20,9 @@ Download the macOS or Windows installer from the [latest GitHub Release](https:/ ## Use It -1. Select the target timeline audio clip. +1. Place the playhead over the target timeline audio clip. 2. Run an OpenBeat action from `Workspace > Scripts > OpenBeat`. -3. If Resolve does not expose the selection cleanly, place the playhead over the clip and run the action again. +3. If audio clips overlap, OpenBeat uses the usable source on the lowest-numbered audio track. Available actions: diff --git a/docs/development.md b/docs/development.md index 2b8be48..fe38f29 100644 --- a/docs/development.md +++ b/docs/development.md @@ -36,6 +36,12 @@ python scripts/build_installers.py --platform windows Generated installer archives are written to `dist/installers/`. +## Run Tests + +Install Lua 5.1 or a later Lua release, then run `python -m unittest discover -s tests -v`. The suite tests timecode conversion, source selection, marker replacement and rollback, and subtitle timing without an open Resolve project. It also sends generated audio through the real CLI and checks both click-track modes. + +After building an installer, test the bundled runtime with `python scripts/smoke_test_runtime.py dist/openbeat` on macOS or `python scripts/smoke_test_runtime.py dist/openbeat/openbeat.exe` on Windows. CI runs these checks on both platform builds. They verify analysis and output generation; placement inside Resolve still needs a manual check. + Installer builds are platform-specific. Build the macOS `.pkg` on macOS and the Windows `.exe` on Windows; PyInstaller does not cross-compile the bundled runtime. ## Release Workflow @@ -59,7 +65,7 @@ The release workflow expects the repository version to already be updated in bot - The non-Studio version of DaVinci Resolve does not reliably allow the kind of `UIManager` popup workflow used by older tools, so OpenBeat ships as no-UI menu scripts - Audio click-track placement onto the timeline is not reliable enough to claim as automatic - Subtitle-track population is still not reliable enough to claim as automatic -- Source resolution still depends on what Resolve exposes for the selected timeline item or the playhead clip +- Source resolution uses the audio clip under the playhead, choosing the lowest-numbered usable audio track when clips overlap ## Verified Behavior diff --git a/docs/guide.md b/docs/guide.md index 9cc630d..9bcb76b 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -4,9 +4,9 @@ OpenBeat works from the active timeline. -1. Select the target timeline audio clip +1. Place the playhead over the target timeline audio clip 2. Run the action from `Workspace > Scripts > OpenBeat` -3. If that does not work, place the playhead over the target clip and run it again +3. If audio clips overlap, OpenBeat uses the usable source on the lowest-numbered audio track ## Choose A Mode @@ -21,9 +21,11 @@ OpenBeat works from the active timeline. Adds markers to the timeline ruler for beats that land inside matching source segments in the active timeline. -- Re-running replaces only the previous OpenBeat timeline markers +- Re-running replaces only the previous OpenBeat timeline markers for that source - User-created timeline markers are left alone - If the same source file appears multiple times, markers can appear across every matching use +- Occupied frames belonging to other sources or user markers are skipped and reported +- Older OpenBeat timeline markers without source information are preserved; remove them manually if you want to replace them ### `Create Clip Markers` @@ -32,6 +34,7 @@ Adds markers to both the Media Pool source clip and matching timeline clip insta - Matching instances are updated from source timing, not only from visible clip position - If the same source appears multiple times, multiple instances can be updated - User-created clip markers are left alone +- A failed marker update attempts to restore the previous markers and reports any restoration failure ### `Create Beat Click Audio` @@ -46,6 +49,8 @@ Renders a click-track WAV next to the source file and imports it into the Media Writes an `.srt` file based on beat-to-beat ranges. - Subtitle files are written relative to the active timeline start +- Cues stay within each matching clip, including trimmed beat ranges, and do not span gaps between clips +- Beat numbers refer to the analyzed source and restart when that source is repeated - OpenBeat can try to import the file and create a subtitle track - Treat the generated `.srt` as the reliable result @@ -64,8 +69,8 @@ If a file already exists, OpenBeat currently overwrites it. ### OpenBeat picked the wrong clip -1. Select the target timeline item -2. If that does not work, place the playhead over the target audio clip +1. Place the playhead inside the target audio clip +2. Check whether a source on a lower-numbered audio track also overlaps the playhead 3. Run the action again ### The click track did not appear on the timeline @@ -93,12 +98,9 @@ Windows: ### Source Selection -OpenBeat checks the timeline in this order: +OpenBeat uses the audio clip under the playhead. If multiple audio tracks overlap, it uses the usable source on the lowest-numbered audio track. Empty tracks and clips without source file paths are skipped. A clip's end belongs to the next clip at a cut. -1. The selected timeline item, if Resolve exposes it and it maps back to a usable audio source -2. The audio clip under the playhead - -It does not use the selected Media Pool item as a fallback. +Resolve's API does not expose selected timeline audio clips. Video and Media Pool selection do not choose OpenBeat's source. ### Shared Behavior diff --git a/docs/install.md b/docs/install.md index f540765..b1d6c8b 100644 --- a/docs/install.md +++ b/docs/install.md @@ -51,10 +51,10 @@ Delete the OpenBeat script and module folders listed above, then restart Resolve ## Verify The Install 1. Open a Resolve project and timeline. -2. Select a timeline audio clip, or place the playhead over an audio clip. +2. Place the playhead over a timeline audio clip. 3. Run `Workspace > Scripts > OpenBeat > Create Timeline Markers (Quantized)`. -If the script runs but cannot identify the clip, place the playhead over the target audio clip and run it again. +If multiple audio clips overlap the playhead, OpenBeat uses the usable source on the lowest-numbered audio track. ## Current Compatibility diff --git a/openbeat.spec b/openbeat.spec deleted file mode 100644 index d5b2afe..0000000 --- a/openbeat.spec +++ /dev/null @@ -1,38 +0,0 @@ -# -*- mode: python ; coding: utf-8 -*- - - -a = Analysis( - ['/Users/parker/Documents/Code/OpenBeat/openbeat/cli.py'], - pathex=[], - binaries=[], - datas=[], - hiddenimports=[], - hookspath=[], - hooksconfig={}, - runtime_hooks=[], - excludes=[], - noarchive=False, - optimize=0, -) -pyz = PYZ(a.pure) - -exe = EXE( - pyz, - a.scripts, - a.binaries, - a.datas, - [], - name='openbeat', - debug=False, - bootloader_ignore_signals=False, - strip=False, - upx=True, - upx_exclude=[], - runtime_tmpdir=None, - console=True, - disable_windowed_traceback=False, - argv_emulation=False, - target_arch=None, - codesign_identity=None, - entitlements_file=None, -) diff --git a/openbeat/analysis.py b/openbeat/analysis.py index 1894a32..85ef71c 100644 --- a/openbeat/analysis.py +++ b/openbeat/analysis.py @@ -5,6 +5,7 @@ import math import os import platform +import tempfile from dataclasses import asdict, dataclass from pathlib import Path from typing import Any @@ -16,7 +17,7 @@ ANALYSIS_TARGET_SR = 22050 -CACHE_SCHEMA_VERSION = "2" +CACHE_SCHEMA_VERSION = "3" @dataclass(slots=True) @@ -30,78 +31,40 @@ class BeatAnalysis: quantized_beats: list[float] -def _normalize_tempo(value: Any) -> float: - if isinstance(value, (list, tuple)): - return float(value[0]) - if isinstance(value, np.ndarray): - return float(value.flatten()[0]) - return float(value) - - -def _median(values: list[float]) -> float: - if not values: - return 0.0 - return float(np.median(np.array(values, dtype=float))) - - def estimate_grid(raw_beats: list[float]) -> tuple[float, float]: + times = np.asarray(raw_beats, dtype=float) + if not np.isfinite(times).all() or (times < 0).any() or (np.diff(times) <= 0).any(): + raise ValueError("Beat times must be finite, nonnegative, and strictly increasing") if len(raw_beats) < 2: if not raw_beats: return 120.0, 0.0 return 120.0, raw_beats[0] - inter_beat_bpms: list[float] = [] - previous = None - for beat in raw_beats: - if previous is not None and beat > previous: - inter_beat_bpms.append(60.0 / (beat - previous)) - previous = beat - - if not inter_beat_bpms: - return 120.0, raw_beats[0] - - bpm_median = _median(inter_beat_bpms) - close_bpms = [bpm for bpm in inter_beat_bpms if abs(bpm - bpm_median) < 5.0] - seed_bpm = int(round(float(np.mean(close_bpms or inter_beat_bpms)))) - - def offset_error(offset: float, period: float) -> float: - capped_distance = 0.1 - half_period = period / 2.0 - total = 0.0 - for beat in raw_beats: - delta = (beat - offset) % period - if delta > half_period: - delta = period - delta - total += min(delta, capped_distance) - return total - - def best_offset_for_bpm(bpm: int) -> tuple[float, float]: - period = 60.0 / bpm - best_offset = 0.0 - best_error = offset_error(0.0, period) - probe = 0.01 - while probe < period: - candidate_error = offset_error(probe, period) - if candidate_error < best_error: - best_error = candidate_error - best_offset = probe - probe += 0.01 - return best_error, best_offset - - winner_bpm = float(seed_bpm) - winner_offset = 0.0 - winner_error = math.inf - - for bpm in range(seed_bpm - 5, seed_bpm + 6): - if bpm <= 0: - continue - error, offset = best_offset_for_bpm(bpm) - if error < winner_error: - winner_error = error - winner_bpm = float(bpm) - winner_offset = offset - - return winner_bpm, winner_offset + intervals = np.diff(times) + seed_period = float(np.median(intervals)) + # Preserve missing beats as gaps in the beat numbers, rather than slowing the grid. + steps = np.maximum(1, np.rint(intervals / seed_period)) + numbers = np.concatenate(([0.0], np.cumsum(steps))) + inliers = np.ones(len(times), dtype=bool) + period = seed_period + offset = float(times[0]) + for _ in range(5): + x, y = numbers[inliers], times[inliers] + centered = x - x.mean() + period = float(np.dot(centered, y - y.mean()) / np.dot(centered, centered)) + offset = float(np.median(y - x * period)) + residuals = times - (offset + numbers * period) + center = float(np.median(residuals)) + spread = float(np.median(np.abs(residuals - center))) + candidates = np.abs(residuals - center) <= max(0.03, 4.0 * 1.4826 * spread) + if candidates.sum() < 2 or np.array_equal(candidates, inliers): + break + inliers = candidates + + phase = offset % period + if math.isclose(phase, period, abs_tol=1e-9): + phase = 0.0 + return 60.0 / period, phase def quantized_beats(tempo_bpm: float, offset_seconds: float, duration_seconds: float) -> list[float]: @@ -120,7 +83,7 @@ def analyze_audio(audio_path: str, target_sr: int = ANALYSIS_TARGET_SR) -> BeatA audio_path = str(Path(audio_path).expanduser().resolve()) signal, sr = librosa.load(audio_path, sr=target_sr, mono=True) duration_seconds = float(librosa.get_duration(y=signal, sr=sr)) - tempo, beat_frames = librosa.beat.beat_track(y=signal, sr=sr, trim=False) + _, beat_frames = librosa.beat.beat_track(y=signal, sr=sr, trim=False) beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist() beat_times = [float(value) for value in beat_times] @@ -133,9 +96,9 @@ def analyze_audio(audio_path: str, target_sr: int = ANALYSIS_TARGET_SR) -> BeatA duration_seconds=duration_seconds, sample_rate=sr, raw_beats=beat_times, - tempo_bpm=_normalize_tempo(tempo_bpm), + tempo_bpm=tempo_bpm, grid_offset_seconds=grid_offset, - quantized_beats=quantized_beats(_normalize_tempo(tempo_bpm), grid_offset, duration_seconds), + quantized_beats=quantized_beats(tempo_bpm, grid_offset, duration_seconds), ) @@ -181,9 +144,68 @@ def cache_key(audio_path: str, target_sr: int = ANALYSIS_TARGET_SR) -> str: def cache_path(audio_path: str, cache_dir: Path | None = None, target_sr: int = ANALYSIS_TARGET_SR) -> Path: + key = cache_key(audio_path, target_sr=target_sr) cache_dir = cache_dir or default_cache_dir() cache_dir.mkdir(parents=True, exist_ok=True) - return cache_dir / f"{cache_key(audio_path, target_sr=target_sr)}.json" + return cache_dir / f"{key}.json" + + +def _cached_analysis(payload: Any, audio_path: str, target_sr: int) -> BeatAnalysis: + if not isinstance(payload, dict): + raise ValueError("Invalid analysis cache object") + analysis = BeatAnalysis(**payload) + + def finite_number(value: Any) -> bool: + if not isinstance(value, (int, float)) or isinstance(value, bool): + return False + try: + return math.isfinite(value) + except OverflowError: + return False + + if analysis.audio_path != audio_path or type(analysis.sample_rate) is not int or analysis.sample_rate != target_sr: + raise ValueError("Analysis cache does not match the source") + if not finite_number(analysis.duration_seconds) or analysis.duration_seconds <= 0: + raise ValueError("Invalid cached duration") + if not finite_number(analysis.tempo_bpm) or analysis.tempo_bpm <= 0: + raise ValueError("Invalid cached tempo") + if not finite_number(analysis.grid_offset_seconds) or not 0 <= analysis.grid_offset_seconds <= analysis.duration_seconds: + raise ValueError("Invalid cached grid offset") + for beats in (analysis.raw_beats, analysis.quantized_beats): + if not isinstance(beats, list) or not beats: + raise ValueError("Invalid cached beats") + if not all(finite_number(beat) and 0 <= beat <= analysis.duration_seconds for beat in beats): + raise ValueError("Invalid cached beat time") + if any(left >= right for left, right in zip(beats, beats[1:])): + raise ValueError("Cached beats must be strictly increasing") + period = 60.0 / analysis.tempo_bpm + if not math.isclose(analysis.quantized_beats[0], analysis.grid_offset_seconds, abs_tol=1e-7): + raise ValueError("Cached grid does not match its offset") + if any(not math.isclose(right - left, period, abs_tol=1e-7) + for left, right in zip(analysis.quantized_beats, analysis.quantized_beats[1:])): + raise ValueError("Cached grid does not match its tempo") + if analysis.quantized_beats[-1] + period < analysis.duration_seconds - 1e-7: + raise ValueError("Cached grid is incomplete") + return analysis + + +def _write_cache(target: Path, analysis: BeatAnalysis) -> None: + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=target.parent, prefix=f".{target.stem}-", suffix=".tmp", delete=False + ) as handle: + temporary = Path(handle.name) + json.dump(asdict(analysis), handle, allow_nan=False) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, target) + finally: + if temporary is not None: + try: + temporary.unlink(missing_ok=True) + except OSError: + pass def load_or_analyze( @@ -191,12 +213,24 @@ def load_or_analyze( cache_dir: Path | None = None, target_sr: int = ANALYSIS_TARGET_SR, ) -> BeatAnalysis: - target = cache_path(audio_path, cache_dir, target_sr=target_sr) - if target.exists(): - return BeatAnalysis(**json.loads(target.read_text())) + audio_path = str(Path(audio_path).expanduser().resolve()) + # Source errors should still be reported; cache errors should not prevent analysis. + key = cache_key(audio_path, target_sr=target_sr) + target = None + try: + directory = cache_dir or default_cache_dir() + directory.mkdir(parents=True, exist_ok=True) + target = directory / f"{key}.json" + return _cached_analysis(json.loads(target.read_text(encoding="utf-8")), audio_path, target_sr) + except (OSError, UnicodeError, ValueError, TypeError): + pass analysis = analyze_audio(audio_path, target_sr=target_sr) - target.write_text(json.dumps(asdict(analysis), indent=2)) + if target is not None: + try: + _write_cache(target, analysis) + except OSError: + pass return analysis @@ -206,9 +240,12 @@ def to_lua(value: Any) -> str: if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): + if not math.isfinite(value): + raise ValueError("Lua numbers must be finite") return repr(value) if isinstance(value, str): - escaped = value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") + escapes = { "\\": "\\\\", '"': '\\"', "\n": "\\n", "\r": "\\r", "\t": "\\t" } + escaped = "".join(escapes.get(char, f"\\{ord(char):03d}" if ord(char) < 32 else char) for char in value) return f'"{escaped}"' if isinstance(value, list): inner = ", ".join(to_lua(item) for item in value) @@ -223,4 +260,4 @@ def to_lua(value: Any) -> str: def write_lua_analysis(analysis: BeatAnalysis, output_path: str) -> None: payload = asdict(analysis) - Path(output_path).write_text("return " + to_lua(payload) + "\n") + Path(output_path).write_text("return " + to_lua(payload) + "\n", encoding="utf-8") diff --git a/openbeat/cli.py b/openbeat/cli.py index c2e1b62..79041be 100644 --- a/openbeat/cli.py +++ b/openbeat/cli.py @@ -14,7 +14,7 @@ def analyze_command(args: argparse.Namespace) -> int: if args.format == "json": payload = json.dumps(asdict(analysis), indent=2) if args.output: - Path(args.output).write_text(payload + "\n") + Path(args.output).write_text(payload + "\n", encoding="utf-8") else: print(payload) return 0 diff --git a/resolve/Fusion/Modules/OpenBeat/OpenBeatCommon.lua b/resolve/Fusion/Modules/OpenBeat/OpenBeatCommon.lua index 9f09f20..431ffbb 100644 --- a/resolve/Fusion/Modules/OpenBeat/OpenBeatCommon.lua +++ b/resolve/Fusion/Modules/OpenBeat/OpenBeatCommon.lua @@ -69,6 +69,10 @@ local function load_local_config() end local local_config = load_local_config() +local Timing = dofile(join_path(script_dir(), "OpenBeatTiming.lua")) +local Source = dofile(join_path(script_dir(), "OpenBeatSource.lua")) +local Subtitles = dofile(join_path(script_dir(), "OpenBeatSubtitles.lua")) +local Markers = dofile(join_path(script_dir(), "OpenBeatMarkers.lua")) local function repo_root() if local_config.repo_root then @@ -152,15 +156,16 @@ end local function command_prefix() local bin = python_bin() - local lowered = string.lower(bin) - if lowered:match("python%.exe$") or lowered:match("python$") then + local name = string.lower(bin):match("[^/\\]+$") or "" + name = name:gsub("%.exe$", "") + if name == "python" or name:match("^python%d[%d%.]*$") then return shell_quote(bin) .. " -m openbeat.cli" end return shell_quote(bin) end local function project_context() - local resolve_app = resolve or app:GetResolve() + local resolve_app = resolve or (app and app:GetResolve()) if not resolve_app then error("Resolve API unavailable") end @@ -176,24 +181,9 @@ local function project_context() error("Open a timeline before running OpenBeat.") end - local fps = timeline:GetSetting("timelineFrameRate") or project:GetSetting("timelineFrameRate") - if type(fps) == "table" then - fps = fps.timelineFrameRate or fps.FPS - end - fps = tonumber(fps) - if not fps or fps <= 0 then - error("Could not determine timeline frame rate.") - end - - return resolve_app, project, timeline, fps -end - -local function parse_timecode(tc, fps) - local h, m, s, f = tc:match("(%d+):(%d+):(%d+):(%d+)") - if not h then - error("Unsupported timecode: " .. tostring(tc)) - end - return (((tonumber(h) * 60) + tonumber(m)) * 60 + tonumber(s)) * fps + tonumber(f) + local setting = timeline:GetSetting("timelineFrameRate") or project:GetSetting("timelineFrameRate") + local fps, drop_frame = Timing.frame_rate(setting) + return resolve_app, project, timeline, fps, drop_frame end local function get_clip_property_value(clip, key) @@ -212,108 +202,6 @@ local function numeric_clip_property(clip, key, fallback) return fallback end -local function media_pool_path(item) - local path = get_clip_property_value(item, "File Path") - if not path or path == "" then - error("Selected timeline item has no file path") - end - return path -end - -local function clip_contains_frame(clip, frame) - return clip:GetStart() <= frame and frame <= clip:GetEnd() -end - -local function all_audio_segments_for_path(timeline, source_path) - local matches = {} - local track_count = timeline:GetTrackCount("audio") - for track_index = 1, track_count do - local items = timeline:GetItemListInTrack("audio", track_index) - for _, item in ipairs(items) do - local media = item:GetMediaPoolItem() - if media then - local path = media_pool_path(media) - if path == source_path then - table.insert(matches, { - track_index = track_index, - timeline_item = item, - media_pool_item = media, - start_frame = item:GetStart(), - end_frame = item:GetEnd(), - left_offset = item:GetLeftOffset() or 0, - }) - end - end - end - end - return matches -end - -local function source_at_playhead(timeline, fps) - local playhead_frame = parse_timecode(timeline:GetCurrentTimecode(), fps) - local track_count = timeline:GetTrackCount("audio") - for track_index = 1, track_count do - local items = timeline:GetItemListInTrack("audio", track_index) - for _, item in ipairs(items) do - if clip_contains_frame(item, playhead_frame) then - local media = item:GetMediaPoolItem() - if media then - local path = media_pool_path(media) - return path, item, media - end - end - end - end - error("Place the playhead over a timeline audio clip before running OpenBeat.") -end - -local function source_from_selected_item(timeline) - local current_item = timeline.GetCurrentVideoItem and timeline:GetCurrentVideoItem() or nil - if not current_item then - return nil - end - - local media = current_item:GetMediaPoolItem() - if not media then - return nil - end - - local path = media_pool_path(media) - local segments = all_audio_segments_for_path(timeline, path) - if #segments == 0 then - return nil - end - - return path, current_item, media -end - -local function selected_or_playhead_source(project, timeline, fps) - local path, item, media = source_from_selected_item(timeline) - if path and media then - return path, item, media - end - - return source_at_playhead(timeline, fps) -end - -local function colors() - return { "Green", "Blue", "Yellow", "Purple", "Cyan", "Pink", "Red", "Fuchsia" } -end - -local function remove_openbeat_markers(holder) - local markers = holder:GetMarkers() or {} - for frame_id, _ in pairs(markers) do - local custom = holder.GetMarkerCustomData and holder:GetMarkerCustomData(frame_id) or "" - if type(custom) == "string" and custom:match("^OpenBeat:") then - holder:DeleteMarkerAtFrame(frame_id) - end - end -end - -local function round(value) - return math.floor(value + 0.5) -end - local function analyze_source(source_path) local output = temp_path("openbeat_analysis", ".lua") local final_command = command_prefix() @@ -335,65 +223,46 @@ local function beats_for_mode(analysis, mode) end local function create_timeline_markers(mode) - local _, project, timeline, fps = project_context() - local source_path = selected_or_playhead_source(project, timeline, fps) + local _, _, timeline, fps, drop_frame = project_context() + local source_path = Source.at_playhead(timeline, fps, drop_frame) local analysis = analyze_source(source_path) local beats = beats_for_mode(analysis, mode) - local segments = all_audio_segments_for_path(timeline, source_path) - local marker_colors = colors() - - remove_openbeat_markers(timeline) - - local timeline_start = timeline:GetStartFrame() - local created = 0 - for _, segment in ipairs(segments) do - local source_zero_frame = segment.start_frame - segment.left_offset - timeline_start - for index, beat in ipairs(beats) do - local frame = round(beat * fps) + source_zero_frame - local segment_start = segment.start_frame - timeline_start - local segment_end = segment.end_frame - timeline_start - if frame >= segment_start and frame <= segment_end then - local color = marker_colors[((index - 1) % #marker_colors) + 1] - timeline:AddMarker(frame, color, "OpenBeat " .. index, "Created by OpenBeat", 1.0, "OpenBeat:timeline") - created = created + 1 - end - end - end - - print(string.format("OpenBeat created %d timeline markers for %s", created, source_path)) - log(string.format("Created %d timeline markers for %s", created, source_path)) + local segments = Source.segments_for_path(timeline, source_path) + local created, skipped = Markers.replace_many({ { + holder = timeline, + plan = Markers.timeline_plan(beats, segments, fps, timeline:GetStartFrame()), + custom_data = Markers.custom_data("timeline", source_path), + } }) + local message = string.format("OpenBeat created %d timeline markers for %s; skipped %d occupied frames", created, source_path, skipped) + print(message) + log(message) end local function create_clip_markers(mode) - local _, project, timeline, fps = project_context() - local source_path, _, media = selected_or_playhead_source(project, timeline, fps) + local _, _, timeline, fps, drop_frame = project_context() + local source_path, _, media = Source.at_playhead(timeline, fps, drop_frame) local analysis = analyze_source(source_path) local beats = beats_for_mode(analysis, mode) - local segments = all_audio_segments_for_path(timeline, source_path) - local marker_colors = colors() + local segments = Source.segments_for_path(timeline, source_path) local clip_fps = numeric_clip_property(media, "FPS", fps) - - remove_openbeat_markers(media) + local groups = { { + holder = media, + plan = Markers.source_plan(beats, clip_fps, analysis.duration_seconds), + custom_data = Markers.custom_data("clip", source_path), + legacy_custom_data = "OpenBeat:clip", + } } for _, segment in ipairs(segments) do - remove_openbeat_markers(segment.timeline_item) - end - - for index, beat in ipairs(beats) do - local frame = round(beat * clip_fps) - local color = marker_colors[((index - 1) % #marker_colors) + 1] - media:AddMarker(frame, color, "OpenBeat " .. index, "Created by OpenBeat", 1.0, "OpenBeat:clip") - for _, segment in ipairs(segments) do - local source_start = segment.timeline_item:GetSourceStartFrame() - local source_end = segment.timeline_item:GetSourceEndFrame() - if source_start and source_end and frame >= source_start and frame <= source_end then - local clip_offset = frame - source_start - segment.timeline_item:AddMarker(clip_offset, color, "OpenBeat " .. index, "Created by OpenBeat", 1.0, "OpenBeat:clip-item") - end - end + table.insert(groups, { + holder = segment.timeline_item, + plan = Markers.clip_plan(beats, segment, fps, clip_fps), + custom_data = Markers.custom_data("clip-item", source_path), + legacy_custom_data = "OpenBeat:clip-item", + }) end - - print(string.format("OpenBeat created %d clip markers for %s", #beats, source_path)) - log(string.format("Created %d clip markers for %s", #beats, source_path)) + local created, skipped = Markers.replace_many(groups) + local message = string.format("OpenBeat created %d source and timeline clip markers for %s; skipped %d occupied frames", created, source_path, skipped) + print(message) + log(message) end local function output_path_for(source_path, suffix, extension) @@ -421,8 +290,8 @@ local function find_media_pool_item_by_path(resolve_app, source_path) end local function create_click_audio(mode) - local resolve_app, project, timeline, fps = project_context() - local source_path = selected_or_playhead_source(project, timeline, fps) + local resolve_app, project, timeline, fps, drop_frame = project_context() + local source_path = Source.at_playhead(timeline, fps, drop_frame) local click_path = render_click_track(source_path, mode) local click_item = find_media_pool_item_by_path(resolve_app, click_path) if not click_item then @@ -450,15 +319,22 @@ end local function capture_track_enabled(timeline, track_type) local states = {} for track_index = 1, timeline:GetTrackCount(track_type) do - states[track_index] = timeline:GetIsTrackEnabled(track_type, track_index) + local enabled = timeline:GetIsTrackEnabled(track_type, track_index) + if type(enabled) ~= "boolean" then + error("Could not read the " .. track_type .. " track state before subtitle placement.") + end + states[track_index] = enabled end return states end local function restore_track_enabled(timeline, track_type, states) + local restored = true for track_index, enabled in pairs(states) do - timeline:SetTrackEnable(track_type, track_index, enabled) + local ok, result = pcall(timeline.SetTrackEnable, timeline, track_type, track_index, enabled) + if not ok or not result then restored = false end end + return restored end local function import_subtitles_to_timeline(resolve_app, project, timeline, subtitle_path) @@ -479,14 +355,7 @@ local function import_subtitles_to_timeline(resolve_app, project, timeline, subt local video_states = capture_track_enabled(timeline, "video") local subtitle_states = capture_track_enabled(timeline, "subtitle") - for track_index = 1, timeline:GetTrackCount("video") do - timeline:SetTrackEnable("video", track_index, false) - end - for track_index = 1, timeline:GetTrackCount("subtitle") do - timeline:SetTrackEnable("subtitle", track_index, track_index == target_track) - end - - local before_count = track_item_count(timeline, "subtitle", target_track) + local before_count = 0 local clip_info = { mediaPoolItem = subtitle_item, startFrame = 0, @@ -496,12 +365,26 @@ local function import_subtitles_to_timeline(resolve_app, project, timeline, subt local appended = nil local after_count = before_count local append_ok, append_error = pcall(function() + for track_index = 1, timeline:GetTrackCount("video") do + if not timeline:SetTrackEnable("video", track_index, false) then + error("Could not disable a video track for subtitle placement.") + end + end + for track_index = 1, timeline:GetTrackCount("subtitle") do + if not timeline:SetTrackEnable("subtitle", track_index, track_index == target_track) then + error("Could not select the subtitle track for placement.") + end + end + before_count = track_item_count(timeline, "subtitle", target_track) appended = project:GetMediaPool():AppendToTimeline({ clip_info }) after_count = track_item_count(timeline, "subtitle", target_track) end) - restore_track_enabled(timeline, "video", video_states) - restore_track_enabled(timeline, "subtitle", subtitle_states) + local video_restored = restore_track_enabled(timeline, "video", video_states) + local subtitles_restored = restore_track_enabled(timeline, "subtitle", subtitle_states) + if not video_restored or not subtitles_restored then + return false, "Resolve could not restore all track-enable states after subtitle placement. Check the video and subtitle tracks." + end if not append_ok then return false, "Resolve failed while placing the subtitle file on the timeline: " .. tostring(append_error) @@ -515,86 +398,24 @@ local function import_subtitles_to_timeline(resolve_app, project, timeline, subt return true, nil end -local function format_srt_time(seconds) - local total_ms = math.floor(seconds * 1000) - local ms = total_ms % 1000 - local total_s = math.floor(total_ms / 1000) - local s = total_s % 60 - local total_m = math.floor(total_s / 60) - local m = total_m % 60 - local h = math.floor(total_m / 60) - return string.format("%02d:%02d:%02d,%03d", h, m, s, ms) -end - -local function unique_sorted(list) - table.sort(list) - local result = {} - local last = nil - for _, value in ipairs(list) do - if last == nil or math.abs(value - last) > 0.0005 then - table.insert(result, value) - last = value - end - end - return result -end - local function export_subtitles(mode) - local resolve_app, project, timeline, fps = project_context() - local source_path = selected_or_playhead_source(project, timeline, fps) + local resolve_app, project, timeline, fps, drop_frame = project_context() + local source_path = Source.at_playhead(timeline, fps, drop_frame) local analysis = analyze_source(source_path) local beats = beats_for_mode(analysis, mode) - local segments = all_audio_segments_for_path(timeline, source_path) - local timeline_start = timeline:GetStartFrame() - local subtitle_beats = {} - - for _, segment in ipairs(segments) do - local source_zero = segment.start_frame - segment.left_offset - timeline_start - local rel_start = segment.start_frame - timeline_start - local rel_end = segment.end_frame - timeline_start - for _, beat in ipairs(beats) do - local beat_time = beat + (source_zero / fps) - if beat_time >= (rel_start / fps) and beat_time <= (rel_end / fps) then - table.insert(subtitle_beats, beat_time) - end - end - end - - subtitle_beats = unique_sorted(subtitle_beats) - local output_lines = {} - local counter = 1 - local absolute_offset = timeline_start / fps - if subtitle_beats[1] and subtitle_beats[1] > 0 then - table.insert(output_lines, tostring(counter)) - table.insert(output_lines, format_srt_time(absolute_offset) .. " --> " .. format_srt_time(absolute_offset + subtitle_beats[1])) - table.insert(output_lines, "before first beat") - table.insert(output_lines, "") - counter = counter + 1 - end - - local beat_number = 1 - local previous = nil - for _, beat_time in ipairs(subtitle_beats) do - if previous ~= nil then - table.insert(output_lines, tostring(counter)) - table.insert(output_lines, format_srt_time(absolute_offset + previous) .. " --> " .. format_srt_time(absolute_offset + beat_time)) - table.insert(output_lines, "beat " .. beat_number) - table.insert(output_lines, ((beat_number - 1) % 4 + 1) .. "/4") - table.insert(output_lines, ((beat_number - 1) % 8 + 1) .. "/8") - table.insert(output_lines, ((beat_number - 1) % 16 + 1) .. "/16") - table.insert(output_lines, "") - counter = counter + 1 - beat_number = beat_number + 1 - end - previous = beat_time - end + local segments = Source.segments_for_path(timeline, source_path) + local cues = Subtitles.plan(beats, segments, fps, timeline:GetStartFrame()) local suffix = mode == "raw" and ".openbeat-raw" or ".openbeat" local output = output_path_for(source_path, suffix, ".srt") local handle = assert(io.open(output, "w")) - handle:write(table.concat(output_lines, "\n")) + handle:write(Subtitles.to_srt(cues)) handle:close() - local imported, reason = import_subtitles_to_timeline(resolve_app, project, timeline, output) + local import_ok, imported, reason = pcall(import_subtitles_to_timeline, resolve_app, project, timeline, output) + if not import_ok then + reason = "Resolve could not place the generated subtitle file: " .. tostring(imported) + imported = false + end if imported then print("OpenBeat wrote subtitles to " .. output .. " and placed them on a subtitle track.") log("Exported subtitles to " .. output .. " and placed them on timeline") @@ -605,6 +426,9 @@ local function export_subtitles(mode) end function OpenBeat.run(action, mode) + if mode ~= "raw" and mode ~= "quantized" then + error("Unknown OpenBeat mode: " .. tostring(mode)) + end math.randomseed(os.time()) local ok, result = pcall(function() if action == "timeline_markers" then diff --git a/resolve/Fusion/Modules/OpenBeat/OpenBeatMarkers.lua b/resolve/Fusion/Modules/OpenBeat/OpenBeatMarkers.lua new file mode 100644 index 0000000..d528d9b --- /dev/null +++ b/resolve/Fusion/Modules/OpenBeat/OpenBeatMarkers.lua @@ -0,0 +1,141 @@ +local Markers = {} +local module_dir = debug.getinfo(1, "S").source:sub(2):match("^(.*)[/\\][^/\\]+$") +local Timing = dofile(module_dir .. "/OpenBeatTiming.lua") + +local function round(value) + return math.floor(value + 0.5) +end + +local function sorted_plan(by_frame) + local plan = {} + for frame, index in pairs(by_frame) do + table.insert(plan, { frame = frame, index = index }) + end + table.sort(plan, function(a, b) return a.frame < b.frame end) + return plan +end + +function Markers.timeline_plan(beats, segments, fps, timeline_start) + local by_frame = {} + for _, segment in ipairs(segments) do + local source_zero = Timing.source_zero_seconds(segment, fps, timeline_start) * fps + local start_frame, end_frame = segment.start_frame - timeline_start, segment.end_frame - timeline_start + for index, beat in ipairs(beats) do + local position = beat * fps + source_zero + local frame = round(position) + if position >= start_frame and position < end_frame and frame >= start_frame and frame < end_frame + and not by_frame[frame] then + by_frame[frame] = index + end + end + end + return sorted_plan(by_frame) +end + +function Markers.source_plan(beats, clip_fps, duration) + local by_frame = {} + for index, beat in ipairs(beats) do + local frame = round(beat * clip_fps) + if beat >= 0 and beat < duration and frame < duration * clip_fps and not by_frame[frame] then + by_frame[frame] = index + end + end + return sorted_plan(by_frame) +end + +function Markers.clip_plan(beats, segment, fps, clip_fps) + local source_start = Timing.source_start_seconds(segment, fps, clip_fps) + local duration_frames = segment.end_frame - segment.start_frame + local by_frame = {} + for index, beat in ipairs(beats) do + local position = (beat - source_start) * fps + local frame = round(position) + if position >= 0 and position < duration_frames and frame < duration_frames and not by_frame[frame] then + by_frame[frame] = index + end + end + return sorted_plan(by_frame) +end + +function Markers.custom_data(kind, source_path) + return "OpenBeat:" .. kind .. ":" .. source_path +end + +local colors = { "Green", "Blue", "Yellow", "Purple", "Cyan", "Pink", "Red", "Fuchsia" } + +local function add(holder, frame, marker) + return holder:AddMarker(frame, marker.color, marker.name, marker.note, marker.duration, marker.customData) +end + +function Markers.replace_many(groups) + local prepared, skipped = {}, 0 + -- Read and prepare every holder before making any destructive API calls. + for _, group in ipairs(groups) do + local existing = group.holder:GetMarkers() or {} + local owned, planned = {}, {} + for frame, marker in pairs(existing) do + local custom = marker.customData + if custom == nil and group.holder.GetMarkerCustomData then + custom = group.holder:GetMarkerCustomData(frame) + end + if custom == group.custom_data or (group.legacy_custom_data and custom == group.legacy_custom_data) then + owned[frame] = { + color = marker.color, name = marker.name, note = marker.note, + duration = marker.duration, customData = custom, + } + end + end + for _, entry in ipairs(group.plan) do + if existing[entry.frame] and not owned[entry.frame] then + skipped = skipped + 1 + else + table.insert(planned, { + frame = entry.frame, + marker = { + color = colors[((entry.index - 1) % #colors) + 1], name = "OpenBeat " .. entry.index, + note = "Created by OpenBeat", duration = 1.0, customData = group.custom_data, + }, + }) + end + end + table.insert(prepared, { holder = group.holder, owned = owned, planned = planned, deleted = {}, added = {} }) + end + + local created = 0 + local ok, reason = pcall(function() + for _, group in ipairs(prepared) do + for frame, marker in pairs(group.owned) do + if not group.holder:DeleteMarkerAtFrame(frame) then + error("Resolve rejected marker deletion at frame " .. frame) + end + group.deleted[frame] = marker + end + for _, entry in ipairs(group.planned) do + if not add(group.holder, entry.frame, entry.marker) then + error("Resolve rejected marker creation at frame " .. entry.frame) + end + table.insert(group.added, entry.frame) + created = created + 1 + end + end + end) + if not ok then + local rollback_ok = true + for index = #prepared, 1, -1 do + local group = prepared[index] + for _, frame in ipairs(group.added) do + local call_ok, deleted = pcall(group.holder.DeleteMarkerAtFrame, group.holder, frame) + rollback_ok = call_ok and deleted and rollback_ok + end + for frame, marker in pairs(group.deleted) do + local call_ok, restored = pcall(add, group.holder, frame, marker) + rollback_ok = call_ok and restored and rollback_ok + end + end + local status = rollback_ok and "Previous markers restored." or "Some previous markers could not be restored." + error("OpenBeat marker update failed. " .. status .. " " .. tostring(reason)) + end + return created, skipped +end + +return Markers diff --git a/resolve/Fusion/Modules/OpenBeat/OpenBeatSource.lua b/resolve/Fusion/Modules/OpenBeat/OpenBeatSource.lua new file mode 100644 index 0000000..b1d3a4b --- /dev/null +++ b/resolve/Fusion/Modules/OpenBeat/OpenBeatSource.lua @@ -0,0 +1,61 @@ +local Source = {} + +local module_dir = debug.getinfo(1, "S").source:sub(2):match("^(.*)[/\\][^/\\]+$") +local Timing = dofile(module_dir .. "/OpenBeatTiming.lua") + +local function media_path(media) + if not media then + return nil + end + local path = media:GetClipProperty("File Path") + if type(path) == "table" then + path = path["File Path"] + end + if type(path) == "string" and path ~= "" then + return path + end + return nil +end + +function Source.segments_for_path(timeline, source_path) + local matches = {} + for track_index = 1, timeline:GetTrackCount("audio") do + for _, item in ipairs(timeline:GetItemListInTrack("audio", track_index) or {}) do + local media = item:GetMediaPoolItem() + if media_path(media) == source_path then + local source_fps = media:GetClipProperty("FPS") + if type(source_fps) == "table" then source_fps = source_fps.FPS end + table.insert(matches, { + track_index = track_index, + timeline_item = item, + media_pool_item = media, + start_frame = item:GetStart(), + end_frame = item:GetEnd(), + left_offset = item:GetLeftOffset() or 0, + source_start_frame = item.GetSourceStartFrame and item:GetSourceStartFrame() or nil, + source_fps = tonumber(source_fps), + }) + end + end + end + return matches +end + +function Source.at_playhead(timeline, fps, drop_frame) + local frame = Timing.timecode_to_frame(timeline:GetCurrentTimecode(), fps, drop_frame) + -- Resolve does not expose the selected timeline audio item. Track order is explicit. + for track_index = 1, timeline:GetTrackCount("audio") do + for _, item in ipairs(timeline:GetItemListInTrack("audio", track_index) or {}) do + if item:GetStart() <= frame and frame < item:GetEnd() then + local media = item:GetMediaPoolItem() + local path = media_path(media) + if path then + return path, item, media + end + end + end + end + error("Place the playhead over a timeline audio clip with a source file before running OpenBeat.") +end + +return Source diff --git a/resolve/Fusion/Modules/OpenBeat/OpenBeatSubtitles.lua b/resolve/Fusion/Modules/OpenBeat/OpenBeatSubtitles.lua new file mode 100644 index 0000000..c08684f --- /dev/null +++ b/resolve/Fusion/Modules/OpenBeat/OpenBeatSubtitles.lua @@ -0,0 +1,72 @@ +local Subtitles = {} +local module_dir = debug.getinfo(1, "S").source:sub(2):match("^(.*)[/\\][^/\\]+$") +local Timing = dofile(module_dir .. "/OpenBeatTiming.lua") + +function Subtitles.plan(beats, segments, fps, timeline_start) + local cues = {} + local function add(start_time, end_time, beat_number) + start_time = math.max(0, start_time) + if end_time - start_time >= 0.001 then + table.insert(cues, { start_time = start_time, end_time = end_time, beat_number = beat_number }) + end + end + for _, segment in ipairs(segments) do + local segment_start = (segment.start_frame - timeline_start) / fps + local segment_end = (segment.end_frame - timeline_start) / fps + local source_zero = Timing.source_zero_seconds(segment, fps, timeline_start) + if beats[1] then + add(segment_start, math.min(segment_end, source_zero + beats[1]), nil) + end + for index, beat in ipairs(beats) do + local start_time = math.max(segment_start, source_zero + beat) + local end_time = math.min(segment_end, beats[index + 1] and source_zero + beats[index + 1] or segment_end) + add(start_time, end_time, index) + end + end + table.sort(cues, function(a, b) + if a.start_time ~= b.start_time then return a.start_time < b.start_time end + if a.end_time ~= b.end_time then return a.end_time < b.end_time end + return (a.beat_number or 0) < (b.beat_number or 0) + end) + local unique = {} + for _, cue in ipairs(cues) do + local previous = unique[#unique] + if not previous or math.abs(cue.start_time - previous.start_time) > 0.0005 + or math.abs(cue.end_time - previous.end_time) > 0.0005 + or cue.beat_number ~= previous.beat_number then + table.insert(unique, cue) + end + end + return unique +end + +local function format_time(seconds) + local total_ms = math.floor(seconds * 1000 + 0.5) + local ms = total_ms % 1000 + local total_s = math.floor(total_ms / 1000) + local s = total_s % 60 + local total_m = math.floor(total_s / 60) + local m = total_m % 60 + local h = math.floor(total_m / 60) + return string.format("%02d:%02d:%02d,%03d", h, m, s, ms) +end + +function Subtitles.to_srt(cues) + local lines = {} + for index, cue in ipairs(cues) do + table.insert(lines, tostring(index)) + table.insert(lines, format_time(cue.start_time) .. " --> " .. format_time(cue.end_time)) + if cue.beat_number then + table.insert(lines, "beat " .. cue.beat_number) + for _, count in ipairs({ 4, 8, 16 }) do + table.insert(lines, ((cue.beat_number - 1) % count + 1) .. "/" .. count) + end + else + table.insert(lines, "before first beat") + end + table.insert(lines, "") + end + return table.concat(lines, "\n") .. "\n" +end + +return Subtitles diff --git a/resolve/Fusion/Modules/OpenBeat/OpenBeatTiming.lua b/resolve/Fusion/Modules/OpenBeat/OpenBeatTiming.lua new file mode 100644 index 0000000..cc265ab --- /dev/null +++ b/resolve/Fusion/Modules/OpenBeat/OpenBeatTiming.lua @@ -0,0 +1,55 @@ +local Timing = {} + +function Timing.frame_rate(setting) + if type(setting) == "table" then setting = setting.timelineFrameRate or setting.FPS end + local text = tostring(setting) + local drop_frame = text:match("%s+[dD][fF]%s*$") ~= nil + local numeric = text:gsub("%s+[dD][fF]%s*$", "") + local fps = tonumber(numeric) + if not fps or fps <= 0 or fps == math.huge then + error("Could not determine timeline frame rate.") + end + return fps, drop_frame +end + +function Timing.timecode_to_frame(tc, fps, drop_frame) + local h, m, s, separator, f = tostring(tc):match("^(%d%d):(%d%d):(%d%d)([:;])(%d+)$") + if not h then + error("Unsupported timecode: " .. tostring(tc)) + end + h, m, s, f = tonumber(h), tonumber(m), tonumber(s), tonumber(f) + local nominal_fps = math.floor(fps + 0.5) + if h > 23 or m > 59 or s > 59 or f >= nominal_fps then + error("Invalid timecode: " .. tc) + end + + -- Timecode labels count nominal frames, even when playback uses a fractional rate. + local frame = ((h * 60 + m) * 60 + s) * nominal_fps + f + if separator == ";" or drop_frame then + if (nominal_fps ~= 30 and nominal_fps ~= 60) + or math.abs(fps - nominal_fps * 1000 / 1001) > 0.01 then + error("Unsupported drop-frame rate: " .. tostring(fps)) + end + local dropped_per_minute = nominal_fps / 15 + if m % 10 ~= 0 and s == 0 and f < dropped_per_minute then + error("Invalid drop-frame timecode: " .. tc) + end + local minutes = h * 60 + m + frame = frame - dropped_per_minute * (minutes - math.floor(minutes / 10)) + end + return frame +end + +function Timing.source_start_seconds(segment, fps, clip_fps) + local source_fps = clip_fps or segment.source_fps + if segment.source_start_frame ~= nil and source_fps and source_fps > 0 then + return segment.source_start_frame / source_fps + end + return segment.left_offset / fps +end + +function Timing.source_zero_seconds(segment, fps, timeline_start) + return (segment.start_frame - timeline_start) / fps - Timing.source_start_seconds(segment, fps) +end + +return Timing diff --git a/scripts/build_installers.py b/scripts/build_installers.py index 6edd7b7..eab55c2 100644 --- a/scripts/build_installers.py +++ b/scripts/build_installers.py @@ -49,6 +49,8 @@ def pyinstaller_binary( "--noconfirm", "--name", output_name, + "--specpath", + str(ROOT / "build"), ] cmd.append("--onefile" if onefile else "--onedir") if windowed: @@ -177,316 +179,7 @@ def create_windows_exe(cli_binary: Path, version: str, python_bin: str = "python shutil.copy2(cli_binary, payload_bin_dir / "openbeat.exe") installer_entry = tmp_path / "windows_installer.py" - installer_entry.write_text( - """from __future__ import annotations - -import os -import shutil -import subprocess -import sys -import threading -from pathlib import Path -import tkinter as tk -from tkinter import messagebox, ttk - - -def resolve_payload_root() -> Path: - if hasattr(sys, '_MEIPASS'): - return Path(getattr(sys, '_MEIPASS')) / 'payload' - return Path(__file__).resolve().parent / 'payload' - - -def default_resolve_root() -> Path: - appdata = Path.home() / 'AppData' / 'Roaming' - if 'APPDATA' in os.environ: - appdata = Path(os.environ['APPDATA']) - return appdata / 'Blackmagic Design' / 'DaVinci Resolve' / 'Support' / 'Fusion' - - -def default_log_path() -> Path: - appdata = Path.home() / 'AppData' / 'Roaming' - if 'APPDATA' in os.environ: - appdata = Path(os.environ['APPDATA']) - return appdata / 'Blackmagic Design' / 'DaVinci Resolve' / 'Support' / 'logs' / 'OpenBeat.log' - - -def resolve_process_running() -> bool: - try: - result = subprocess.run( - ['tasklist', '/FI', 'IMAGENAME eq Resolve.exe'], - check=False, - capture_output=True, - text=True, - timeout=5, - ) - except Exception: - return False - return 'Resolve.exe' in result.stdout - - -def installed_runtime_exe(module_target: Path) -> Path: - onedir_exe = module_target / 'bin' / 'openbeat' / 'openbeat.exe' - if onedir_exe.exists(): - return onedir_exe - return module_target / 'bin' / 'openbeat.exe' - - -def open_path(path: Path) -> None: - target = path if path.is_dir() else path.parent - if not target.exists(): - target.mkdir(parents=True, exist_ok=True) - os.startfile(str(target)) - - -def install_openbeat(destination: Path) -> dict[str, str]: - payload = resolve_payload_root() - resolve_root = destination.expanduser() - - utility_target = resolve_root / 'Scripts' / 'Utility' / 'OpenBeat' - module_target = resolve_root / 'Modules' / 'OpenBeat' - - (resolve_root / 'Scripts' / 'Utility').mkdir(parents=True, exist_ok=True) - (resolve_root / 'Modules').mkdir(parents=True, exist_ok=True) - - if utility_target.exists(): - shutil.rmtree(utility_target) - if module_target.exists(): - shutil.rmtree(module_target) - - shutil.copytree(payload / 'Utility' / 'OpenBeat', utility_target) - shutil.copytree(payload / 'Modules' / 'OpenBeat', module_target) - (module_target / 'bin').mkdir(parents=True, exist_ok=True) - shutil.copytree(payload / 'bin', module_target / 'bin', dirs_exist_ok=True) - - exe_path = installed_runtime_exe(module_target) - config_text = f'return {{\\n python_bin = "{exe_path.as_posix()}",\\n}}\\n' - (module_target / 'OpenBeatConfig.local.lua').write_text(config_text, encoding='utf-8') - return { - 'destination': str(resolve_root), - 'utility_target': str(utility_target), - 'module_target': str(module_target), - 'runtime': str(exe_path), - 'log_path': str(default_log_path()), - } - - -class InstallerWizard(tk.Tk): - def __init__(self) -> None: - super().__init__() - self.title('OpenBeat Setup') - self.geometry('720x520') - self.minsize(640, 460) - self.resizable(False, False) - self.protocol('WM_DELETE_WINDOW', self.cancel) - - self.step = 0 - self.install_error: str | None = None - self.install_result: dict[str, str] | None = None - self.destination = default_resolve_root() - - self.columnconfigure(0, weight=1) - self.rowconfigure(0, weight=1) - self.configure(background='#f6f7f9') - - self.content = ttk.Frame(self, padding=(28, 24, 28, 18)) - self.content.grid(row=0, column=0, sticky='nsew') - self.content.columnconfigure(0, weight=1) - self.content.rowconfigure(5, weight=1) - - self.nav = ttk.Frame(self, padding=(18, 12)) - self.nav.grid(row=1, column=0, sticky='ew') - self.nav.columnconfigure(0, weight=1) - - self.back_button = ttk.Button(self.nav, text='< Back', command=self.go_back) - self.next_button = ttk.Button(self.nav, text='Next >', command=self.go_next) - self.cancel_button = ttk.Button(self.nav, text='Cancel', command=self.cancel) - self.back_button.grid(row=0, column=1, padx=(0, 8)) - self.next_button.grid(row=0, column=2, padx=(0, 8)) - self.cancel_button.grid(row=0, column=3) - - self.frames = [ - self.welcome_step, - self.ready_step, - self.installing_step, - self.finish_step, - ] - self.show_step(0) - - def clear_content(self) -> None: - for child in self.content.winfo_children(): - child.destroy() - - def add_heading(self, title: str, body: str) -> None: - ttk.Label(self.content, text=title, font=('Segoe UI', 18, 'bold')).grid( - row=0, column=0, sticky='w', pady=(0, 12) - ) - ttk.Label(self.content, text=body, wraplength=640, justify='left').grid( - row=1, column=0, sticky='nw' - ) - - def show_step(self, step: int) -> None: - self.step = step - self.clear_content() - self.frames[step]() - - self.back_button.configure(state='normal' if step == 1 else 'disabled') - self.cancel_button.configure(state='normal' if step not in (2, 3) else 'disabled') - - if step == 1: - self.next_button.configure(text='Install', state='normal') - elif step == 2: - self.next_button.configure(text='Next >', state='disabled') - elif step == 3: - self.next_button.configure(text='Finish', state='normal') - else: - self.next_button.configure(text='Next >', state='normal') - - def welcome_step(self) -> None: - self.add_heading( - 'Welcome to OpenBeat Setup', - 'This wizard will install the OpenBeat scripts and bundled runtime for DaVinci Resolve. Close Resolve before continuing if it is currently open.', - ) - - def ready_step(self) -> None: - self.add_heading( - 'Ready to Install', - 'OpenBeat will be copied into a DaVinci Resolve Fusion support folder and configured to use the bundled runtime.', - ) - summary = ttk.LabelFrame(self.content, text='Install summary', padding=14) - summary.grid(row=2, column=0, sticky='ew', pady=(24, 0)) - summary.columnconfigure(1, weight=1) - ttk.Label(summary, text='Destination').grid(row=0, column=0, sticky='nw', padx=(0, 12)) - ttk.Label(summary, text=str(self.destination), wraplength=500).grid(row=0, column=1, sticky='w') - ttk.Button(summary, text='Copy', command=lambda: self.copy_to_clipboard(str(self.destination))).grid(row=0, column=2, padx=(8, 0)) - - ttk.Label(summary, text='Runtime').grid(row=1, column=0, sticky='nw', padx=(0, 12), pady=(10, 0)) - ttk.Label(summary, text='Bundled OpenBeat runtime, no separate Python setup needed.', wraplength=470).grid( - row=1, column=1, columnspan=2, sticky='w', pady=(10, 0) - ) - - if resolve_process_running(): - warning = ttk.Label( - self.content, - text='DaVinci Resolve appears to be running. Close it before installing so the script menu reloads cleanly.', - wraplength=640, - foreground='#8a4b00', - ) - warning.grid(row=3, column=0, sticky='w', pady=(16, 0)) - - def copy_to_clipboard(self, value: str) -> None: - self.clipboard_clear() - self.clipboard_append(value) - self.update_idletasks() - - def open_install_folder(self) -> None: - result = self.install_result or {'destination': str(self.destination)} - open_path(Path(result['destination'])) - - def open_log_location(self) -> None: - result = self.install_result or {'log_path': str(default_log_path())} - open_path(Path(result['log_path'])) - - def installing_step(self) -> None: - self.add_heading('Installing OpenBeat', 'Please wait while setup copies files into Resolve.') - self.progress = ttk.Progressbar(self.content, mode='indeterminate') - self.progress.grid(row=2, column=0, sticky='ew', pady=(24, 8)) - self.status = ttk.Label(self.content, text='Preparing install...') - self.status.grid(row=3, column=0, sticky='w') - self.progress.start(12) - - def finish_step(self) -> None: - if self.install_error: - self.add_heading( - 'Installation Did Not Complete', - f'OpenBeat could not be installed.\\n\\n{self.install_error}', - ) - actions = ttk.Frame(self.content) - actions.grid(row=2, column=0, sticky='w', pady=(24, 0)) - ttk.Button(actions, text='Open install folder', command=self.open_install_folder).grid(row=0, column=0, padx=(0, 8)) - ttk.Button(actions, text='Open log folder', command=self.open_log_location).grid(row=0, column=1, padx=(0, 8)) - return - - self.add_heading( - 'OpenBeat Setup Complete', - 'OpenBeat has been installed successfully. Restart DaVinci Resolve if it is open, then use Workspace > Scripts > OpenBeat.', - ) - result = self.install_result or {} - summary = ttk.LabelFrame(self.content, text='Installed files', padding=14) - summary.grid(row=2, column=0, sticky='ew', pady=(24, 0)) - summary.columnconfigure(1, weight=1) - rows = [ - ('Scripts', result.get('utility_target', '')), - ('Runtime', result.get('runtime', '')), - ('Log file', result.get('log_path', str(default_log_path()))), - ] - for row_index, (label, value) in enumerate(rows): - ttk.Label(summary, text=label).grid(row=row_index, column=0, sticky='nw', padx=(0, 12), pady=(0, 8)) - ttk.Label(summary, text=value, wraplength=500).grid(row=row_index, column=1, sticky='w', pady=(0, 8)) - - actions = ttk.Frame(self.content) - actions.grid(row=3, column=0, sticky='w', pady=(18, 0)) - ttk.Button(actions, text='Open install folder', command=self.open_install_folder).grid(row=0, column=0, padx=(0, 8)) - ttk.Button(actions, text='Open log folder', command=self.open_log_location).grid(row=0, column=1, padx=(0, 8)) - ttk.Button(actions, text='Copy destination', command=lambda: self.copy_to_clipboard(result.get('destination', ''))).grid(row=0, column=2) - - def go_back(self) -> None: - if self.step > 0: - self.show_step(self.step - 1) - - def go_next(self) -> None: - if self.step == 3: - self.destroy() - return - if self.step == 1: - self.start_install() - return - self.show_step(self.step + 1) - - def cancel(self) -> None: - if self.step == 2: - messagebox.showinfo('OpenBeat Setup', 'Setup is currently installing OpenBeat.') - return - if messagebox.askyesno('Cancel Setup', 'Are you sure you want to cancel OpenBeat Setup?'): - self.destroy() - - def start_install(self) -> None: - if resolve_process_running(): - should_continue = messagebox.askyesno( - 'DaVinci Resolve Is Running', - 'DaVinci Resolve appears to be running. Close it before installing for the cleanest result. Continue anyway?', - ) - if not should_continue: - return - - self.install_error = None - self.install_result = None - destination = self.destination.expanduser() - self.show_step(2) - - def worker() -> None: - try: - self.install_result = install_openbeat(destination) - except Exception as exc: - self.install_error = str(exc) - self.after(0, self.install_finished) - - threading.Thread(target=worker, daemon=True).start() - - def install_finished(self) -> None: - if hasattr(self, 'progress'): - self.progress.stop() - self.show_step(3) - - -def main() -> int: - InstallerWizard().mainloop() - return 0 - - -if __name__ == '__main__': - raise SystemExit(main()) -""" - ) + shutil.copy2(ROOT / "scripts" / "windows_installer.py", installer_entry) run( [ diff --git a/scripts/smoke_test_runtime.py b/scripts/smoke_test_runtime.py new file mode 100644 index 0000000..4b5e8de --- /dev/null +++ b/scripts/smoke_test_runtime.py @@ -0,0 +1,86 @@ +"""Exercise the actual CLI/runtime with generated audio, outside the checkout.""" +from __future__ import annotations + +import argparse +import json +import subprocess +import tempfile +from pathlib import Path + +import numpy as np +import soundfile as sf + +from openbeat.analysis import cache_key, default_cache_dir + + +def write_test_audio(path: Path) -> None: + sample_rate = 22050 + signal = np.zeros(16 * sample_rate, dtype=np.float32) + time = np.arange(int(0.03 * sample_rate)) / sample_rate + click = np.sin(2 * np.pi * 1800 * time) * np.exp(-time * 55) + for beat in 0.2 + np.arange(32) * 0.5: + start = round(float(beat) * sample_rate) + signal[start : start + len(click)] += click + sf.write(path, signal, sample_rate) + + +def verify_runtime(command: list[str]) -> dict: + with tempfile.TemporaryDirectory(prefix="openbeat-smoke-") as temporary: + work_dir = Path(temporary) + audio = work_dir / "test music 音楽.wav" + write_test_audio(audio) + cache = default_cache_dir() / f"{cache_key(str(audio))}.json" + + def run(*args: str) -> None: + result = subprocess.run(command + list(args), cwd=work_dir, text=True, capture_output=True, timeout=180) + if result.returncode: + raise RuntimeError(f"Runtime failed ({result.returncode}): {result.stdout}\n{result.stderr}") + + try: + analysis_file = work_dir / "analysis.json" + run("analyze", "--audio", str(audio), "--output", str(analysis_file)) + analysis = json.loads(analysis_file.read_text(encoding="utf-8")) + if not 119 <= analysis["tempo_bpm"] <= 121 or len(analysis["raw_beats"]) < 25: + raise AssertionError(f"Unexpected real-audio analysis: {analysis}") + if abs(analysis["duration_seconds"] - 16) > 0.001 or analysis["audio_path"] != str(audio.resolve()): + raise AssertionError("Analysis did not preserve source duration/path") + period = 60 / analysis["tempo_bpm"] + phase_error = abs((analysis["grid_offset_seconds"] - 0.2 + period / 2) % period - period / 2) + if phase_error > 0.05: + raise AssertionError(f"Beat phase differs by {phase_error:.3f}s") + + lua_file = work_dir / "analysis.lua" + run("analyze", "--audio", str(audio), "--format", "lua", "--output", str(lua_file)) + lua = lua_file.read_text(encoding="utf-8") + if not lua.startswith("return {") or "音楽" not in lua or "raw_beats" not in lua: + raise AssertionError("Runtime did not produce its Lua transport format") + + for mode in ("raw", "quantized"): + output = work_dir / f"{mode}.wav" + run("click-track", "--audio", str(audio), "--mode", mode, "--output", str(output)) + waveform, sample_rate = sf.read(output) + if sample_rate != 48000 or waveform.shape != (17 * 48000, 2): + raise AssertionError("Click track has incorrect rate, duration, or channels") + if not np.array_equal(waveform[:, 0], waveform[:, 1]): + raise AssertionError("Click channels differ") + beats = analysis["raw_beats"] if mode == "raw" else analysis["quantized_beats"] + for beat in beats: + start = round(beat * sample_rate) + if np.max(np.abs(waveform[start : start + 1000, 0])) < 0.1: + raise AssertionError(f"Missing {mode} click at {beat:.3f}s") + return analysis + finally: + cache.unlink(missing_ok=True) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("runtime", type=Path) + args = parser.parse_args() + analysis = verify_runtime([str(args.runtime.resolve())]) + print(f"Runtime smoke test passed: {len(analysis['raw_beats'])} beats, {analysis['tempo_bpm']:.3f} BPM") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/windows_installer.py b/scripts/windows_installer.py new file mode 100644 index 0000000..a54929d --- /dev/null +++ b/scripts/windows_installer.py @@ -0,0 +1,307 @@ +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import threading +from pathlib import Path +import tkinter as tk +from tkinter import messagebox, ttk + + +def resolve_payload_root() -> Path: + if hasattr(sys, '_MEIPASS'): + return Path(getattr(sys, '_MEIPASS')) / 'payload' + return Path(__file__).resolve().parent / 'payload' + + +def default_resolve_root() -> Path: + appdata = Path.home() / 'AppData' / 'Roaming' + if 'APPDATA' in os.environ: + appdata = Path(os.environ['APPDATA']) + return appdata / 'Blackmagic Design' / 'DaVinci Resolve' / 'Support' / 'Fusion' + + +def default_log_path() -> Path: + appdata = Path.home() / 'AppData' / 'Roaming' + if 'APPDATA' in os.environ: + appdata = Path(os.environ['APPDATA']) + return appdata / 'Blackmagic Design' / 'DaVinci Resolve' / 'Support' / 'logs' / 'OpenBeat.log' + + +def resolve_process_running() -> bool: + try: + result = subprocess.run( + ['tasklist', '/FI', 'IMAGENAME eq Resolve.exe'], + check=False, + capture_output=True, + text=True, + timeout=5, + ) + except Exception: + return False + return 'Resolve.exe' in result.stdout + + +def installed_runtime_exe(module_target: Path) -> Path: + onedir_exe = module_target / 'bin' / 'openbeat' / 'openbeat.exe' + if onedir_exe.exists(): + return onedir_exe + return module_target / 'bin' / 'openbeat.exe' + + +def open_path(path: Path) -> None: + target = path if path.is_dir() else path.parent + if not target.exists(): + target.mkdir(parents=True, exist_ok=True) + os.startfile(str(target)) + + +def install_openbeat(destination: Path) -> dict[str, str]: + payload = resolve_payload_root() + resolve_root = destination.expanduser() + + utility_target = resolve_root / 'Scripts' / 'Utility' / 'OpenBeat' + module_target = resolve_root / 'Modules' / 'OpenBeat' + + (resolve_root / 'Scripts' / 'Utility').mkdir(parents=True, exist_ok=True) + (resolve_root / 'Modules').mkdir(parents=True, exist_ok=True) + + if utility_target.exists(): + shutil.rmtree(utility_target) + if module_target.exists(): + shutil.rmtree(module_target) + + shutil.copytree(payload / 'Utility' / 'OpenBeat', utility_target) + shutil.copytree(payload / 'Modules' / 'OpenBeat', module_target) + (module_target / 'bin').mkdir(parents=True, exist_ok=True) + shutil.copytree(payload / 'bin', module_target / 'bin', dirs_exist_ok=True) + + exe_path = installed_runtime_exe(module_target) + config_text = f'return {{\n python_bin = "{exe_path.as_posix()}",\n}}\n' + (module_target / 'OpenBeatConfig.local.lua').write_text(config_text, encoding='utf-8') + return { + 'destination': str(resolve_root), + 'utility_target': str(utility_target), + 'module_target': str(module_target), + 'runtime': str(exe_path), + 'log_path': str(default_log_path()), + } + + +class InstallerWizard(tk.Tk): + def __init__(self) -> None: + super().__init__() + self.title('OpenBeat Setup') + self.geometry('720x520') + self.minsize(640, 460) + self.resizable(False, False) + self.protocol('WM_DELETE_WINDOW', self.cancel) + + self.step = 0 + self.install_error: str | None = None + self.install_result: dict[str, str] | None = None + self.destination = default_resolve_root() + + self.columnconfigure(0, weight=1) + self.rowconfigure(0, weight=1) + self.configure(background='#f6f7f9') + + self.content = ttk.Frame(self, padding=(28, 24, 28, 18)) + self.content.grid(row=0, column=0, sticky='nsew') + self.content.columnconfigure(0, weight=1) + self.content.rowconfigure(5, weight=1) + + self.nav = ttk.Frame(self, padding=(18, 12)) + self.nav.grid(row=1, column=0, sticky='ew') + self.nav.columnconfigure(0, weight=1) + + self.back_button = ttk.Button(self.nav, text='< Back', command=self.go_back) + self.next_button = ttk.Button(self.nav, text='Next >', command=self.go_next) + self.cancel_button = ttk.Button(self.nav, text='Cancel', command=self.cancel) + self.back_button.grid(row=0, column=1, padx=(0, 8)) + self.next_button.grid(row=0, column=2, padx=(0, 8)) + self.cancel_button.grid(row=0, column=3) + + self.frames = [ + self.welcome_step, + self.ready_step, + self.installing_step, + self.finish_step, + ] + self.show_step(0) + + def clear_content(self) -> None: + for child in self.content.winfo_children(): + child.destroy() + + def add_heading(self, title: str, body: str) -> None: + ttk.Label(self.content, text=title, font=('Segoe UI', 18, 'bold')).grid( + row=0, column=0, sticky='w', pady=(0, 12) + ) + ttk.Label(self.content, text=body, wraplength=640, justify='left').grid( + row=1, column=0, sticky='nw' + ) + + def show_step(self, step: int) -> None: + self.step = step + self.clear_content() + self.frames[step]() + + self.back_button.configure(state='normal' if step == 1 else 'disabled') + self.cancel_button.configure(state='normal' if step not in (2, 3) else 'disabled') + + if step == 1: + self.next_button.configure(text='Install', state='normal') + elif step == 2: + self.next_button.configure(text='Next >', state='disabled') + elif step == 3: + self.next_button.configure(text='Finish', state='normal') + else: + self.next_button.configure(text='Next >', state='normal') + + def welcome_step(self) -> None: + self.add_heading( + 'Welcome to OpenBeat Setup', + 'This wizard will install the OpenBeat scripts and bundled runtime for DaVinci Resolve. Close Resolve before continuing if it is currently open.', + ) + + def ready_step(self) -> None: + self.add_heading( + 'Ready to Install', + 'OpenBeat will be copied into a DaVinci Resolve Fusion support folder and configured to use the bundled runtime.', + ) + summary = ttk.LabelFrame(self.content, text='Install summary', padding=14) + summary.grid(row=2, column=0, sticky='ew', pady=(24, 0)) + summary.columnconfigure(1, weight=1) + ttk.Label(summary, text='Destination').grid(row=0, column=0, sticky='nw', padx=(0, 12)) + ttk.Label(summary, text=str(self.destination), wraplength=500).grid(row=0, column=1, sticky='w') + ttk.Button(summary, text='Copy', command=lambda: self.copy_to_clipboard(str(self.destination))).grid(row=0, column=2, padx=(8, 0)) + + ttk.Label(summary, text='Runtime').grid(row=1, column=0, sticky='nw', padx=(0, 12), pady=(10, 0)) + ttk.Label(summary, text='Bundled OpenBeat runtime, no separate Python setup needed.', wraplength=470).grid( + row=1, column=1, columnspan=2, sticky='w', pady=(10, 0) + ) + + if resolve_process_running(): + warning = ttk.Label( + self.content, + text='DaVinci Resolve appears to be running. Close it before installing so the script menu reloads cleanly.', + wraplength=640, + foreground='#8a4b00', + ) + warning.grid(row=3, column=0, sticky='w', pady=(16, 0)) + + def copy_to_clipboard(self, value: str) -> None: + self.clipboard_clear() + self.clipboard_append(value) + self.update_idletasks() + + def open_install_folder(self) -> None: + result = self.install_result or {'destination': str(self.destination)} + open_path(Path(result['destination'])) + + def open_log_location(self) -> None: + result = self.install_result or {'log_path': str(default_log_path())} + open_path(Path(result['log_path'])) + + def installing_step(self) -> None: + self.add_heading('Installing OpenBeat', 'Please wait while setup copies files into Resolve.') + self.progress = ttk.Progressbar(self.content, mode='indeterminate') + self.progress.grid(row=2, column=0, sticky='ew', pady=(24, 8)) + self.status = ttk.Label(self.content, text='Preparing install...') + self.status.grid(row=3, column=0, sticky='w') + self.progress.start(12) + + def finish_step(self) -> None: + if self.install_error: + self.add_heading( + 'Installation Did Not Complete', + f'OpenBeat could not be installed.\n\n{self.install_error}', + ) + actions = ttk.Frame(self.content) + actions.grid(row=2, column=0, sticky='w', pady=(24, 0)) + ttk.Button(actions, text='Open install folder', command=self.open_install_folder).grid(row=0, column=0, padx=(0, 8)) + ttk.Button(actions, text='Open log folder', command=self.open_log_location).grid(row=0, column=1, padx=(0, 8)) + return + + self.add_heading( + 'OpenBeat Setup Complete', + 'OpenBeat has been installed successfully. Restart DaVinci Resolve if it is open, then use Workspace > Scripts > OpenBeat.', + ) + result = self.install_result or {} + summary = ttk.LabelFrame(self.content, text='Installed files', padding=14) + summary.grid(row=2, column=0, sticky='ew', pady=(24, 0)) + summary.columnconfigure(1, weight=1) + rows = [ + ('Scripts', result.get('utility_target', '')), + ('Runtime', result.get('runtime', '')), + ('Log file', result.get('log_path', str(default_log_path()))), + ] + for row_index, (label, value) in enumerate(rows): + ttk.Label(summary, text=label).grid(row=row_index, column=0, sticky='nw', padx=(0, 12), pady=(0, 8)) + ttk.Label(summary, text=value, wraplength=500).grid(row=row_index, column=1, sticky='w', pady=(0, 8)) + + actions = ttk.Frame(self.content) + actions.grid(row=3, column=0, sticky='w', pady=(18, 0)) + ttk.Button(actions, text='Open install folder', command=self.open_install_folder).grid(row=0, column=0, padx=(0, 8)) + ttk.Button(actions, text='Open log folder', command=self.open_log_location).grid(row=0, column=1, padx=(0, 8)) + ttk.Button(actions, text='Copy destination', command=lambda: self.copy_to_clipboard(result.get('destination', ''))).grid(row=0, column=2) + + def go_back(self) -> None: + if self.step > 0: + self.show_step(self.step - 1) + + def go_next(self) -> None: + if self.step == 3: + self.destroy() + return + if self.step == 1: + self.start_install() + return + self.show_step(self.step + 1) + + def cancel(self) -> None: + if self.step == 2: + messagebox.showinfo('OpenBeat Setup', 'Setup is currently installing OpenBeat.') + return + if messagebox.askyesno('Cancel Setup', 'Are you sure you want to cancel OpenBeat Setup?'): + self.destroy() + + def start_install(self) -> None: + if resolve_process_running(): + should_continue = messagebox.askyesno( + 'DaVinci Resolve Is Running', + 'DaVinci Resolve appears to be running. Close it before installing for the cleanest result. Continue anyway?', + ) + if not should_continue: + return + + self.install_error = None + self.install_result = None + destination = self.destination.expanduser() + self.show_step(2) + + def worker() -> None: + try: + self.install_result = install_openbeat(destination) + except Exception as exc: + self.install_error = str(exc) + self.after(0, self.install_finished) + + threading.Thread(target=worker, daemon=True).start() + + def install_finished(self) -> None: + if hasattr(self, 'progress'): + self.progress.stop() + self.show_step(3) + + +def main() -> int: + InstallerWizard().mainloop() + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tests/lua/test_markers.lua b/tests/lua/test_markers.lua new file mode 100644 index 0000000..6bbbee1 --- /dev/null +++ b/tests/lua/test_markers.lua @@ -0,0 +1,75 @@ +local Markers = dofile("resolve/Fusion/Modules/OpenBeat/OpenBeatMarkers.lua") + +local function marker(custom, name) + return { customData = custom, name = name or "old", note = "note", color = "Blue", duration = 2 } +end +local function holder(markers) + return { + markers = markers or {}, + GetMarkers = function(self) + local copy = {} + for frame, value in pairs(self.markers) do copy[frame] = value end + return copy + end, + DeleteMarkerAtFrame = function(self, frame) + if self.fail_delete == frame then self.fail_delete = nil; return false end + self.markers[frame] = nil + return true + end, + AddMarker = function(self, frame, color, name, note, duration, custom) + if self.fail_add == frame then self.fail_add = nil; return false end + if self.markers[frame] then return false end + self.markers[frame] = { color = color, name = name, note = note, duration = duration, customData = custom } + return true + end, + } +end +local custom_a = Markers.custom_data("timeline", "a.wav") +local custom_b = Markers.custom_data("timeline", "b.wav") +local target = holder({ [0] = marker(custom_a), [12] = marker("user"), [24] = marker(custom_b), [36] = marker("OpenBeat:timeline") }) +local created, skipped = Markers.replace_many({ { + holder = target, custom_data = custom_a, + plan = { { frame = 6, index = 1 }, { frame = 12, index = 2 }, { frame = 24, index = 3 }, { frame = 36, index = 4 } }, +} }) +assert(created == 1 and skipped == 3) +assert(not target.markers[0] and target.markers[6].customData == custom_a) +assert(target.markers[12].customData == "user" and target.markers[24].customData == custom_b) +assert(target.markers[36].customData == "OpenBeat:timeline", "Legacy ownership is ambiguous") + +for _, failure in ipairs({ "add", "delete" }) do + local first = holder({ [0] = marker(custom_a, "first") }) + local second = holder({ [1] = marker(custom_a, "second") }) + if failure == "add" then second.fail_add = 12 else second.fail_delete = 1 end + local ok, reason = pcall(Markers.replace_many, { + { holder = first, custom_data = custom_a, plan = { { frame = 6, index = 1 } } }, + { holder = second, custom_data = custom_a, plan = { { frame = 12, index = 2 } } }, + }) + assert(not ok and reason:find("Previous markers restored", 1, true)) + assert(first.markers[0].name == "first" and first.markers[0].duration == 2 and not first.markers[6]) + assert(second.markers[1].name == "second" and not second.markers[12]) +end + +local first = holder({ [0] = marker(custom_a) }) +local second = { GetMarkers = function() error("Simulated preparation failure") end } +assert(not pcall(Markers.replace_many, { + { holder = first, custom_data = custom_a, plan = { { frame = 6, index = 1 } } }, + { holder = second, custom_data = custom_a, plan = {} }, +})) +assert(first.markers[0] and not first.markers[6], "Preparation must finish before deletion") + +local plan = Markers.timeline_plan({ 0, 0.5, 1, 1.5, 2 }, { + { start_frame = 86400, end_frame = 86424, left_offset = 12 }, + { start_frame = 86400, end_frame = 86424, left_offset = 12 }, +}, 24, 86400) +assert(#plan == 2 and plan[1].frame == 0 and plan[1].index == 2 and plan[2].frame == 12) +plan = Markers.clip_plan({ 0.5, 1, 1.5, 2 }, { + source_start_frame = 30, left_offset = 24, start_frame = 86400, end_frame = 86424, +}, 24, 30) +assert(#plan == 2 and plan[1].frame == 0 and plan[2].frame == 12, "Clip offsets must use timeline frames") +assert(#Markers.source_plan({ 0, 1 }, 24, 1) == 1, "Source end is exclusive") +assert(#Markers.timeline_plan({ 0, 0.01 }, { { start_frame = 0, end_frame = 24, left_offset = 0 } }, 24, 0) == 1) + +plan = Markers.timeline_plan({ 0.5, 1, 1.5, 2 }, { + { source_start_frame = 30, source_fps = 30, left_offset = 30, start_frame = 86400, end_frame = 86424 }, +}, 24, 86400) +assert(#plan == 2 and plan[1].frame == 0 and plan[2].frame == 12, "Source and timeline rates must be distinct") diff --git a/tests/lua/test_source.lua b/tests/lua/test_source.lua new file mode 100644 index 0000000..f332a08 --- /dev/null +++ b/tests/lua/test_source.lua @@ -0,0 +1,35 @@ +local Source = dofile("resolve/Fusion/Modules/OpenBeat/OpenBeatSource.lua") + +local function clip(path, start_frame, end_frame) + local media = { GetClipProperty = function() return path end } + return { + GetMediaPoolItem = function() return media end, + GetStart = function() return start_frame end, + GetEnd = function() return end_frame end, + GetLeftOffset = function() return 12 end, + } +end + +local first = clip("first.wav", 86400, 86412) +local second = clip({ ["File Path"] = "second.wav" }, 86412, 86424) +local timeline = { + GetCurrentTimecode = function() return "01:00:00:12" end, + GetTrackCount = function() return 3 end, + GetItemListInTrack = function(_, _, index) + if index == 1 then return nil end + if index == 2 then return { clip("", 86400, 86424), first, second } end + return { clip("other.wav", 86400, 86424) } + end, + GetCurrentVideoItem = function() error("Video item must not select the audio source") end, +} +local path, item = Source.at_playhead(timeline, 23.976) +assert(path == "second.wav" and item == second, "Wrong source at the cut boundary") +local segments = Source.segments_for_path(timeline, "second.wav") +assert(#segments == 1 and segments[1].track_index == 2 and segments[1].left_offset == 12) + +timeline.GetCurrentTimecode = function() return "01:00:01:00" end +assert(not pcall(Source.at_playhead, timeline, 24), "Clip end should be exclusive") + +timeline.GetCurrentTimecode = function() return "01:00:00;00" end +timeline.GetItemListInTrack = function() return { clip("drop.wav", 107892, 107922) } end +assert(Source.at_playhead(timeline, 29.97) == "drop.wav") diff --git a/tests/lua/test_subtitle_import.lua b/tests/lua/test_subtitle_import.lua new file mode 100644 index 0000000..1c66e07 --- /dev/null +++ b/tests/lua/test_subtitle_import.lua @@ -0,0 +1,64 @@ +local OpenBeat = dofile("resolve/Fusion/Modules/OpenBeat/OpenBeatCommon.lua") +local function upvalue(fn, name) + for index = 1, 100 do + local key, value = debug.getupvalue(fn, index) + if key == name then return value end + if not key then break end + end + error("Missing upvalue " .. name) +end +local place = upvalue(upvalue(OpenBeat.run, "export_subtitles"), "import_subtitles_to_timeline") + +-- A restoration failure must not prevent attempts to restore the other tracks. +local states = { video = { true, false }, subtitle = { false } } +local restoring = false +local timeline = { + GetTrackCount = function(_, kind) return #states[kind] end, + AddTrack = function(_, kind) table.insert(states[kind], true); return true end, + GetIsTrackEnabled = function(_, kind, index) return states[kind][index] end, + SetTrackEnable = function(_, kind, index, enabled) + if restoring and kind == "video" and index == 1 then error("Cannot restore first video track") end + states[kind][index] = enabled + return true + end, + GetItemListInTrack = function() return {} end, + GetStartFrame = function() return 86400 end, +} +local project = { GetMediaPool = function() + return { + ImportMedia = function() return { {} } end, + AppendToTimeline = function() restoring = true; error("Cannot append") end, + } +end } +local ok, reason = place({ OpenPage = function() end }, project, timeline, "mock.srt") +assert(not ok and reason:find("could not restore all", 1, true)) +assert(states.video[2] == false and states.subtitle[1] == false and states.subtitle[2] == true) +for _, failure in ipairs({ "append", "disable" }) do + local states = { video = { true, false }, subtitle = { false } } + local fail_once = true + local timeline = { + GetTrackCount = function(_, kind) return #states[kind] end, + AddTrack = function(_, kind) table.insert(states[kind], true); return true end, + GetIsTrackEnabled = function(_, kind, index) return states[kind][index] end, + SetTrackEnable = function(_, kind, index, enabled) + states[kind][index] = enabled + if failure == "disable" and fail_once then + fail_once = false + error("Simulated track failure") + end + return true + end, + GetItemListInTrack = function() return {} end, + GetStartFrame = function() return 86400 end, + } + local project = { GetMediaPool = function() + return { + ImportMedia = function() return { {} } end, + AppendToTimeline = function() error("Simulated append failure") end, + } + end } + local ok, reason = place({ OpenPage = function() end }, project, timeline, "mock.srt") + assert(not ok and reason:find("Simulated")) + assert(states.video[1] == true and states.video[2] == false, "Video states were not restored") + assert(states.subtitle[1] == false and states.subtitle[2] == true, "Subtitle states were not restored") +end diff --git a/tests/lua/test_subtitles.lua b/tests/lua/test_subtitles.lua new file mode 100644 index 0000000..460a6bf --- /dev/null +++ b/tests/lua/test_subtitles.lua @@ -0,0 +1,36 @@ +local Subtitles = dofile("resolve/Fusion/Modules/OpenBeat/OpenBeatSubtitles.lua") +local segments = { + { start_frame = 86400, end_frame = 86424, left_offset = 0 }, + { start_frame = 86640, end_frame = 86664, left_offset = 0 }, +} +local cues = Subtitles.plan({ 0, 0.5, 1 }, segments, 24, 86400) +assert(#cues == 4, "Cues must stay inside the two source segments") +assert(cues[1].start_time == 0 and cues[2].end_time == 1) +assert(cues[3].start_time == 10 and cues[4].end_time == 11) +assert(cues[1].beat_number == 1 and cues[3].beat_number == 1) +local srt = Subtitles.to_srt(cues) +assert(srt:find("00:00:00,000 --> 00:00:00,500", 1, true)) +assert(not srt:find("01:00:", 1, true), "SRT must be relative to the timeline origin") +assert(not srt:find("00:00:01,000 --> 00:00:10,000", 1, true), "Cue crosses a gap") + +-- Trims keep the partially visible beat and stop the final cue at the clip end. +cues = Subtitles.plan({ 0.2, 0.7, 1.2, 1.7 }, { + { start_frame = 240, end_frame = 264, left_offset = 12 }, +}, 24, 0) +assert(#cues == 3) +assert(cues[1].start_time == 10 and cues[1].end_time == 10.2 and cues[1].beat_number == 1) +assert(cues[3].end_time == 11) + +cues = Subtitles.plan({ 0.2, 0.7 }, { + { start_frame = 240, end_frame = 264, left_offset = 0 }, + { start_frame = 240, end_frame = 264, left_offset = 0 }, +}, 24, 0) +assert(#cues == 3, "Identical overlapping uses should not duplicate cues") +assert(cues[1].start_time == 10 and cues[1].end_time == 10.2 and not cues[1].beat_number) +assert(cues[3].end_time == 11) +assert(#Subtitles.plan({}, segments, 24, 86400) == 0) + +cues = Subtitles.plan({ 0.5, 1, 1.5, 2 }, { + { source_start_frame = 30, source_fps = 30, left_offset = 30, start_frame = 86400, end_frame = 86424 }, +}, 24, 86400) +assert(#cues == 2 and cues[1].start_time == 0 and cues[2].start_time == 0.5 and cues[2].end_time == 1) diff --git a/tests/lua/test_timing.lua b/tests/lua/test_timing.lua new file mode 100644 index 0000000..b11f26a --- /dev/null +++ b/tests/lua/test_timing.lua @@ -0,0 +1,34 @@ +local Timing = dofile("resolve/Fusion/Modules/OpenBeat/OpenBeatTiming.lua") + +local function equal(actual, expected) + assert(actual == expected, tostring(actual) .. " ~= " .. tostring(expected)) +end + +equal(Timing.timecode_to_frame("01:00:00:00", 24), 86400) +equal(Timing.timecode_to_frame("01:00:00:00", 23.976), 86400) +equal(Timing.timecode_to_frame("01:00:00:00", 24000 / 1001), 86400) +equal(Timing.timecode_to_frame("01:00:00:00", 29.97), 108000) +equal(Timing.timecode_to_frame("01:00:00:00", 59.94), 216000) +equal(Timing.timecode_to_frame("00:00:59;29", 29.97), 1799) +equal(Timing.timecode_to_frame("00:01:00;02", 29.97), 1800) +equal(Timing.timecode_to_frame("00:10:00;00", 29.97), 17982) +equal(Timing.timecode_to_frame("01:00:00;00", 30000 / 1001), 107892) +equal(Timing.timecode_to_frame("01:00:00:00", 29.97, true), 107892) +equal(Timing.timecode_to_frame("01:00:00;00", 59.94), 215784) +equal(Timing.timecode_to_frame("00:01:00;04", 59.94), 3600) + +for _, tc in ipairs({ "00:01:00;00", "00:01:00;01", "00:60:00:00", "00:00:00:30", "bad", "x01:00:00:00" }) do + assert(not pcall(Timing.timecode_to_frame, tc, 29.97), "Accepted invalid timecode " .. tc) +end +assert(not pcall(Timing.timecode_to_frame, "01:00:00;00", 24)) + +local fps, drop_frame = Timing.frame_rate("29.97 DF") +equal(fps, 29.97) +assert(drop_frame) +fps, drop_frame = Timing.frame_rate({ timelineFrameRate = "59.94 DF" }) +equal(fps, 59.94) +assert(drop_frame) +fps, drop_frame = Timing.frame_rate("23.976") +equal(fps, 23.976) +assert(not drop_frame) +assert(not pcall(Timing.frame_rate, "invalid")) diff --git a/tests/lua/test_workflow.lua b/tests/lua/test_workflow.lua new file mode 100644 index 0000000..ad76c85 --- /dev/null +++ b/tests/lua/test_workflow.lua @@ -0,0 +1,141 @@ +local directory = assert(arg[1]) +local source_path = directory .. "/test music 音楽.wav" +local open_file = io.open +io.open = function(path, mode) + if tostring(path):match("OpenBeat%.log$") then return nil end + return open_file(path, mode) +end +local function holder() + return { + markers = {}, + GetMarkers = function(self) + local copy = {} + for frame, marker in pairs(self.markers) do copy[frame] = marker end + return copy + end, + DeleteMarkerAtFrame = function(self, frame) self.markers[frame] = nil; return true end, + AddMarker = function(self, frame, color, name, note, duration, custom) + if self.markers[frame] then return false end + self.markers[frame] = { color = color, name = name, note = note, duration = duration, customData = custom } + return true + end, + } +end +local media = holder() +media.GetClipProperty = function(_, key) return key == "File Path" and source_path or "24" end +local function segment(start_frame) + local item = holder() + item.GetMediaPoolItem = function() return media end + item.GetStart = function() return start_frame end + item.GetEnd = function() return start_frame + 36 end + item.GetLeftOffset = function() return 12 end + item.GetSourceStartFrame = function() return 12 end + return item +end +local clips = { segment(86400), segment(86640) } +local states = { video = { true, false }, subtitle = { false } } +local subtitle_items = {} +local timeline = holder() +timeline.GetSetting = function() return "24" end +timeline.GetCurrentTimecode = function() return "01:00:00:06" end +timeline.GetStartFrame = function() return 86400 end +timeline.GetTrackCount = function(_, kind) return kind == "audio" and 1 or #states[kind] end +timeline.GetItemListInTrack = function(_, kind, index) + if kind == "audio" then return clips end + return subtitle_items[index] or {} +end +timeline.GetCurrentVideoItem = function() error("Video selection should never be used") end +timeline.AddTrack = function(_, kind) table.insert(states[kind], true); return true end +timeline.GetIsTrackEnabled = function(_, kind, index) return states[kind][index] end +timeline.SetTrackEnable = function(_, kind, index, enabled) states[kind][index] = enabled; return true end +local placement_fails = true +local pool = { + ImportMedia = function(_, paths) + local handle = assert(io.open(paths[1])); handle:close() + return { {} } + end, + AppendToTimeline = function(_, info) + assert(info[1].recordFrame == 86400 and info[1].startFrame == 0) + if placement_fails then error("Simulated placement failure") end + subtitle_items[info[1].trackIndex] = { {} } + return { {} } + end, +} +local project = { GetCurrentTimeline = function() return timeline end, GetMediaPool = function() return pool end } +local imports = {} +resolve = { + GetProjectManager = function() return { GetCurrentProject = function() return project end } end, + GetMediaStorage = function() return { AddItemListToMediaPool = function(_, path) + local handle = assert(io.open(path)); handle:close() + table.insert(imports, path) + return { {} } + end } end, + OpenPage = function() return true end, +} + +local commands = {} +io.popen = function(command) + table.insert(commands, command) + local output = assert(command:match("%-%-output '([^']+)'")) + local file = assert(io.open(output, "w")) + if command:find(" analyze ", 1, true) then + file:write("return { duration_seconds = 2, raw_beats = {0.25, 0.75, 1.25, 1.75}, quantized_beats = {0, 0.5, 1, 1.5} }") + else + assert(command:find(" click-track ", 1, true)) + file:write("mock click audio") + end + file:close() + return { read = function() return "" end, close = function() return true, "exit", 0 end } +end + +local OpenBeat = dofile("resolve/Fusion/Modules/OpenBeat/OpenBeatCommon.lua") +timeline:AddMarker(42, "Red", "user", "", 1, "user") +timeline:AddMarker(48, "Blue", "other song", "", 1, "OpenBeat:timeline:other.wav") +for _, mode in ipairs({ "raw", "quantized" }) do + OpenBeat.run("timeline_markers", mode) + local count = 0 + for frame, marker in pairs(timeline.markers) do + if marker.customData == "OpenBeat:timeline:" .. source_path then + count = count + 1 + assert(frame < 36 or (frame >= 240 and frame < 276), "Marker outside used ranges") + end + end + assert(count == 6 and timeline.markers[42].customData == "user" and timeline.markers[48]) + OpenBeat.run("clip_markers", mode) + count = 0 + for _ in pairs(media.markers) do count = count + 1 end + assert(count == 4) + for _, clip in ipairs(clips) do + count = 0 + for frame in pairs(clip.markers) do count = count + 1; assert(frame >= 0 and frame < 36) end + assert(count == 3) + end + OpenBeat.run("click_track", mode) + OpenBeat.run("subtitles", mode) + local output = source_path:gsub("%.wav$", mode == "raw" and ".openbeat-raw.srt" or ".openbeat.srt") + local file = assert(io.open(output)); local srt = file:read("*a"); file:close() + assert(srt:find("00:00:00,000", 1, true) and not srt:find("01:00:", 1, true)) + assert(not srt:find("00:00:01,500 --> 00:00:10,000", 1, true)) + assert(states.video[1] == true and states.video[2] == false and states.subtitle[1] == false) + placement_fails = false +end +assert(#commands == 8 and #imports == 2) +assert(imports[1]:find(".openbeat-raw-clicks.wav", 1, true) and imports[2]:find(".openbeat-clicks.wav", 1, true)) +assert(not pcall(OpenBeat.run, "timeline_markers", "invalid")) +assert(not pcall(OpenBeat.run, "invalid", "raw")) + +-- Resolve can indicate drop-frame in the setting while returning colon timecode. +timeline.GetSetting = function() return "29.97 DF" end +timeline.GetCurrentTimecode = function() return "01:00:00:06" end +timeline.GetStartFrame = function() return 107892 end +for index, clip in ipairs(clips) do + local start_frame = 107892 + (index - 1) * 240 + clip.GetStart = function() return start_frame end + clip.GetEnd = function() return start_frame + 36 end +end +OpenBeat.run("timeline_markers", "raw") +local count = 0 +for _, marker in pairs(timeline.markers) do + if marker.customData == "OpenBeat:timeline:" .. source_path then count = count + 1 end +end +assert(count == 4, "Drop-frame settings must reach source selection and frame conversion") diff --git a/tests/test_analysis.py b/tests/test_analysis.py index d152128..39a30f2 100644 --- a/tests/test_analysis.py +++ b/tests/test_analysis.py @@ -4,6 +4,7 @@ from unittest import mock import soundfile as sf +import numpy as np from openbeat.analysis import cache_key, default_cache_dir, estimate_grid, quantized_beats from openbeat.artifacts import render_click_track @@ -20,6 +21,32 @@ def test_quantized_beats_extends_until_duration(self) -> None: beats = quantized_beats(120.0, 0.25, 1.3) self.assertEqual(beats, [0.25, 0.75, 1.25]) + def test_fractional_tempos_stay_aligned_over_three_minutes(self) -> None: + for tempo in (37.3, 120.5, 127.3, 180.25): + with self.subTest(tempo=tempo): + times = 0.2 + np.arange(int(180 * tempo / 60)) * (60 / tempo) + bpm, offset = estimate_grid(times.tolist()) + self.assertAlmostEqual(bpm, tempo, places=8) + self.assertAlmostEqual(offset, 0.2, places=8) + grid = np.asarray(quantized_beats(bpm, offset, float(times[-1]) + 0.01)) + self.assertLess(float(np.max(np.abs(grid - times))), 1e-8) + + def test_grid_handles_missing_beats_and_detection_jitter(self) -> None: + rng = np.random.default_rng(10) + numbers = np.asarray([i for i in range(360) if i % 7 != 3]) + times = 0.2 + numbers * (60 / 120.5) + rng.normal(0, 0.005, len(numbers)) + times[20] += 0.08 + bpm, offset = estimate_grid(times.tolist()) + self.assertLess(abs(bpm - 120.5), 0.01) + expected = 0.2 + numbers * (60 / 120.5) + fitted = offset + numbers * (60 / bpm) + self.assertLess(float(np.max(np.abs(fitted - expected))), 0.003) + + def test_grid_rejects_invalid_beat_times(self) -> None: + for beats in ([0.2, float("nan")], [0.2, float("inf")], [-1, 0], [1, 0], [0, 0]): + with self.subTest(beats=beats), self.assertRaises(ValueError): + estimate_grid(beats) + def test_render_click_track_defaults_to_stereo(self) -> None: with TemporaryDirectory() as tmpdir: output = Path(tmpdir) / "clicks.wav" diff --git a/tests/test_build_installers.py b/tests/test_build_installers.py index 323d6fb..bb0ffc2 100644 --- a/tests/test_build_installers.py +++ b/tests/test_build_installers.py @@ -7,6 +7,34 @@ class BuildInstallerTests(unittest.TestCase): + def test_windows_installer_packages_its_source_and_complete_runtime(self) -> None: + with TemporaryDirectory() as temporary: + root = Path(temporary) + runtime = root / "openbeat" + runtime.mkdir() + binary = runtime / "openbeat.exe" + binary.write_bytes(b"runtime") + (runtime / "_internal").mkdir() + (runtime / "_internal" / "dependency.dll").write_bytes(b"dependency") + installers = root / "installers" + installers.mkdir() + + def fake_run(cmd: list[str], cwd: Path | None = None) -> None: + self.assertIsNotNone(cwd) + entry = cwd / "windows_installer.py" + self.assertEqual(entry.read_bytes(), (build_installers.ROOT / "scripts" / "windows_installer.py").read_bytes()) + payload = cwd / "payload" + self.assertEqual((payload / "bin" / "openbeat" / "_internal" / "dependency.dll").read_bytes(), b"dependency") + for name in ("OpenBeatTiming.lua", "OpenBeatSource.lua", "OpenBeatMarkers.lua", "OpenBeatSubtitles.lua"): + self.assertTrue((payload / "Modules" / "OpenBeat" / name).is_file()) + (cwd / "dist").mkdir() + (cwd / "dist" / "openbeat-installer.exe").write_bytes(b"installer") + + with mock.patch.object(build_installers, "DIST_INSTALLERS", installers): + with mock.patch.object(build_installers, "run", side_effect=fake_run): + result = build_installers.create_windows_exe(binary, "0.0.2") + self.assertEqual(result.read_bytes(), b"installer") + def test_windows_cli_runtime_is_onedir_and_windowed(self) -> None: with TemporaryDirectory() as tmpdir: dist_root = Path(tmpdir) / "dist" diff --git a/tests/test_cache.py b/tests/test_cache.py new file mode 100644 index 0000000..25ad374 --- /dev/null +++ b/tests/test_cache.py @@ -0,0 +1,86 @@ +import json +import os +import unittest +from dataclasses import asdict +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import mock + +from openbeat.analysis import BeatAnalysis, cache_path, load_or_analyze + + +class CacheTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.audio = self.root / "song.wav" + self.audio.write_bytes(b"audio is mocked in cache tests") + self.cache_dir = self.root / "cache" + self.target = cache_path(str(self.audio), self.cache_dir) + self.analysis = BeatAnalysis(str(self.audio.resolve()), 2.0, 22050, [0.1, 0.6, 1.1], 120.0, 0.1, [0.1, 0.6, 1.1, 1.6]) + + def test_valid_cache_avoids_reanalysis(self) -> None: + self.target.write_text(json.dumps(asdict(self.analysis)), encoding="utf-8") + with mock.patch("openbeat.analysis.analyze_audio") as analyze: + self.assertEqual(load_or_analyze(str(self.audio), self.cache_dir), self.analysis) + analyze.assert_not_called() + + def test_invalid_cache_is_reanalyzed_and_replaced(self) -> None: + invalid = ["{", "[]", "{}"] + for field, value in ( + ("raw_beats", [float("nan")]), ("raw_beats", [0.6, 0.1]), + ("raw_beats", []), ("raw_beats", [3]), ("raw_beats", [True]), + ("duration_seconds", -1), ("tempo_bpm", 0), ("tempo_bpm", float("inf")), + ("sample_rate", "22050"), ("sample_rate", 44100), ("audio_path", "other.wav"), + ("grid_offset_seconds", "0.1"), ("quantized_beats", [0.1, 0.8]), + ("tempo_bpm", 10 ** 1000), ("quantized_beats", [0.1, 0.6]), + ): + payload = asdict(self.analysis) + payload[field] = value + invalid.append(json.dumps(payload)) + for payload in invalid: + with self.subTest(payload=payload): + self.target.write_text(payload, encoding="utf-8") + with mock.patch("openbeat.analysis.analyze_audio", return_value=self.analysis) as analyze: + self.assertEqual(load_or_analyze(str(self.audio), self.cache_dir), self.analysis) + analyze.assert_called_once_with(str(self.audio.resolve()), target_sr=22050) + self.assertEqual(json.loads(self.target.read_text()), asdict(self.analysis)) + self.assertEqual(list(self.cache_dir.glob("*.tmp")), []) + + def test_cache_directory_failure_does_not_block_analysis(self) -> None: + with mock.patch.object(Path, "mkdir", side_effect=PermissionError("read-only cache")): + with mock.patch("openbeat.analysis.analyze_audio", return_value=self.analysis): + self.assertEqual(load_or_analyze(str(self.audio), self.cache_dir), self.analysis) + + def test_failed_atomic_replace_preserves_old_file_and_cleans_temporary(self) -> None: + self.target.write_text("old incomplete cache", encoding="utf-8") + with mock.patch("openbeat.analysis.os.replace", side_effect=OSError("disk failure")): + with mock.patch("openbeat.analysis.analyze_audio", return_value=self.analysis): + self.assertEqual(load_or_analyze(str(self.audio), self.cache_dir), self.analysis) + self.assertEqual(self.target.read_text(), "old incomplete cache") + self.assertEqual(list(self.cache_dir.glob("*.tmp")), []) + + def test_final_cache_is_replaced_only_after_complete_json_is_written(self) -> None: + self.target.write_text("old incomplete cache", encoding="utf-8") + replace = os.replace + + def check_replace(source: Path, target: Path) -> None: + self.assertEqual(target.read_text(), "old incomplete cache") + self.assertEqual(json.loads(source.read_text()), asdict(self.analysis)) + replace(source, target) + + with mock.patch("openbeat.analysis.os.replace", side_effect=check_replace): + with mock.patch("openbeat.analysis.analyze_audio", return_value=self.analysis): + load_or_analyze(str(self.audio), self.cache_dir) + self.assertEqual(json.loads(self.target.read_text()), asdict(self.analysis)) + + def test_missing_source_is_still_reported(self) -> None: + with mock.patch("openbeat.analysis.analyze_audio") as analyze: + with self.assertRaises(FileNotFoundError): + load_or_analyze(str(self.root / "missing.wav"), self.cache_dir) + analyze.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cli.py b/tests/test_cli.py index 9c27a73..38ba834 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3,11 +3,16 @@ import unittest from pathlib import Path +from scripts.smoke_test_runtime import verify_runtime + ROOT = Path(__file__).resolve().parents[1] class CliTests(unittest.TestCase): + def test_cli_analyzes_real_audio_and_renders_both_modes(self) -> None: + verify_runtime([sys.executable, "-m", "openbeat.cli"]) + def test_cli_script_runs_as_top_level_entrypoint(self) -> None: cli_path = ROOT / "openbeat" / "cli.py" previous_argv = sys.argv[:] diff --git a/tests/test_resolve.py b/tests/test_resolve.py new file mode 100644 index 0000000..c68182b --- /dev/null +++ b/tests/test_resolve.py @@ -0,0 +1,24 @@ +import shutil +import subprocess +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + + +ROOT = Path(__file__).resolve().parents[1] +LUA = shutil.which("lua5.1") or shutil.which("lua") or shutil.which("lua5.4") + + +@unittest.skipUnless(LUA, "Lua is required to test the Resolve scripts") +class ResolveTests(unittest.TestCase): + def test_lua_behavior(self) -> None: + for script in sorted((ROOT / "tests" / "lua").glob("test_*.lua")): + with self.subTest(script=script.name), TemporaryDirectory(prefix="openbeat-lua-") as temporary: + result = subprocess.run( + [LUA, str(script), temporary], cwd=ROOT, text=True, capture_output=True, timeout=30 + ) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_transport.py b/tests/test_transport.py new file mode 100644 index 0000000..45939ed --- /dev/null +++ b/tests/test_transport.py @@ -0,0 +1,35 @@ +import shutil +import subprocess +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + +from openbeat.analysis import BeatAnalysis, to_lua, write_lua_analysis + + +LUA = shutil.which("lua5.1") or shutil.which("lua") or shutil.which("lua5.4") + + +class TransportTests(unittest.TestCase): + @unittest.skipUnless(LUA, "Lua is required to verify serialization") + def test_lua_round_trip_preserves_unicode_quotes_and_control_characters(self) -> None: + source_path = 'music 音楽/quote" slash\\\n\r\t\x00é.wav' + analysis = BeatAnalysis(source_path, 2, 22050, [0.1, 0.6], 120, 0.1, [0.1, 0.6, 1.1, 1.6]) + with TemporaryDirectory() as temporary: + path = Path(temporary) / "analysis.lua" + write_lua_analysis(analysis, str(path)) + result = subprocess.run( + [LUA, "-", str(path)], input=b"local data = dofile(arg[1]); io.write(data.audio_path)", + capture_output=True, timeout=30, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.decode("utf-8"), source_path) + + def test_nonfinite_numbers_are_rejected(self) -> None: + for value in (float("nan"), float("inf"), -float("inf")): + with self.subTest(value=value), self.assertRaises(ValueError): + to_lua(value) + + +if __name__ == "__main__": + unittest.main()