16461262250

Committed 23 Jul 2025 04:05AM UTC coverage: 80.11% (+2.4%) from 77.726%

Build # 16461262250

Build Type

Pull #121

github

Committed by

web-flow

Commit Message

Merge 2976cead5 into 78870beec

Pull Request Pull Request #121: Online frequency filter (stripe removal)

Coverage Stats

14 of 17 new or added lines in 5 files covered. (82.35%)

1889 of 2358 relevant lines covered (80.11%)

11.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.95

/mio/cli/stream.py

"""
CLI commands for running streamDaq
"""

import os
from pathlib import Path
from typing import Callable, Optional

import click

from mio.cli.common import ConfigIDOrPath
from mio.models.process import FreqencyMaskingConfig
from mio.stream_daq import StreamDaq


@click.group()
def stream() -> None:
    """
    Command group for StreamDaq
    """
    pass


def _common_options(fn: Callable) -> Callable:
    fn = click.option(
        "-c",
        "--device_config",
        required=True,
        help=(
            "Either a config `id` or a path to device config YAML file for streamDaq. "
            "If you aren't using `id` you can ignore them."
            "(see models.stream.StreamDevConfig). If path is relative, treated as "
            "relative to the current directory, and then if no matching file is found, "
            "relative to the user `config_dir` (see `mio config --help`)."
        ),
        type=ConfigIDOrPath(),
    )(fn)
    return fn


def _capture_options(fn: Callable) -> Callable:
    fn = click.option(
        "-o",
        "--output",
        help="Output file basename for video, metadata, and binary exports",
        type=click.Path(),
    )(fn)
    fn = click.option(
        "-ok",
        "--output-kwarg",
        "okwarg",
        help="Output kwargs (passed to StreamDaq.init_video). \n"
        "passed as (potentially multiple) calls like\n\n"
        "mio stream capture -ok key1 val1 -ok key2 val2",
        multiple=True,
    )(fn)
    fn = click.option("--no-display", is_flag=True, help="Don't show video in real time")(fn)
    fn = click.option("-b", "--binary_export", is_flag=True, help="Save binary to a .bin file")(fn)
    fn = click.option(
        "-m",
        "--metadata_display",
        is_flag=True,
        help="Display metadata in real time. \n"
        "**WARNING:** This is still an **EXPERIMENTAL** feature and is **UNSTABLE**.",
    )(fn)
    fn = click.option(
        "-f",
        "--freq_mask_config",
        help="Path to, or ID of frequency masking config YAML file - "
        "applies frequency masking to the displayed video, "
        "but preserves raw video and does not modify the video output written to disk "
        "(apply postprocessing separately).",
        type=ConfigIDOrPath(),
    )(fn)
    return fn


@stream.command()
@_common_options
@_capture_options
def capture(
    device_config: Path,
    freq_mask_config: Optional[Path],
    output: Optional[Path],
    okwarg: Optional[dict],
    no_display: Optional[bool],
    binary_export: Optional[bool],
    metadata_display: Optional[bool],
    **kwargs: dict,
) -> None:
    """
    Capture video from a StreamDaq device, optionally saving as an encoded video or as raw binary
    """
    daq_inst = StreamDaq(device_config=device_config)
    okwargs = dict(okwarg)

    if output:
        unique_stem_path = get_unique_stempath(Path(output))
        video_output = unique_stem_path.with_suffix(".avi")
        metadata_output = unique_stem_path.with_suffix(".csv")

        binary_output = unique_stem_path.with_suffix(".bin") if binary_export else None
    else:
        video_output = None
        metadata_output = None
        binary_output = None

    if freq_mask_config:
        freq_mask_config = FreqencyMaskingConfig.from_any(freq_mask_config)
    else:
        freq_mask_config = None

    daq_inst.capture(
        source="fpga",
        video=video_output,
        video_kwargs=okwargs,
        metadata=metadata_output,
        binary=binary_output,
        show_video=not no_display,
        show_metadata=metadata_display,
        freq_mask_config=freq_mask_config,
    )


@stream.command()
@_common_options
@click.option(
    "-s",
    "--source",
    required=True,
    help="Path to RAW FPGA data to plug into okDevMock",
    type=click.Path(exists=True),
)
@click.option(
    "-p", "--profile", is_flag=True, default=False, help="Run with profiler (not implemented yet)"
)
@_capture_options
@click.pass_context
def test(ctx: click.Context, source: Path, profile: bool, **kwargs: dict) -> None:
    """
    Run StreamDaq in testing mode, using the okDevMock rather than the actual device
    """
    if profile:
        raise NotImplementedError("Profiling mode is not implemented")

    os.environ["STREAMDAQ_MOCKRUN"] = "just_placeholder"
    os.environ["PYTEST_OKDEV_DATA_FILE"] = str(source)

    ctx.forward(capture)


def get_unique_stempath(base_output: Path) -> Path:
    """
    Check the target directory if there are any files with the same basename (ignoring extensions)
    If so, append a number to the basename to make it unique.
    """
    directory = base_output.parent
    stem = base_output.stem

    # Ensure the directory exists
    directory.mkdir(parents=True, exist_ok=True)

    index = 1
    candidate_stem = stem

    def _any_stem_exists(candidate_stem_str: str) -> bool:
        # List all files and check if any have the same stem as the candidate
        return any(candidate_stem_str == p.stem for p in directory.iterdir() if p.is_file())

    # Iterate to find a unique stem
    while _any_stem_exists(candidate_stem):
        candidate_stem = f"{stem}-{index}"
        index += 1

    return directory / candidate_stem

1	"""
2	CLI commands for running streamDaq
3	"""
4
5	import os	15✔
6	from pathlib import Path	15✔
7	from typing import Callable, Optional	15✔
8
9	import click	15✔
10
11	from mio.cli.common import ConfigIDOrPath	15✔
12	from mio.models.process import FreqencyMaskingConfig	15✔
13	from mio.stream_daq import StreamDaq	15✔
14
15
16	@click.group()	15✔
17	def stream() -> None:	15✔
18	"""
19	Command group for StreamDaq
20	"""
21	pass	×
22
23
24	def _common_options(fn: Callable) -> Callable:	15✔
25	fn = click.option(	15✔
26	"-c",
27	"--device_config",
28	required=True,
29	help=(
30	"Either a config `id` or a path to device config YAML file for streamDaq. "
31	"If you aren't using `id` you can ignore them."
32	"(see models.stream.StreamDevConfig). If path is relative, treated as "
33	"relative to the current directory, and then if no matching file is found, "
34	"relative to the user `config_dir` (see `mio config --help`)."
35	),
36	type=ConfigIDOrPath(),
37	)(fn)
38	return fn	15✔
39
40
41	def _capture_options(fn: Callable) -> Callable:	15✔
42	fn = click.option(	15✔
43	"-o",
44	"--output",
45	help="Output file basename for video, metadata, and binary exports",
46	type=click.Path(),
47	)(fn)
48	fn = click.option(	15✔
49	"-ok",
50	"--output-kwarg",
51	"okwarg",
52	help="Output kwargs (passed to StreamDaq.init_video). \n"
53	"passed as (potentially multiple) calls like\n\n"
54	"mio stream capture -ok key1 val1 -ok key2 val2",
55	multiple=True,
56	)(fn)
57	fn = click.option("--no-display", is_flag=True, help="Don't show video in real time")(fn)	15✔
58	fn = click.option("-b", "--binary_export", is_flag=True, help="Save binary to a .bin file")(fn)	15✔
59	fn = click.option(	15✔
60	"-m",
61	"--metadata_display",
62	is_flag=True,
63	help="Display metadata in real time. \n"
64	"WARNING: This is still an EXPERIMENTAL feature and is UNSTABLE.",
65	)(fn)
66	fn = click.option(	15✔
67	"-f",
68	"--freq_mask_config",
69	help="Path to, or ID of frequency masking config YAML file - "
70	"applies frequency masking to the displayed video, "
71	"but preserves raw video and does not modify the video output written to disk "
72	"(apply postprocessing separately).",
73	type=ConfigIDOrPath(),
74	)(fn)
75	return fn	15✔
76
77
78	@stream.command()	15✔
79	@_common_options	15✔
80	@_capture_options	15✔
81	def capture(	15✔
82	device_config: Path,
83	freq_mask_config: Optional[Path],
84	output: Optional[Path],
85	okwarg: Optional[dict],
86	no_display: Optional[bool],
87	binary_export: Optional[bool],
88	metadata_display: Optional[bool],
89	**kwargs: dict,
90	) -> None:
91	"""
92	Capture video from a StreamDaq device, optionally saving as an encoded video or as raw binary
93	"""
94	daq_inst = StreamDaq(device_config=device_config)	15✔
95	okwargs = dict(okwarg)	15✔
96
97	if output:	15✔
98	unique_stem_path = get_unique_stempath(Path(output))	15✔
99	video_output = unique_stem_path.with_suffix(".avi")	15✔
100	metadata_output = unique_stem_path.with_suffix(".csv")	15✔
101
102	binary_output = unique_stem_path.with_suffix(".bin") if binary_export else None	15✔
103	else:
104	video_output = None	×
105	metadata_output = None	×
106	binary_output = None	×
107
108	if freq_mask_config:	15✔
NEW 109	freq_mask_config = FreqencyMaskingConfig.from_any(freq_mask_config)	×
110	else:
111	freq_mask_config = None	15✔
112
113	daq_inst.capture(	15✔
114	source="fpga",
115	video=video_output,
116	video_kwargs=okwargs,
117	metadata=metadata_output,
118	binary=binary_output,
119	show_video=not no_display,
120	show_metadata=metadata_display,
121	freq_mask_config=freq_mask_config,
122	)
123
124
125	@stream.command()	15✔
126	@_common_options	15✔
127	@click.option(	15✔
128	"-s",
129	"--source",
130	required=True,
131	help="Path to RAW FPGA data to plug into okDevMock",
132	type=click.Path(exists=True),
133	)
134	@click.option(	15✔
135	"-p", "--profile", is_flag=True, default=False, help="Run with profiler (not implemented yet)"
136	)
137	@_capture_options	15✔
138	@click.pass_context	15✔
139	def test(ctx: click.Context, source: Path, profile: bool, **kwargs: dict) -> None:	15✔
140	"""
141	Run StreamDaq in testing mode, using the okDevMock rather than the actual device
142	"""
143	if profile:	×
144	raise NotImplementedError("Profiling mode is not implemented")	×
145
146	os.environ["STREAMDAQ_MOCKRUN"] = "just_placeholder"	×
147	os.environ["PYTEST_OKDEV_DATA_FILE"] = str(source)	×
148
149	ctx.forward(capture)	×
150
151
152	def get_unique_stempath(base_output: Path) -> Path:	15✔
153	"""
154	Check the target directory if there are any files with the same basename (ignoring extensions)
155	If so, append a number to the basename to make it unique.
156	"""
157	directory = base_output.parent	15✔
158	stem = base_output.stem	15✔
159
160	# Ensure the directory exists
161	directory.mkdir(parents=True, exist_ok=True)	15✔
162
163	index = 1	15✔
164	candidate_stem = stem	15✔
165
166	def _any_stem_exists(candidate_stem_str: str) -> bool:	15✔
167	# List all files and check if any have the same stem as the candidate
168	return any(candidate_stem_str == p.stem for p in directory.iterdir() if p.is_file())	15✔
169
170	# Iterate to find a unique stem
171	while _any_stem_exists(candidate_stem):	15✔
172	candidate_stem = f"{stem}-{index}"	×
173	index += 1	×
174
175	return directory / candidate_stem	15✔

Aharoni-Lab / mio / 16461262250

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous