Coverage for src/pythia/cli/app.py: 61%
167 statements
« prev ^ index » next coverage.py v6.4.4, created at 2022-10-07 19:27 +0000
1"""Command-line interfact typer application for pythia.
3Its a typer application mimmicking `gst-launch`.
5"""
6from __future__ import annotations
8import enum
9import re
10import sys
11import traceback
12from collections import defaultdict
13from pathlib import Path
14from typing import Any
15from typing import Callable
16from typing import cast
17from typing import Collection
18from typing import Dict
19from typing import List
20from typing import Optional
21from typing import Tuple
23import typer
24from fire import decorators
25from fire.core import _MakeParseFn
27from pythia import __version__
28from pythia.applications.command_line import CliApplication
29from pythia.exceptions import InvalidPipelineError
30from pythia.pipelines.base import UNABLE_TO_PLAY_PIPELINE
31from pythia.types import PadDirection
32from pythia.types import Probes
33from pythia.utils.ext import import_from_str
34from pythia.utils.gst import GLib
35from pythia.utils.gst import Gst
36from pythia.utils.gst import gst_init
38LOOKS_LIKE_JINJA_WARN = (
39 "It looks like you're attemplting to use a pipeline using jinja syntax,"
40 " but jinja is not installed."
41 " Falling back to native python renderer."
42 " Please reinstall pythia with jinja extra"
43 " (eg 'pip install pythia[jinja]')"
44)
47class Exit(enum.Enum):
48 """Pythia cli exit wrapper."""
50 INVALID_PIPELINE = 2
51 INVALID_EXTRACTION = 3
52 EXTRACTION_NOT_BOUND = 4
53 UNPLAYABLE_PIPELINE = 5
55 def __call__(self, exc) -> typer.Exit:
57 print(traceback.format_exc())
58 typer.secho(
59 str(exc),
60 fg="red",
61 )
62 return typer.Exit(self.value)
65app = typer.Typer(
66 add_completion=False,
67)
70def version_callback(value: bool) -> None: # noqa: FBT001
71 """Echo pythia version and exit.
73 Args:
74 value: If set, echos pythia version and exit.
76 Raises:
77 Exit: when value is passed, to exit the program.
79 """
80 if value: 80 ↛ 81line 80 didn't jump to line 81, because the condition on line 80 was never true
81 typer.echo(__version__)
82 raise typer.Exit()
85def pipe_from_file(path: Path) -> str:
86 """Read a gst-launch-like pipeline from a file.
88 Args:
89 path: Location of the pipeline. If set to `Path("-")`, reads
90 from stdin instead.
92 Returns:
93 The pipeline string.
95 """
96 if path == Path("-"): 96 ↛ 97line 96 didn't jump to line 97, because the condition on line 96 was never true
97 return sys.stdin.read()
98 return path.read_text()
101def pipe_from_parts(parts: Collection[str]) -> str:
102 """Join `sys.argv`-split strings wuith spaces.
104 Args:
105 parts: Strings to be joined.
107 Returns:
108 The strings, joined with simple spaces.
110 """
111 raw = " ".join(parts)
112 return raw
115class Mode(str, enum.Enum):
116 """Pipeline template loading from cli."""
118 FILE = "FILE"
119 """Pipeline template from file."""
121 ARGV = "ARGV"
122 """Pipeline template from argv."""
125def define_template(
126 ctx,
127 pipeline_file: Optional[Path],
128 pipeline_parts: Optional[List[str]],
129) -> Tuple[Mode, str]:
130 """Construct a pipeline template wither from a file or its parts.
132 Args:
133 ctx: click application context - used to print help on incorrect
134 usage.
135 pipeline_file: path to a nonempty file with a `gst-launch`-like
136 pipeline as its contents.
137 pipeline_parts: console-split sections of a pipeline, eg as
138 received from `sys.argv`.
140 Returns:
141 The pipeline template as a string.
143 Raises:
144 Abort: Either both or none of the exclusive (pipeline_file,
145 pipeline_parts) was supplied.
147 Either `pipeline_file` or `pipeline_parts` must be passed.
149 """
151 if pipeline_file and not pipeline_parts: 151 ↛ 161line 151 didn't jump to line 161, because the condition on line 151 was never false
152 try:
153 return Mode.FILE, pipe_from_file(pipeline_file)
154 except FileNotFoundError as exc:
155 typer.secho(
156 f"Pipeline file '{str(pipeline_file)}' not found.",
157 fg="red",
158 )
159 raise typer.Abort() from exc
161 if pipeline_parts and not pipeline_file:
162 return Mode.ARGV, pipe_from_parts(pipeline_parts)
164 typer.secho(
165 "Pass either a pipeline file or use gst-launch syntax, exclusively.",
166 fg="red",
167 )
168 typer.echo(ctx.get_help())
169 raise typer.Abort()
172def _receive_any_return_kw(*_, **kw):
173 return kw
176CtxRetType = Tuple[List[str], Dict[str, Any]]
179def parse_arbitrary_argv(
180 component: Callable,
181 args: List[str],
182) -> CtxRetType:
183 """Parse arguments into positional and named according to a function.
185 Args:
186 component: function to supply param spec and metadata.
187 args: list of positional arguments to parse.
189 Returns:
190 Positional arguments.
191 Named value pairs dictionary extracted from the input arg list.
193 Examples:
194 >>> def asdf()
196 See Also:
197 :func:`fire.core._Fire` and its call to
198 :func:`fire.core._CallAndUpdateTrace`
200 """
201 parse = _MakeParseFn(component, decorators.GetMetadata(component))
202 (parts, kwargs), *_ = parse(args)
203 return parts, kwargs
206def _ctx_cb() -> CtxRetType:
207 return parse_arbitrary_argv(_receive_any_return_kw, sys.argv[1:])
210Renderer = Callable[[str, Dict[str, Any]], str]
213def _native_renderer(pipeline_template: str, context: dict) -> str:
214 ret = pipeline_template
215 found = {}
216 for required in re.findall(r"(?<=\{).*?(?=\})", pipeline_template): 216 ↛ 218line 216 didn't jump to line 218, because the loop on line 216 never started
218 found[required] = context.pop(
219 required, context.pop(required.replace("-", "_"))
220 )
222 ret = ret.format_map(found)
223 return ret
226def _jinja_renderer(pipeline_template: str, context: dict) -> str:
227 from jinja2 import meta # noqa: C0415
228 from jinja2 import Template # noqa: C0415
230 jinja_template = Template(pipeline_template)
231 all_required = meta.find_undeclared_variables(
232 jinja_template.environment.parse(pipeline_template)
233 )
235 found = {}
236 for required in all_required:
237 found[required] = context.pop(required)
239 return jinja_template.render(found)
242def choose_renderer(pipeline_template: str) -> Renderer:
243 """Decide wether to use jinja or vanilla python template.
245 If the syntax is detected as jinja but its not installed, issue a
246 warning and fall back to native python.
248 Args:
249 pipeline_template: Inspected to decide its underlying syntax.
251 Returns:
252 One of the available template renderers - :func:`_native_renderer`
253 or :func:`_jinja_renderer`.
255 """
256 looks_like_jinja = "{{" in pipeline_template
257 try:
258 import jinja2 # noqa: F401
259 except ImportError:
260 if looks_like_jinja:
261 typer.secho(LOOKS_LIKE_JINJA_WARN, fg="yellow")
262 return _native_renderer
263 else:
264 return _jinja_renderer if looks_like_jinja else _native_renderer
267def build_pipeline(
268 pipeline_parts: Optional[List[str]],
269 pipeline_file: Optional[Path],
270 ctx: typer.Context,
271 *,
272 extra_ctx: Optional[CtxRetType],
273) -> str:
274 """Construct a pipeline either from args or file.
276 Args:
277 pipeline_parts: console-split sections of a pipeline, eg as
278 received from `sys.argv`.
279 pipeline_file: Location to read the template pipieline from.
280 ctx: typer context - used to print app help in case of problems.
281 extra_ctx: Additional context to render the pipeline template.
283 Returns:
284 The pipeline (either args or from file), parsed as a template,
285 then injected with the additional context (if any).
287 """
288 if extra_ctx is None: 288 ↛ 289line 288 didn't jump to line 289, because the condition on line 288 was never true
289 template_context: Dict[str, Any] = {}
290 else:
291 pipeline_parts, template_context = extra_ctx
292 _, pipeline_template = define_template(ctx, pipeline_file, pipeline_parts)
293 render_backend = choose_renderer(pipeline_template)
295 return render_backend(pipeline_template, template_context)
298def _validate_pipeline(pipeline_string: str, *, check: bool = True) -> int:
299 typer.echo(f"Pipeline:\n```gst\n{pipeline_string}\n```")
301 if not check:
302 return 0
304 gst_init()
306 try:
307 Gst.parse_launch(pipeline_string)
308 except GLib.Error as exc:
309 typer.secho(
310 f"Invalid pipeline - reason: {exc}",
311 fg="red",
312 )
313 return 1
314 else:
315 typer.secho(
316 "Pipeline syntax looks good",
317 fg="green",
318 )
319 return 0
322EXTRACTOR_PARSER = (
323 r"^(?P<module>\/?\w[\w_-]*(?:[\.\/]\w[\w_-]*)*?(?P<suffix>\.py)?)"
324 r":"
325 r"(?P<probe>\w[\w_]*)"
326 r"@"
327 r"(?P<element>\w[\w_-]*)"
328 r"\."
329 r"(?P<direction>src|sink)"
330 r"$"
331)
334def _validate_extractors(extractor: Optional[List[str]]) -> Probes:
335 if not extractor:
336 return {}
337 parser_re = re.compile(EXTRACTOR_PARSER, flags=re.MULTILINE)
338 probes: Probes = defaultdict(lambda: defaultdict(list))
339 for extractor_string in extractor:
340 match = parser_re.match(extractor_string)
341 if not match: 341 ↛ 342line 341 didn't jump to line 342, because the condition on line 341 was never true
342 raise ValueError(
343 f"Unable to parse extractor '{extractor_string}'."
344 " Make sure it has the form "
345 "'my_module:my_function@element-name.pad-direction'."
346 )
347 data = match.groupdict()
348 try:
349 module = import_from_str(data["module"], suffix=data["suffix"])
350 except ImportError as exc:
351 raise ValueError(
352 f"Unable to import module from extractor '{extractor_string}'."
353 " Make sure it exists, and is importable."
354 ) from exc
355 try:
356 raw_probe = getattr(module, data["probe"])
357 except (KeyError, NameError) as exc:
358 raise ValueError(
359 "Unable to get probe from imported module: "
360 f"'{extractor_string}'."
361 " Make sure the function is available in the it's namespace."
362 ) from exc
363 direction = cast(PadDirection, data["direction"])
364 probes[data["element"]][direction].append(raw_probe)
365 return probes
368@app.command(
369 context_settings={
370 "ignore_unknown_options": True,
371 },
372 help="gst-launch on steroids - use without caution.",
373)
374def main( # noqa: C0116, R0913
375 ctx: typer.Context,
376 extra_ctx: Optional[str] = typer.Option( # noqa: B008
377 None,
378 callback=_ctx_cb,
379 is_eager=True,
380 help="extra kw.",
381 ),
382 pipeline_parts: Optional[List[str]] = typer.Argument( # noqa: B008
383 None,
384 help=(
385 "A gstreamer pipeline, as used with vanilla gst-launch. For "
386 "example: `videotestsrc ! identity eos-after=5 ! fakesink`."
387 " Can be empty, in which case the `-p` flag *must* be passed"
388 " to load a pipeline form file."
389 ),
390 ),
391 pipeline_file: Optional[Path] = typer.Option( # noqa: B008
392 None,
393 "--pipeline-file",
394 "-p",
395 help=(
396 "Load pipeline from a path instead."
397 " You can use '-' to read from stdin."
398 ),
399 envvar="PYTHIA_PIPELINE_FILE",
400 ),
401 extractor: Optional[List[str]] = typer.Option( # noqa: B008
402 None,
403 "--probe",
404 "-x",
405 help=(
406 "Install metadata extraction buffer probe in a gst pad."
407 " The syntax must be: 'my_module:my_function@pgie.src'."
408 ),
409 envvar="PYTHIA_EXTRACTOR",
410 ),
411 _: Optional[bool] = typer.Option( # noqa: B008
412 None,
413 "--version",
414 callback=version_callback,
415 is_eager=True,
416 help="Show the version and exit.",
417 ),
418 dry_run: bool = typer.Option( # noqa: B008,FBT001
419 False,
420 "--dry-run",
421 help="Show the resulting pipeline and exit.",
422 ),
423 check: bool = typer.Option( # noqa: B008,FBT001
424 False,
425 "--check",
426 help=(
427 "If set, validates the pipeline with Gst."
428 " Only used with '--dry-run', otherwise ignored."
429 ),
430 ),
431) -> None:
433 pipeline_string = build_pipeline(
434 pipeline_parts,
435 pipeline_file,
436 ctx,
437 extra_ctx=cast(CtxRetType, extra_ctx),
438 )
440 if dry_run: 440 ↛ 441line 440 didn't jump to line 441, because the condition on line 440 was never true
441 retcode = _validate_pipeline(pipeline_string, check=check)
442 raise typer.Exit(retcode)
444 try:
445 extractors = _validate_extractors(extractor)
446 except (ImportError, ValueError, AttributeError) as exc:
447 raise Exit.INVALID_EXTRACTION(exc) from exc
449 try:
450 gst_init()
451 run = CliApplication.from_pipeline_string(pipeline_string, extractors)
452 except NameError as exc:
453 raise Exit.EXTRACTION_NOT_BOUND(exc) from exc
454 except InvalidPipelineError as exc:
455 raise Exit.INVALID_PIPELINE(exc) from exc
456 try:
457 run()
458 except RuntimeError as exc:
459 if UNABLE_TO_PLAY_PIPELINE in str(exc):
460 raise Exit.UNPLAYABLE_PIPELINE(exc) from exc
461 raise
464if __name__ == "__main__": 464 ↛ 465line 464 didn't jump to line 465, because the condition on line 464 was never true
465 app()