dqm/cli/dependency.py source
1 from contextlib import contextmanager
2 import logging
3 from typing import Any
4
5 from dqm_ml_core import PluginLoadedRegistry
6
7 logger = logging.getLogger(__name__)
8
9
10 @contextmanager
11 def optional_dependencies(error: str = "ignore"):
12 assert error in {"raise", "warn", "ignore"}
13 try:
14 yield None
15 except ImportError as e:
16 if error == "raise":
17 raise e
18 if error == "warn":
19 msg = f'Missing optional dependency "{e.name}". Use pip to install.'
20 print(f"Warning: {msg}")
21
22
23 def display_list_of(arg_list: list[str] | None = None):
24
25 print("Available data metrics")
26 for key, value in PluginLoadedRegistry.get_metrics_registry().items():
27 print(f"- {key} - {value}")
28
29 print("Available data loaders")
30 for key, value in PluginLoadedRegistry.get_dataloaders_registry().items():
31 print(f"- {key} - {value}")
32
33 print("Available outputs writter")
34 for key, value in PluginLoadedRegistry.get_outputwiter_registry().items():
35 print(f"- {key} - {value}")
36
- E302 Expected 2 blank lines, found 1
38 command_list = {"list": display_list_of}
39 optional_dep_mode = "warn"
40
41 # We import available command for dqml cli
42 with optional_dependencies(optional_dep_mode):
43 from dqm_ml_pipeline._version_ import version as pipeline_version
44 from dqm_ml_pipeline.cli import execute
45
46 logger.debug(f"Different dqm-ml-pipeline version {pipeline_version}")
47 command_list["process"] = execute
48
49 return command_list