kopia lustrzana https://github.com/corrscope/corrscope
nyanpasu64
GitHub
commit
b5c49e5649
|
|
@ -7,6 +7,7 @@
|
|||
<list>
|
||||
<option value="E302" />
|
||||
<option value="E203" />
|
||||
<option value="E128" />
|
||||
</list>
|
||||
</option>
|
||||
</inspection_tool>
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ from ovgenpy.ovgenpy import default_config, Config, Ovgen
|
|||
|
||||
Folder = click.Path(exists=True, file_okay=False)
|
||||
File = click.Path(exists=True, dir_okay=False)
|
||||
OutFile = click.Path(dir_okay=False)
|
||||
|
||||
|
||||
# https://github.com/pallets/click/issues/473
|
||||
|
|
@ -26,7 +27,12 @@ File = click.Path(exists=True, dir_okay=False)
|
|||
# name = possible_names[-1][1].replace('-', '_').lower()
|
||||
|
||||
|
||||
# List of recognized Config file extensions.
|
||||
YAML_EXTS = ['.yaml']
|
||||
# Default extension when writing Config.
|
||||
YAML_NAME = YAML_EXTS[0]
|
||||
DEFAULT_CONFIG_PATH = Path('ovgenpy').with_suffix(YAML_NAME)
|
||||
|
||||
PROFILE_DUMP_NAME = 'cprofile'
|
||||
|
||||
|
||||
|
|
@ -35,25 +41,25 @@ PROFILE_DUMP_NAME = 'cprofile'
|
|||
@click.argument('files', nargs=-1)
|
||||
# Override default .yaml settings (only if YAML file not supplied)
|
||||
# Incorrect [option] name order: https://github.com/pallets/click/issues/793
|
||||
@click.option('--audio', '-a', type=File,
|
||||
help='Config: Input path for master audio file')
|
||||
@click.option('--video-output', '-o', type=click.Path(dir_okay=False),
|
||||
help='Config: Output video path')
|
||||
@click.option('--audio', '-a', type=File, help=
|
||||
'Config: Input path for master audio file')
|
||||
@click.option('--video-output', '-o', type=OutFile, help=
|
||||
'Config: Output video path')
|
||||
# Disables GUI
|
||||
@click.option('--write-cfg', '-w', nargs=1, type=click.Path(dir_okay=False),
|
||||
help="Write config YAML file to path (don't open GUI).")
|
||||
@click.option('--play', '-p', is_flag=True,
|
||||
help="Preview or render (don't open GUI).")
|
||||
@click.option('--write', '-w', is_flag=True, help=
|
||||
"Write config YAML file to path (don't open GUI).")
|
||||
@click.option('--play', '-p', is_flag=True, help=
|
||||
"Preview or render (don't open GUI).")
|
||||
# Debugging
|
||||
@click.option('--profile', is_flag=True,
|
||||
help='Debug: Write CProfiler snapshot')
|
||||
@click.option('--profile', is_flag=True, help=
|
||||
'Debug: Write CProfiler snapshot')
|
||||
def main(
|
||||
files: Tuple[str],
|
||||
# cfg
|
||||
audio: Optional[str],
|
||||
video_output: Optional[str],
|
||||
# gui
|
||||
write_cfg: Optional[str],
|
||||
write: bool,
|
||||
play: bool,
|
||||
profile: bool,
|
||||
):
|
||||
|
|
@ -76,21 +82,20 @@ def main(
|
|||
# - You can specify as many wildcards or wav files as you want.
|
||||
# - You can only supply one folder, with no files/wildcards.
|
||||
|
||||
show_gui = (not write_cfg and not play)
|
||||
show_gui = (not write and not play)
|
||||
|
||||
# Create cfg: Config object.
|
||||
cfg: Config = None
|
||||
|
||||
wav_prefix = Path()
|
||||
wav_list: List[Path] = []
|
||||
for name in files:
|
||||
path = Path(name)
|
||||
if path.is_dir():
|
||||
# Add a directory.
|
||||
if len(files) > 1:
|
||||
# Warning is technically optional, since wav_prefix has been removed.
|
||||
raise click.ClickException(
|
||||
f'When supplying folder {path}, you cannot supply other files/folders')
|
||||
wav_prefix = path
|
||||
matches = sorted(path.glob('*.wav'))
|
||||
wav_list += matches
|
||||
break
|
||||
|
|
@ -110,11 +115,10 @@ def main(
|
|||
matches = [path]
|
||||
if not path.exists():
|
||||
raise click.ClickException(
|
||||
f'Supplied nonexistent file or wildcard {path}')
|
||||
f'Supplied nonexistent file or wildcard: {path}')
|
||||
wav_list += matches
|
||||
|
||||
if not cfg:
|
||||
wav_prefix = str(wav_prefix)
|
||||
wav_list = [str(wav_path) for wav_path in wav_list]
|
||||
|
||||
channels = [ChannelConfig(wav_path) for wav_path in wav_list]
|
||||
|
|
@ -124,13 +128,9 @@ def main(
|
|||
else:
|
||||
outputs = [FFplayOutputConfig()]
|
||||
|
||||
# TODO test cfg, ensure wav_prefix and wav_list are correct
|
||||
# maybe I should use a list comprehension to parse cfg.channels to List[str].
|
||||
|
||||
cfg = default_config(
|
||||
master_audio=audio,
|
||||
# fps=default,
|
||||
wav_prefix=wav_prefix,
|
||||
channels=channels,
|
||||
# width_ms...trigger=default,
|
||||
# amplification...render=default,
|
||||
|
|
@ -140,9 +140,16 @@ def main(
|
|||
if show_gui:
|
||||
raise OvgenError('GUI not implemented')
|
||||
else:
|
||||
if write_cfg:
|
||||
if not files:
|
||||
raise click.ClickException('Must specify files or folders to play')
|
||||
if write:
|
||||
if audio:
|
||||
write_path = Path(audio).with_suffix(YAML_NAME)
|
||||
else:
|
||||
write_path = DEFAULT_CONFIG_PATH
|
||||
|
||||
# TODO test writing YAML file
|
||||
yaml.dump(cfg, Path(write_cfg))
|
||||
yaml.dump(cfg, write_path)
|
||||
|
||||
if play:
|
||||
if profile:
|
||||
|
|
|
|||
|
|
@ -26,13 +26,12 @@ class BenchmarkMode(IntEnum):
|
|||
OUTPUT = 3
|
||||
|
||||
|
||||
@register_config(always_dump='begin_time wave_prefix')
|
||||
@register_config(always_dump='begin_time')
|
||||
class Config:
|
||||
master_audio: Optional[str]
|
||||
fps: int
|
||||
begin_time: float = 0
|
||||
|
||||
wav_prefix: str = '' # if wave/glob..., pwd. if folder, folder.
|
||||
channels: List[ChannelConfig] = field(default_factory=list)
|
||||
|
||||
width_ms: int
|
||||
|
|
@ -69,7 +68,6 @@ def default_config(**kwargs):
|
|||
fps=_FPS,
|
||||
# begin_time=0,
|
||||
|
||||
# wav_prefix='',
|
||||
channels=[],
|
||||
|
||||
width_ms=25,
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ from ovgenpy.wave import FLOAT
|
|||
if TYPE_CHECKING:
|
||||
from ovgenpy.wave import Wave
|
||||
|
||||
# Abstract classes
|
||||
|
||||
class ITriggerConfig:
|
||||
cls: Type['Trigger']
|
||||
|
|
@ -48,9 +49,7 @@ class Trigger(ABC):
|
|||
...
|
||||
|
||||
|
||||
def lerp(x: np.ndarray, y: np.ndarray, a: float):
|
||||
return x * (1 - a) + y * a
|
||||
|
||||
# CorrelationTrigger
|
||||
|
||||
@register_config
|
||||
class CorrelationTriggerConfig(ITriggerConfig):
|
||||
|
|
@ -215,6 +214,12 @@ def get_period(data: np.ndarray) -> int:
|
|||
return peakX
|
||||
|
||||
|
||||
def lerp(x: np.ndarray, y: np.ndarray, a: float):
|
||||
return x * (1 - a) + y * a
|
||||
|
||||
|
||||
# ZeroCrossingTrigger
|
||||
|
||||
class ZeroCrossingTrigger(Trigger):
|
||||
# TODO support subsampling
|
||||
def get_trigger(self, index: int):
|
||||
|
|
@ -260,3 +265,16 @@ class ZeroCrossingTrigger(Trigger):
|
|||
|
||||
- To be consistent, we should increment zeros whenever we *start* there.
|
||||
"""
|
||||
|
||||
|
||||
# NullTrigger
|
||||
|
||||
@register_config
|
||||
class NullTriggerConfig(ITriggerConfig):
|
||||
pass
|
||||
|
||||
|
||||
@register_trigger(NullTriggerConfig)
|
||||
class NullTrigger(Trigger):
|
||||
def get_trigger(self, index: int) -> int:
|
||||
return index
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
from functools import wraps
|
||||
from itertools import chain
|
||||
from typing import Callable, Tuple, TypeVar, Iterator
|
||||
|
||||
|
|
@ -77,3 +78,21 @@ def find(a: 'np.ndarray[T]', predicate: 'Callable[[np.ndarray[T]], np.ndarray[bo
|
|||
for idx in predicate(chunk).nonzero()[0]:
|
||||
yield (idx + i0, ), chunk[idx]
|
||||
i0 = i1
|
||||
|
||||
|
||||
# Adapted from https://stackoverflow.com/a/9458386
|
||||
def curry(x, argc=None):
|
||||
if argc is None:
|
||||
argc = x.__code__.co_argcount
|
||||
|
||||
@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,77 @@
|
|||
import shlex
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import click
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from ovgenpy import cli
|
||||
from ovgenpy.config import yaml
|
||||
from ovgenpy.ovgenpy import Config
|
||||
from ovgenpy.util import curry
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest_mock
|
||||
|
||||
|
||||
def call_main(args):
|
||||
return CliRunner().invoke(cli.main, args, 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')
|
||||
|
||||
args = shlex.split(command) + ['-w']
|
||||
call_main(args)
|
||||
|
||||
dump.assert_called_once()
|
||||
args, kwargs = dump.call_args
|
||||
|
||||
cfg = args[0] # yaml.dump(cfg, out)
|
||||
assert isinstance(cfg, Config)
|
||||
return cfg
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@curry
|
||||
def player_sink(mocker, command):
|
||||
Ovgen = mocker.patch.object(cli, 'Ovgen')
|
||||
|
||||
args = shlex.split(command) + ['-p']
|
||||
call_main(args)
|
||||
|
||||
Ovgen.assert_called_once()
|
||||
args, kwargs = Ovgen.call_args
|
||||
|
||||
cfg = args[0] # Ovgen(cfg)
|
||||
assert isinstance(cfg, Config)
|
||||
return cfg
|
||||
|
||||
|
||||
@pytest.fixture(params=[yaml_sink, player_sink])
|
||||
def any_sink(request, mocker):
|
||||
sink = request.param
|
||||
return sink(mocker)
|
||||
|
||||
|
||||
# ovgenpy configuration sources
|
||||
|
||||
def test_no_files(any_sink):
|
||||
with pytest.raises(click.ClickException):
|
||||
any_sink('')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('wav_dir', '. tests'.split())
|
||||
def test_cwd(any_sink, wav_dir):
|
||||
wavs = Path(wav_dir).glob('*.wav')
|
||||
wavs = sorted(str(x) for x in wavs)
|
||||
|
||||
cfg = any_sink(wav_dir)
|
||||
assert isinstance(cfg, Config)
|
||||
|
||||
assert [chan.wav_path for chan in cfg.channels] == wavs
|
||||
Ładowanie…
Reference in New Issue