src/grepros/plugins/__init__.py
Go to the documentation of this file.
1 # -*- coding: utf-8 -*-
2 """
3 Plugins interface.
4 
5 Allows specifying custom plugins for "source", "scan" or "sink".
6 Auto-inits any plugins in grepros.plugins.auto.
7 
8 Supported (but not required) plugin interface methods:
9 
10 - `init(args)`: invoked at startup with command-line arguments
11 - `load(category, args)`: invoked with category "scan" or "source" or "sink",
12  using returned value if not None
13 
14 Plugins are free to modify package internals, like adding command-line arguments
15 to `main.ARGUMENTS` or sink types to `outputs.MultiSink`.
16 
17 Convenience methods:
18 
19 - `plugins.add_write_format(name, cls, label=None, options=((name, help), ))`:
20  adds an output plugin to defaults
21 - `plugins.add_output_label(label, flags)`:
22  adds plugin label to outputs enumerated in given argument help texts
23 - `plugins.get_argument(name, group=None)`:
24  returns a command-line argument configuration dictionary, or None
25 
26 ------------------------------------------------------------------------------
27 This file is part of grepros - grep for ROS bag files and live topics.
28 Released under the BSD License.
29 
30 @author Erki Suurjaak
31 @created 18.12.2021
32 @modified 14.07.2023
33 ------------------------------------------------------------------------------
34 """
35 
36 import glob
37 import os
38 import re
39 
40 import six
41 
42 from .. common import ConsolePrinter, ensure_namespace, get_name, import_item
43 from .. outputs import MultiSink
44 from . import auto
45 
46 
47 
48 PLUGINS = {}
49 
50 
51 OUTPUT_LABELS = {}
52 
53 
54 WRITE_OPTIONS = {}
55 
56 
57 DEFAULT_ARGS = dict(PLUGIN=[], STOP_ON_ERROR=False)
58 
59 
60 def init(args=None, **kwargs):
61  """
62  Imports and initializes all plugins from auto and from given arguments.
63 
64  @param args arguments as namespace or dictionary, case-insensitive
65  @param args.plugin list of Python modules or classes to import,
66  as ["my.module", "other.module.SomeClass", ],
67  or module or class instances
68  @param args.stop_on_error stop execution on any error like failing to load plugin
69  @param kwargs any and all arguments as keyword overrides, case-insensitive
70  """
71  args = ensure_namespace(args, DEFAULT_ARGS, **kwargs)
72  for f in sorted(glob.glob(os.path.join(os.path.dirname(__file__), "auto", "*"))):
73  if not f.lower().endswith((".py", ".pyc")): continue # for f
74  name = os.path.splitext(os.path.split(f)[-1])[0]
75  if name.startswith("__") or name in PLUGINS: continue # for f
76 
77  modulename = "%s.auto.%s" % (__package__, name)
78  try:
79  plugin = import_item(modulename)
80  if callable(getattr(plugin, "init", None)): plugin.init(args)
81  PLUGINS[name] = plugin
82  except Exception:
83  ConsolePrinter.error("Error loading plugin %s.", modulename)
84  if args.STOP_ON_ERROR: raise
85  if args: configure(args)
89 
90 
91 def configure(args=None, **kwargs):
92  """
93  Imports plugin Python packages, invokes init(args) if any, raises on error.
94 
95  @param args arguments as namespace or dictionary, case-insensitive
96  @param args.plugin list of Python modules or classes to import,
97  as ["my.module", "other.module.SomeClass", ],
98  or module or class instances
99  @param kwargs any and all arguments as keyword overrides, case-insensitive
100  """
101  args = ensure_namespace(args, DEFAULT_ARGS, **kwargs)
102  for obj in args.PLUGIN:
103  name = obj if isinstance(obj, six.string_types) else get_name(obj)
104  if name in PLUGINS: continue # for obj
105  try:
106  plugin = import_item(name) if isinstance(obj, six.string_types) else obj
107  if callable(getattr(plugin, "init", None)): plugin.init(args)
108  PLUGINS[name] = plugin
109  except ImportWarning:
110  raise
111  except Exception:
112  ConsolePrinter.error("Error loading plugin %s.", name)
113  raise
114 
115 
116 def load(category, args, collect=False):
117  """
118  Returns a plugin category instance loaded from any configured plugin, or None.
119 
120  @param category item category like "source", "sink", or "scan"
121  @param args arguments as namespace or dictionary, case-insensitive
122  @param collect if true, returns a list of instances,
123  using all plugins that return something
124  """
125  result = []
126  args = ensure_namespace(args)
127  for name, plugin in PLUGINS.items():
128  if callable(getattr(plugin, "load", None)):
129  try:
130  instance = plugin.load(category, args)
131  if instance is not None:
132  result.append(instance)
133  if not collect:
134  break # for name, plugin
135  except Exception:
136  ConsolePrinter.error("Error invoking %s.load(%r, args).", name, category)
137  raise
138  return result if collect else result[0] if result else None
139 
140 
141 def add_output_label(label, flags):
142  """
143  Adds plugin label to outputs enumerated in given argument help texts.
144 
145  @param label output label to add, like "Parquet"
146  @param flags list of argument flags like "--emit-field" to add the output label to
147  """
148  OUTPUT_LABELS.setdefault(label, []).extend(flags)
149 
150 
151 def add_write_format(name, cls, label=None, options=()):
152  """
153  Adds plugin to `--write` in main.ARGUMENTS and MultiSink formats.
154 
155  @param name format name like "csv", added to `--write .. format=FORMAT`
156  @param cls class providing Sink interface
157  @param label plugin label; if multiple plugins add the same option,
158  "label output" in help text is replaced with "label1/label2/.. output"
159  @param options a sequence of (name, help) to add to --write help, like
160  [("template=/my/path.tpl", "custom template to use for HTML output")]
161  """
162  MultiSink.FORMAT_CLASSES[name] = cls
163  if options: WRITE_OPTIONS.setdefault(label, []).extend(options)
164 
165 
166 def get_argument(name, group=None):
167  """
168  Returns a command-line argument dictionary, or None if not found.
169 
170  @param name argument name like "--write"
171  @param group argument group like "Output control", if any
172  """
173  from .. import main # Late import to avoid circular
174  if group:
175  return next((d for d in main.ARGUMENTS.get("groups", {}).get(group, [])
176  if name in d.get("args")), None)
177  return next((d for d in main.ARGUMENTS.get("arguments", [])
178  if name in d.get("args")), None)
179 
180 
182  """Populates argument texts with added output labels."""
183  if not OUTPUT_LABELS: return
184  from .. import main # Late import to avoid circular
185 
186  argslist = sum(main.ARGUMENTS.get("groups", {}).values(), main.ARGUMENTS["arguments"][:])
187  args = {f: x for x in argslist for f in x["args"]} # {flag or id(argdict): argdict}
188  args.update((id(x), x) for x in argslist)
189  arglabels = {} # {id(argdict): [label, ]}
190 
191  # First pass: collect arguments where to update output labels
192  for label, flag in ((l, f) for l, ff in OUTPUT_LABELS.items() for f in ff):
193  if flag in args: arglabels.setdefault(id(args[flag]), []).append(label)
194  else: ConsolePrinter.warn("Unknown command-line flag %r from output %r.", flag, label)
195 
196  # Second pass: replace argument help with full set of output labels
197  for arg, labels in ((args[x], ll) for x, ll in arglabels.items()):
198  match = re.search(r"(\A.*?\s*in\s)(\S+)(\s+output.*\Z)", arg["help"], re.DOTALL)
199  if not match:
200  ConsolePrinter.warn("Command-line flag %s has no text on output for labels %s.",
201  arg["args"], ", ".join(map(repr, sorted(set(labels)))))
202  continue # for arg, labels
203  labels2 = sorted(set(labels + match.group(2).split("/")), key=lambda x: x.lower())
204  arg["help"] = match.expand(r"\1%s\3" % "/".join(labels2))
205 
206  OUTPUT_LABELS.clear()
207 
208 
210  """Adds known non-auto plugins to `--plugin` argument help."""
211  plugins = []
212  for f in sorted(glob.glob(os.path.join(os.path.dirname(__file__), "*"))):
213  if not f.lower().endswith((".py", ".pyc")): continue # for f
214  name = os.path.splitext(os.path.split(f)[-1])[0]
215  if not name.startswith("__"):
216  plugins.append("%s.%s" % (__package__, name))
217 
218  pluginarg = get_argument("--plugin")
219  if pluginarg and plugins:
220  MAXLINELEN = 60
221  lines = ["load a Python module or class as plugin", "(built-in plugins: "]
222  for i, name in enumerate(plugins):
223  if not i: lines[-1] += name
224  else:
225  if len(lines[-1] + ", " + name) > MAXLINELEN:
226  lines[-1] += ", "
227  lines.append(" " + name)
228  else: lines[-1] += ", " + name
229  lines[-1] += ")"
230  pluginarg["help"] = "\n".join(lines)
231 
232 
234  """Populates main.ARGUMENTS with added write formats and options."""
235  writearg = get_argument("--write")
236  if not writearg: return
237 
238  formats = sorted(set(MultiSink.FORMAT_CLASSES))
239  writearg["metavar"] = "TARGET [format=%s] [KEY=VALUE ...]" % "|".join(formats)
240  if not WRITE_OPTIONS: return
241 
242  MAXNAME = 24 # Maximum space for name on same line as help
243  LEADING = " " # Leading indent on all option lines
244 
245  texts = {} # {name: help}
246  inters = {} # {name: indent between name and first line of help}
247  namelabels = {} # {name: [label,]}
248  namelens = {} # {name: len}
249 
250  # First pass: collect names
251  for label, opts in WRITE_OPTIONS.items():
252  for name, help in opts:
253  texts.setdefault(name, help)
254  namelabels.setdefault(name, []).append(label)
255  namelens[name] = len(name)
256 
257  # Second pass: calculate indent and inters
258  maxname = max(x if x <= MAXNAME else 0 for x in namelens.values())
259  for label, opts in WRITE_OPTIONS.items():
260  for name, help in opts:
261  inters[name] = "\n" if len(name) > MAXNAME else " " * (maxname - len(name) + 2)
262  indent = LEADING + " " + " " * (maxname or MAXNAME)
263 
264  # Third pass: replace labels for duplicate options
265  PLACEHOLDER = "<plugin label replacement>"
266  for name in list(texts):
267  if len(namelabels[name]) > 1:
268  for label in namelabels[name]:
269  texts[name] = texts[name].replace("%s output" % label, PLACEHOLDER)
270  labels = "/".join(sorted(filter(bool, namelabels[name]), key=lambda x: x.lower()))
271  texts[name] = texts[name].replace(PLACEHOLDER, labels + " output")
272 
273  fmt = lambda n, h: "\n".join((indent if i or "\n" == inters[n] else "") + l
274  for i, l in enumerate(h.splitlines()))
275  text = "\n".join(sorted("".join((LEADING, n, inters[n], fmt(n, h)))
276  for n, h in texts.items()))
277  writearg["help"] += "\n" + text
278 
279  WRITE_OPTIONS.clear()
280 
281 
282 
283 
284 __all__ = [
285  "PLUGINS", "init", "configure", "load", "add_write_format", "get_argument",
286  "populate_known_plugins", "populate_write_formats",
287 ]
grepros.common.get_name
def get_name(obj)
Definition: common.py:796
grepros.plugins.get_argument
def get_argument(name, group=None)
Definition: src/grepros/plugins/__init__.py:166
grepros.plugins.init
def init(args=None, **kwargs)
Definition: src/grepros/plugins/__init__.py:60
grepros.plugins.load
def load(category, args, collect=False)
Definition: src/grepros/plugins/__init__.py:116
grepros.plugins.populate_known_plugins
def populate_known_plugins()
Definition: src/grepros/plugins/__init__.py:209
grepros.plugins.populate_write_formats
def populate_write_formats()
Definition: src/grepros/plugins/__init__.py:233
grepros.plugins.add_write_format
def add_write_format(name, cls, label=None, options=())
Definition: src/grepros/plugins/__init__.py:151
grepros.plugins.populate_output_arguments
def populate_output_arguments()
Definition: src/grepros/plugins/__init__.py:181
grepros.common.ensure_namespace
def ensure_namespace(val, defaults=None, dashify=("WRITE_OPTIONS",), **kwargs)
Definition: common.py:658
grepros.plugins.add_output_label
def add_output_label(label, flags)
Definition: src/grepros/plugins/__init__.py:141
grepros.plugins.configure
def configure(args=None, **kwargs)
Definition: src/grepros/plugins/__init__.py:91
grepros.common.import_item
def import_item(name)
Definition: common.py:823


grepros
Author(s): Erki Suurjaak
autogenerated on Sat Jan 6 2024 03:11:29