kopia lustrzana https://github.com/corrscope/corrscope
Merge pull request #49 from nyanpasu64/fix-cli-dirs
Fix bugs when dumping/loading YAML in another directorypull/357/head
nyanpasu64
GitHub
commit
cc13f3c50c
|
|
@ -1,3 +1,4 @@
|
|||
from os.path import abspath
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from ovgenpy.config import register_config, Alias
|
||||
|
|
@ -42,7 +43,7 @@ class Channel:
|
|||
|
||||
# Create a Wave object.
|
||||
wcfg = _WaveConfig(amplification=ovgen_cfg.amplification * cfg.ampl_ratio)
|
||||
self.wave = Wave(wcfg, cfg.wav_path)
|
||||
self.wave = Wave(wcfg, abspath(cfg.wav_path))
|
||||
|
||||
# Compute subsampling (array stride).
|
||||
tw = coalesce(cfg.trigger_width, ovgen_cfg.trigger_width)
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ PROFILE_DUMP_NAME = 'cprofile'
|
|||
'Config: Output video path')
|
||||
# Disables GUI
|
||||
@click.option('--write', '-w', is_flag=True, help=
|
||||
"Write config YAML file to path (don't open GUI).")
|
||||
"Write config YAML file to current directory (don't open GUI).")
|
||||
@click.option('--play', '-p', is_flag=True, help=
|
||||
"Preview or render (don't open GUI).")
|
||||
# Debugging
|
||||
|
|
@ -86,6 +86,7 @@ def main(
|
|||
|
||||
# Create cfg: Config object.
|
||||
cfg: Config = None
|
||||
cfg_dir: str = None
|
||||
|
||||
wav_list: List[Path] = []
|
||||
for name in files:
|
||||
|
|
@ -106,6 +107,7 @@ def main(
|
|||
raise click.ClickException(
|
||||
f'When supplying config {path}, you cannot supply other files/folders')
|
||||
cfg = yaml.load(path)
|
||||
cfg_dir = path.parent
|
||||
break
|
||||
|
||||
else:
|
||||
|
|
@ -136,6 +138,7 @@ def main(
|
|||
# amplification...render=default,
|
||||
outputs=outputs
|
||||
)
|
||||
cfg_dir = '.'
|
||||
|
||||
if show_gui:
|
||||
raise OvgenError('GUI not implemented')
|
||||
|
|
@ -144,11 +147,13 @@ def main(
|
|||
raise click.ClickException('Must specify files or folders to play')
|
||||
if write:
|
||||
if audio:
|
||||
write_path = Path(audio).with_suffix(YAML_NAME)
|
||||
# Write file to current working dir, not audio dir.
|
||||
audio_name = Path(audio).name
|
||||
# Add .yaml extension
|
||||
write_path = Path(audio_name).with_suffix(YAML_NAME)
|
||||
else:
|
||||
write_path = DEFAULT_CONFIG_PATH
|
||||
|
||||
# TODO test writing YAML file
|
||||
yaml.dump(cfg, write_path)
|
||||
|
||||
if play:
|
||||
|
|
@ -169,4 +174,4 @@ def main(
|
|||
cProfile.runctx('Ovgen(cfg).play()', globals(), locals(), path)
|
||||
|
||||
else:
|
||||
Ovgen(cfg).play()
|
||||
Ovgen(cfg, cfg_dir).play()
|
||||
|
|
|
|||
|
|
@ -2,7 +2,8 @@
|
|||
import shlex
|
||||
import subprocess
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import TYPE_CHECKING, Type, List, Union
|
||||
from os.path import abspath
|
||||
from typing import TYPE_CHECKING, Type, List, Union, Optional
|
||||
|
||||
from ovgenpy.config import register_config
|
||||
|
||||
|
|
@ -55,7 +56,7 @@ class _FFmpegCommand:
|
|||
|
||||
self.templates += ffmpeg_input_video(ovgen_cfg) # video
|
||||
if ovgen_cfg.master_audio:
|
||||
audio_path = shlex.quote(ovgen_cfg.master_audio)
|
||||
audio_path = shlex.quote(abspath(ovgen_cfg.master_audio))
|
||||
self.templates.append(f'-ss {ovgen_cfg.begin_time}')
|
||||
self.templates += ffmpeg_input_audio(audio_path) # audio
|
||||
|
||||
|
|
@ -123,7 +124,8 @@ class PipeOutput(Output):
|
|||
|
||||
@register_config
|
||||
class FFmpegOutputConfig(IOutputConfig):
|
||||
path: str
|
||||
# path=None writes to stdout.
|
||||
path: Optional[str]
|
||||
args: str = ''
|
||||
|
||||
# Do not use `-movflags faststart`, I get corrupted mp4 files (missing MOOV)
|
||||
|
|
@ -141,7 +143,13 @@ class FFmpegOutput(PipeOutput):
|
|||
ffmpeg = _FFmpegCommand([FFMPEG, '-y'], ovgen_cfg)
|
||||
ffmpeg.add_output(cfg)
|
||||
ffmpeg.templates.append(cfg.args)
|
||||
self.open(ffmpeg.popen([cfg.path], self.bufsize))
|
||||
|
||||
if cfg.path is None:
|
||||
video_path = '-' # Write to stdout
|
||||
else:
|
||||
video_path = abspath(cfg.path)
|
||||
|
||||
self.open(ffmpeg.popen([video_path], self.bufsize))
|
||||
|
||||
|
||||
# FFplayOutput
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ from ovgenpy.channel import Channel, ChannelConfig
|
|||
from ovgenpy.config import register_config, register_enum, Ignored
|
||||
from ovgenpy.renderer import MatplotlibRenderer, RendererConfig, LayoutConfig
|
||||
from ovgenpy.triggers import ITriggerConfig, CorrelationTriggerConfig, Trigger
|
||||
from ovgenpy.util import pushd
|
||||
from ovgenpy.utils import keyword_dataclasses as dc
|
||||
from ovgenpy.utils.keyword_dataclasses import field
|
||||
from ovgenpy.wave import Wave
|
||||
|
|
@ -97,8 +98,9 @@ def default_config(**kwargs):
|
|||
|
||||
|
||||
class Ovgen:
|
||||
def __init__(self, cfg: Config):
|
||||
def __init__(self, cfg: Config, cfg_dir: str):
|
||||
self.cfg = cfg
|
||||
self.cfg_dir = cfg_dir
|
||||
self.has_played = False
|
||||
|
||||
if len(self.cfg.channels) == 0:
|
||||
|
|
@ -110,19 +112,21 @@ class Ovgen:
|
|||
nchan: int
|
||||
|
||||
def _load_channels(self):
|
||||
self.channels = [Channel(ccfg, self.cfg) for ccfg in self.cfg.channels]
|
||||
self.waves = [channel.wave for channel in self.channels]
|
||||
self.triggers = [channel.trigger for channel in self.channels]
|
||||
self.nchan = len(self.channels)
|
||||
with pushd(self.cfg_dir):
|
||||
self.channels = [Channel(ccfg, self.cfg) for ccfg in self.cfg.channels]
|
||||
self.waves = [channel.wave for channel in self.channels]
|
||||
self.triggers = [channel.trigger for channel in self.channels]
|
||||
self.nchan = len(self.channels)
|
||||
|
||||
@contextmanager
|
||||
def _load_outputs(self):
|
||||
with ExitStack() as stack:
|
||||
self.outputs = [
|
||||
stack.enter_context(output_cfg(self.cfg))
|
||||
for output_cfg in self.cfg.outputs
|
||||
]
|
||||
yield
|
||||
with pushd(self.cfg_dir):
|
||||
with ExitStack() as stack:
|
||||
self.outputs = [
|
||||
stack.enter_context(output_cfg(self.cfg))
|
||||
for output_cfg in self.cfg.outputs
|
||||
]
|
||||
yield
|
||||
|
||||
def _load_renderer(self):
|
||||
renderer = MatplotlibRenderer(self.cfg.render, self.cfg.layout, self.nchan)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
from functools import wraps
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
from itertools import chain
|
||||
from typing import Callable, Tuple, TypeVar, Iterator
|
||||
from pathlib import Path
|
||||
from typing import Callable, Tuple, TypeVar, Iterator, Union
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
|
@ -80,19 +82,12 @@ def find(a: 'np.ndarray[T]', predicate: 'Callable[[np.ndarray[T]], np.ndarray[bo
|
|||
i0 = i1
|
||||
|
||||
|
||||
# Adapted from https://stackoverflow.com/a/9458386
|
||||
def curry(x, argc=None):
|
||||
if argc is None:
|
||||
argc = x.__code__.co_argcount
|
||||
@contextmanager
|
||||
def pushd(new_dir: Union[Path, str]):
|
||||
previous_dir = os.getcwd()
|
||||
os.chdir(str(new_dir))
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
os.chdir(previous_dir)
|
||||
|
||||
@wraps(x)
|
||||
def p(*a):
|
||||
if len(a) == argc:
|
||||
return x(*a)
|
||||
|
||||
def q(*b):
|
||||
return x(*(a + b))
|
||||
|
||||
return curry(q, argc - len(a))
|
||||
|
||||
return p
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
import os
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest_mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def Popen(mocker: 'pytest_mock.MockFixture'):
|
||||
Popen = mocker.patch.object(subprocess, 'Popen', autospec=True)
|
||||
popen = Popen.return_value
|
||||
|
||||
popen.stdin = open(os.devnull, "wb")
|
||||
popen.stdout = open(os.devnull, "rb")
|
||||
popen.wait.return_value = 0
|
||||
|
||||
yield Popen
|
||||
|
|
@ -64,7 +64,7 @@ def test_channel_subsampling(
|
|||
assert trigger._subsampling == channel.trigger_subsampling
|
||||
|
||||
# Ensure ovgenpy calls render using channel.window_samp and render_subsampling.
|
||||
ovgen = Ovgen(cfg)
|
||||
ovgen = Ovgen(cfg, '.')
|
||||
renderer = mocker.patch.object(Ovgen, '_load_renderer').return_value
|
||||
ovgen.play()
|
||||
|
||||
|
|
|
|||
|
|
@ -1,56 +1,65 @@
|
|||
import shlex
|
||||
from os.path import abspath
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Callable
|
||||
|
||||
import click
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import ovgenpy.channel
|
||||
from ovgenpy import cli
|
||||
from ovgenpy.cli import YAML_NAME
|
||||
from ovgenpy.config import yaml
|
||||
from ovgenpy.ovgenpy import Config
|
||||
from ovgenpy.util import curry
|
||||
from ovgenpy.ovgenpy import Config, Ovgen
|
||||
from ovgenpy.util import pushd
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest_mock
|
||||
|
||||
|
||||
def call_main(args):
|
||||
return CliRunner().invoke(cli.main, args, catch_exceptions=False, standalone_mode=False)
|
||||
def call_main(argv):
|
||||
return CliRunner().invoke(cli.main, argv, catch_exceptions=False, standalone_mode=False)
|
||||
|
||||
|
||||
# ovgenpy configuration sinks
|
||||
|
||||
@pytest.fixture
|
||||
@curry
|
||||
def yaml_sink(mocker: 'pytest_mock.MockFixture', command):
|
||||
dump = mocker.patch.object(yaml, 'dump')
|
||||
def yaml_sink(mocker: 'pytest_mock.MockFixture') -> Callable:
|
||||
def _yaml_sink(command):
|
||||
dump = mocker.patch.object(yaml, 'dump')
|
||||
|
||||
args = shlex.split(command) + ['-w']
|
||||
call_main(args)
|
||||
argv = shlex.split(command) + ['-w']
|
||||
call_main(argv)
|
||||
|
||||
dump.assert_called_once()
|
||||
args, kwargs = dump.call_args
|
||||
dump.assert_called_once()
|
||||
(cfg, stream), kwargs = dump.call_args
|
||||
|
||||
cfg = args[0] # yaml.dump(cfg, out)
|
||||
assert isinstance(cfg, Config)
|
||||
return cfg
|
||||
assert isinstance(cfg, Config)
|
||||
return cfg, stream
|
||||
return _yaml_sink
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@curry
|
||||
def player_sink(mocker, command):
|
||||
Ovgen = mocker.patch.object(cli, 'Ovgen')
|
||||
def player_sink(mocker) -> Callable:
|
||||
def _player_sink(command):
|
||||
Ovgen = mocker.patch.object(cli, 'Ovgen')
|
||||
|
||||
args = shlex.split(command) + ['-p']
|
||||
call_main(args)
|
||||
argv = shlex.split(command) + ['-p']
|
||||
call_main(argv)
|
||||
|
||||
Ovgen.assert_called_once()
|
||||
args, kwargs = Ovgen.call_args
|
||||
Ovgen.assert_called_once()
|
||||
args, kwargs = Ovgen.call_args
|
||||
cfg = args[0]
|
||||
|
||||
cfg = args[0] # Ovgen(cfg)
|
||||
assert isinstance(cfg, Config)
|
||||
return cfg
|
||||
assert isinstance(cfg, Config)
|
||||
return cfg,
|
||||
return _player_sink
|
||||
|
||||
|
||||
def test_sink_fixture(yaml_sink, player_sink):
|
||||
""" Ensure we can use yaml_sink and player_sink as a fixture directly """
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(params=[yaml_sink, player_sink])
|
||||
|
|
@ -67,11 +76,69 @@ def test_no_files(any_sink):
|
|||
|
||||
|
||||
@pytest.mark.parametrize('wav_dir', '. tests'.split())
|
||||
def test_cwd(any_sink, wav_dir):
|
||||
def test_file_dirs(any_sink, wav_dir):
|
||||
""" Ensure loading files from `dir` places `dir/*.wav` in config. """
|
||||
wavs = Path(wav_dir).glob('*.wav')
|
||||
wavs = sorted(str(x) for x in wavs)
|
||||
|
||||
cfg = any_sink(wav_dir)
|
||||
cfg = any_sink(wav_dir)[0]
|
||||
assert isinstance(cfg, Config)
|
||||
|
||||
assert [chan.wav_path for chan in cfg.channels] == wavs
|
||||
|
||||
|
||||
def test_write_dir(yaml_sink):
|
||||
""" Loading `--audio another/dir` should write YAML to current dir.
|
||||
Writing YAML to audio dir: causes relative paths (relative to pwd) to break. """
|
||||
|
||||
audio_path = Path('tests/sine440.wav')
|
||||
arg_str = f'tests -a {audio_path}'
|
||||
|
||||
cfg, outpath = yaml_sink(arg_str) # type: Config, Path
|
||||
assert isinstance(outpath, Path)
|
||||
|
||||
# Ensure YAML config written to current dir.
|
||||
assert outpath.parent == Path()
|
||||
assert outpath.name == str(outpath)
|
||||
assert str(outpath) == audio_path.with_suffix(YAML_NAME).name
|
||||
|
||||
# Ensure config paths are valid.
|
||||
assert outpath.parent / cfg.master_audio == audio_path
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('Popen')
|
||||
def test_load_yaml_another_dir(yaml_sink, mocker, Popen):
|
||||
""" YAML file located in `another/dir` should resolve `master_audio`, `channels[].
|
||||
wav_path`, and video `path` from `another/dir`. """
|
||||
|
||||
subdir = 'tests'
|
||||
wav = 'sine440.wav'
|
||||
mp4 = 'foo.mp4'
|
||||
with pushd(subdir):
|
||||
arg_str = f'{wav} -a {wav} -o {mp4}'
|
||||
cfg, outpath = yaml_sink(arg_str) # type: Config, Path
|
||||
|
||||
cfg.begin_time = 100 # To skip all actual rendering
|
||||
|
||||
# Log execution of Ovgen().play()
|
||||
Wave = mocker.spy(ovgenpy.channel, 'Wave')
|
||||
ovgen = Ovgen(cfg, subdir)
|
||||
ovgen.play()
|
||||
|
||||
# Compute absolute paths
|
||||
wav_abs = abspath(f'{subdir}/{wav}')
|
||||
mp4_abs = abspath(f'{subdir}/{mp4}')
|
||||
|
||||
# Test `wave_path`
|
||||
args, kwargs = Wave.call_args
|
||||
cfg, wave_path = args
|
||||
assert wave_path == wav_abs
|
||||
|
||||
# Test output `master_audio` and video `path`
|
||||
args, kwargs = Popen.call_args
|
||||
argv = args[0]
|
||||
assert argv[-1] == mp4_abs
|
||||
assert f'-i {wav_abs}' in ' '.join(argv)
|
||||
|
||||
|
||||
# TODO integration test without --audio
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
|
@ -15,14 +14,13 @@ if TYPE_CHECKING:
|
|||
import pytest_mock
|
||||
|
||||
CFG = default_config(render=RendererConfig(WIDTH, HEIGHT))
|
||||
STDOUT_CFG = FFmpegOutputConfig('-', '-f nut')
|
||||
NULL_CFG = FFmpegOutputConfig(None, '-f null')
|
||||
|
||||
|
||||
def test_render_output():
|
||||
""" Ensure rendering to output does not raise exceptions. """
|
||||
renderer = MatplotlibRenderer(CFG.render, CFG.layout, nplots=1)
|
||||
output_cfg = FFmpegOutputConfig('-', '-f nut')
|
||||
out = FFmpegOutput(CFG, output_cfg)
|
||||
out: FFmpegOutput = NULL_CFG(CFG)
|
||||
|
||||
renderer.render_frame([ALL_ZEROS])
|
||||
out.write_frame(renderer.get_frame())
|
||||
|
|
@ -31,17 +29,20 @@ def test_render_output():
|
|||
|
||||
|
||||
def test_output():
|
||||
out = FFmpegOutput(CFG, STDOUT_CFG)
|
||||
out: FFmpegOutput = NULL_CFG(CFG)
|
||||
|
||||
frame = bytes(WIDTH * HEIGHT * RGB_DEPTH)
|
||||
out.write_frame(frame)
|
||||
|
||||
assert out.close() == 0
|
||||
# Ensure video is written to stdout, and not current directory.
|
||||
assert not Path('-').exists()
|
||||
|
||||
|
||||
# Ensure ovgen terminates FFplay upon exceptions.
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('Popen')
|
||||
def test_terminate_ffplay(Popen):
|
||||
""" FFplayOutput unit test: Ensure ffmpeg and ffplay are terminated when Python
|
||||
exceptions occur.
|
||||
|
|
@ -58,6 +59,7 @@ def test_terminate_ffplay(Popen):
|
|||
popen.terminate.assert_called()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('Popen')
|
||||
def test_ovgen_terminate_ffplay(Popen, mocker: 'pytest_mock.MockFixture'):
|
||||
""" Integration test: Ensure ffmpeg and ffplay are terminated when Python exceptions
|
||||
occur. """
|
||||
|
|
@ -67,7 +69,7 @@ def test_ovgen_terminate_ffplay(Popen, mocker: 'pytest_mock.MockFixture'):
|
|||
master_audio='tests/sine440.wav',
|
||||
outputs=[FFplayOutputConfig()]
|
||||
)
|
||||
ovgen = Ovgen(cfg)
|
||||
ovgen = Ovgen(cfg, '.')
|
||||
|
||||
render_frame = mocker.patch.object(MatplotlibRenderer, 'render_frame')
|
||||
render_frame.side_effect = DummyException()
|
||||
|
|
@ -86,16 +88,5 @@ def test_ovgen_terminate_ffplay(Popen, mocker: 'pytest_mock.MockFixture'):
|
|||
# TODO integration test on ???
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def Popen(mocker: 'pytest_mock.MockFixture'):
|
||||
Popen = mocker.patch.object(subprocess, 'Popen', autospec=True).return_value
|
||||
|
||||
Popen.stdin = open(os.devnull, "wb")
|
||||
Popen.stdout = open(os.devnull, "rb")
|
||||
Popen.wait.return_value = 0
|
||||
|
||||
yield Popen
|
||||
|
||||
|
||||
class DummyException(Exception):
|
||||
pass
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue