mirror
/
corrscope
kopia lustrzana https://github.com/corrscope/corrscope
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

Rewrite CLI and Ovgenpy() using Channel class

pull/357/head
nyanpasu64 2018-07-28 04:06:27 -07:00
rodzic 992702f317
commit 7304464560
4 zmienionych plików z 206 dodań i 86 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

4
ovgenpy/__main__.py
Wyświetl plik

@ -1,4 +1,4 @@
from ovgenpy import ovgenpy
from ovgenpy import cli
if __name__ == '__main__':
ovgenpy.main()
cli.main()

35
ovgenpy/channel.py
Wyświetl plik

@ -1,28 +1,47 @@
from typing import NamedTuple, TYPE_CHECKING, Any, Optional
from typing import TYPE_CHECKING, Any
from ovgenpy.config import register_config
from ovgenpy.wave import _WaveConfig, Wave
if TYPE_CHECKING:
from ovgenpy.triggers import Trigger
from ovgenpy.wave import Wave
from ovgenpy.triggers import ITriggerConfig
from ovgenpy.ovgenpy import Config
@register_config
class ChannelConfig:
_main_cfg: 'Config'
wav_path: str
trigger: 'ITriggerConfig' = None # TODO test channel-specific triggers
trigger_width_ratio: int = 1
render_width_ratio: int = 1
amplification: float = 1.0
ampl_ratio: float = 1.0 # TODO use amplification = None instead?
line_color: Any = None
background_color: Any = None
class Channel:
def __init__(self, cfg: ChannelConfig, wave: 'Wave', trigger: 'Trigger'):
def __init__(self, cfg: ChannelConfig, ovgen_cfg: 'Config'):
self.cfg = cfg
self.wave = wave
self.trigger = trigger
# Compute subsampling factors.
self.trigger_subsampling = ovgen_cfg.subsampling * cfg.trigger_width_ratio
self.render_subsampling = ovgen_cfg.subsampling * cfg.render_width_ratio
# Create a Wave object. (TODO maybe create in Ovgenpy()?)
wcfg = _WaveConfig(amplification=ovgen_cfg.amplification * cfg.ampl_ratio)
self.wave = Wave(wcfg, cfg.wav_path)
# Create a Trigger object.
tcfg = cfg.trigger or ovgen_cfg.trigger
trigger_nsamp = round(
ovgen_cfg.render_width_s * cfg.trigger_width_ratio * self.wave.smp_s
)
self.trigger = tcfg(
wave=self.wave,
scan_nsamp=trigger_nsamp, # TODO rename to trigger_nsamp
# FIXME self.trigger_subsampling
)

136
ovgenpy/cli.py 100644
Wyświetl plik

@ -0,0 +1,136 @@
from pathlib import Path
from typing import Optional, List, Tuple
import click
from ovgenpy.channel import ChannelConfig
from ovgenpy.config import OvgenError, yaml
from ovgenpy.outputs import FFmpegOutputConfig, FFplayOutputConfig
from ovgenpy.ovgenpy import default_config, Config, Ovgen
Folder = click.Path(exists=True, file_okay=False)
File = click.Path(exists=True, dir_okay=False)
# https://github.com/pallets/click/issues/473
# @platformio requires some functionality which doesn't work in Click 6.
# Click 6 is marked as stable, but http://click.pocoo.org/ redirects to /5/.
# wat
# If multiple `--` names are supplied to @click.option, the last one will be used.
# possible_names = [('-', 'w'), ('--', 'write')]
# possible_names.sort(key=lambda x: len(x[0]))
# name = possible_names[-1][1].replace('-', '_').lower()
YAML_EXTS = ['.yaml']
@click.command()
# Inputs
@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)
@click.option('--video-output', '-o', type=click.Path(dir_okay=False))
# Disables GUI
@click.option('--write-cfg', '-w', nargs=1, type=click.Path(dir_okay=False))
@click.option('--play', '-p', is_flag=True)
def main(
files: Tuple[str],
# cfg
audio: Optional[str],
video_output: Optional[str],
# gui
write_cfg: Optional[str],
play: bool,
):
"""
GUI:
ovgenpy
ovgenpy file.yaml
ovgenpy wildcard/wav/folder ... [--options]
CLI:
ovgenpy wildcard/wav/folder ... [--options] --write-cfg file.yaml [--play]
??? ovgenpy wildcard/wav/folder ... --play
ovgenpy file.yaml --play
ovgenpy file.yaml --write-yaml
- 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)
# 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:
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
elif path.suffix in YAML_EXTS:
# Load a YAML file to cfg, and skip default_config().
if len(files) > 1:
raise click.ClickException(
f'When supplying config {path}, you cannot supply other files/folders')
cfg = yaml.load(path)
break
else:
# Load one or more wav files.
matches = sorted(Path().glob(name))
if not matches:
matches = [path]
if not path.exists():
raise click.ClickException(
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]
if video_output:
outputs = [FFmpegOutputConfig(video_output)]
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,
outputs=outputs
)
if show_gui:
raise OvgenError('GUI not implemented')
else:
if write_cfg:
# TODO test writing YAML file
yaml.dump(cfg, Path(write_cfg))
if play:
Ovgen(cfg).play()
print(cfg)

117
ovgenpy/ovgenpy.py
Wyświetl plik

@ -1,19 +1,16 @@
# -*- coding: utf-8 -*-
import cgitb
import sys
import time
from pathlib import Path
from typing import Optional, List
import click
from ovgenpy import outputs
from ovgenpy.channel import Channel
from ovgenpy.channel import Channel, ChannelConfig
from ovgenpy.config import register_config, yaml
from ovgenpy.renderer import MatplotlibRenderer, RendererConfig
from ovgenpy.triggers import ITriggerConfig, CorrelationTriggerConfig
from ovgenpy.utils import keyword_dataclasses as dc
from ovgenpy.utils.keyword_dataclasses import field
from ovgenpy.wave import _WaveConfig, Wave
from ovgenpy.wave import Wave
# cgitb.enable(format='text')
@ -21,55 +18,51 @@ from ovgenpy.wave import _WaveConfig, Wave
RENDER_PROFILING = True
@register_config(always_dump='*')
@register_config(always_dump='wave_prefix')
class Config:
wave_prefix: str = '' # if wave/glob..., pwd. if folder, folder.
channels: List[Channel] = field(default_factory=lambda: [])
master_audio: Optional[str]
fps: int
time_visible_ms: int
scan_ratio: float
wav_prefix: str = '' # if wave/glob..., pwd. if folder, folder.
channels: List[ChannelConfig] = field(default_factory=list)
width_ms: int
subsampling: int
trigger: ITriggerConfig # Maybe overriden per Wave
amplification: float
render: RendererConfig
outputs: List[outputs.IOutputConfig]
create_window: bool
create_window: bool = False
@property
def time_visible_s(self) -> float:
return self.time_visible_ms / 1000
def render_width_s(self) -> float:
return self.width_ms / 1000
Folder = click.Path(exists=True, file_okay=False)
File = click.Path(exists=True, dir_okay=False)
_FPS = 60 # f_s
def main():
def default_config(**kwargs):
cfg = Config(
wave_prefix='foo',
master_audio='',
fps=_FPS,
wav_prefix='',
channels=[],
master_audio=None,
fps=69,
time_visible_ms=25,
scan_ratio=1,
width_ms=25,
subsampling=1,
trigger=CorrelationTriggerConfig(
trigger_strength=1,
use_edge_trigger=False,
trigger_strength=10,
use_edge_trigger=True,
responsiveness=1,
falloff_width=1,
responsiveness=0.1,
falloff_width=0.5,
),
amplification=5,
render=RendererConfig( # todo
amplification=1,
render=RendererConfig(
1280, 720,
ncols=1
),
@ -77,62 +70,33 @@ def main():
outputs=[
# outputs.FFmpegOutputConfig(output),
outputs.FFplayOutputConfig(),
],
create_window=False
]
)
yaml.dump(cfg, sys.stdout)
return
ovgen = Ovgen(cfg)
ovgen.write()
return dc.replace(cfg, **kwargs)
class Ovgen:
def __init__(self, cfg: Config):
self.cfg = cfg
self.waves: List[Wave] = []
self.channels: List[Channel] = []
self.outputs: List[outputs.Output] = []
self.nchan: int = None
def write(self):
self._load_waves() # self.waves =
self._load_channels() # self.waves =
self._load_outputs() # self.outputs =
self._render()
def _load_waves(self):
wave_dir = Path(self.cfg.wave_folder)
waves: List[Wave]
channels: List[Channel]
outputs: List[outputs.Output]
nchan: int
waves = sorted(wave_dir.glob('*.wav'))
for idx, path in enumerate(waves):
wcfg = _WaveConfig(
amplification=self.cfg.amplification
)
wave = Wave(wcfg, str(path))
self.waves.append(wave)
trigger = self.cfg.trigger(
wave=wave,
scan_nsamp=round(
self.cfg.time_visible_s * self.cfg.scan_ratio * wave.smp_s),
# I tried extracting variable, but got confused as a result
)
channel = Channel(None, wave, trigger)
self.channels.append(channel)
self.nchan = len(self.waves)
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.nchan = len(self.channels)
def _load_outputs(self):
self.outputs = []
for output_cfg in self.cfg.outputs:
output = output_cfg(self.cfg)
self.outputs.append(output)
self.outputs = [output_cfg(self.cfg) for output_cfg in self.cfg.outputs]
def _render(self):
def play(self):
# Calculate number of frames (TODO master file?)
time_visible_s = self.cfg.time_visible_s
render_width_s = self.cfg.render_width_s
fps = self.cfg.fps
create_window = self.cfg.create_window
@ -158,10 +122,11 @@ class Ovgen:
# Get data from each wave
for wave, channel in zip(self.waves, self.channels):
sample = round(wave.smp_s * time_seconds)
region_len = round(wave.smp_s * time_visible_s)
region_len = round(wave.smp_s * render_width_s)
trigger_sample = channel.trigger.get_trigger(sample)
datas.append(wave.get_around(trigger_sample, region_len))
# FIXME channel.render_subsampling
# Render frame
renderer.render_frame(datas)