mirror/weewx
1
0
Fork 0
mirror of https://github.com/weewx/weewx.git synced 2026-04-17 08:06:58 -04:00
Code Issues Packages Projects Releases Wiki Activity

Allow extra command line options to be passed to extension installer.

Browse Source 
Addresses issue #[1041](https://github.com/weewx/weewx/issues/1041).
This commit is contained in:
Tom Keffer
2025-11-29 10:42:15 -08:00
parent 13b714736a
commit a9254fa7fb
5 changed files with 62 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
docs_src/changes.md
View File

@@ -1,6 +1,12 @@
WeeWX change history
--------------------
### 5.3.0 MM/DD/YYYY
Allow extra command line options to be passed to extension installer.
Addresses issue #[1041](https://github.com/weewx/weewx/issues/1041).
### 5.2.1 MM/DD/YYYY
Remove unnecessary `UNIQUE` index on `PRIMARY KEY` columns in SQLite, achieving

25
src/weecfg/extension.py
View File

@@ -33,6 +33,10 @@ class InstallError(Exception):
class ExtensionInstaller(dict):
"""Base class for extension installers."""
def process_args(self, args):
"""Can be overridden by installers. It should parse any extra command line arguments."""
pass
def configure(self, engine):
"""Can be overridden by installers. It should return True if the installer modifies
the configuration dictionary."""
@@ -95,13 +99,15 @@ class ExtensionEngine:
_, installer = weecfg.get_extension_installer(ext_cache_dir)
return installer
def install_extension(self, extension_path, no_confirm=False):
def install_extension(self, extension_path, no_confirm=False, extra_args=None):
"""Install an extension.
Args:
extension_path(str): Either a file path, a directory path, or an URL.
no_confirm(bool): If False, ask for a confirmation before installing. Otherwise,
just do it.
extra_args(list[str]|None): Extra arguments from the command line to pass to the
extension installer.
"""
ans = weeutil.weeutil.y_or_n(f"Install extension '{extension_path}' (y/n)? ",
noprompt=no_confirm)
@@ -127,7 +133,7 @@ class ExtensionEngine:
filetype = 'zip'
else:
filetype = 'tar'
extension_name = self._install_from_file(test_fd.name, filetype)
extension_name = self._install_from_file(test_fd.name, filetype, extra_args)
elif not os.path.exists(extension_path):
raise InstallError(f"Path {extension_path} does not exist.")
elif os.path.isfile(extension_path):
@@ -137,22 +143,24 @@ class ExtensionEngine:
filetype = 'zip'
else:
filetype = 'tar'
extension_name = self._install_from_file(extension_path, filetype)
extension_name = self._install_from_file(extension_path, filetype, extra_args)
elif os.path.isdir(extension_path):
# It's a directory. Install directly.
extension_name = self.install_from_dir(extension_path)
extension_name = self.install_from_dir(extension_path, extra_args)
else:
raise InstallError(f"Unrecognized type for {extension_path}")
self.printer.out(f"Finished installing extension {extension_name} from {extension_path}")
def _install_from_file(self, filepath, filetype):
def _install_from_file(self, filepath, filetype, extra_args):
"""Install an extension from a file.
Args:
filepath(str): A path to the file holding the extension.
filetype(str): The type of file. If 'zip', it's assumed to be a zipfile. Anything else,
and it's assumed to be a tarfile.
extra_args(list[str]|None): Extra arguments from the command line to pass to the
extension installer.
"""
# Make a temporary directory into which to extract the file.
with tempfile.TemporaryDirectory() as dir_name:
@@ -167,11 +175,11 @@ class ExtensionEngine:
"(the extension archive contains more than a "
"single root directory)")
extension_dir = os.path.join(dir_name, extension_reldir)
extension_name = self.install_from_dir(extension_dir)
extension_name = self.install_from_dir(extension_dir, extra_args)
return extension_name
def install_from_dir(self, extension_dir):
def install_from_dir(self, extension_dir, extra_args):
"""Install the extension whose components are in extension_dir"""
self.printer.out(f"Request to install extension found in directory {extension_dir}",
level=2)
@@ -209,6 +217,9 @@ class ExtensionEngine:
self.config_dict['Engine']['Services'][service_group] = svc_list
save_config = True
self.printer.out(f"Added new service {svc} to {service_group}.", level=3)
# Pass any extra arguments on to the installer
if extra_args:
installer.process_args(extra_args)
# Give the installer a chance to do any customized configuration
save_config |= installer.configure(self)

21
src/weectl.py
View File

@@ -7,6 +7,7 @@
import argparse
import importlib
import inspect
import sys
import weewx
@@ -59,15 +60,21 @@ def main():
module = importlib.import_module(f'weectllib.{subcommand}_cmd')
module.add_subparser(subparsers)
# Time to parse the whole tree
namespace = parser.parse_args()
if hasattr(namespace, 'func'):
# Call the appropriate action function:
# Parse what we can. This gives us access to the namespace.
namespace, extra_args = parser.parse_known_args()
# Now take a look at the signature of the dispatch function and see how many arguments it has.
sig = inspect.signature(namespace.func)
if len(sig.parameters) == 1:
# No optional arguments. Reparse everything, this time using the more restrictive
# parse_args() method. This will raise an error if there are any unrecognized arguments.
namespace = parser.parse_args()
namespace.func(namespace)
elif len(sig.parameters) == 2:
# Optional arguments are possible. Pass them on to the dispatch function.
namespace.func(namespace, extra_args)
else:
# Shouldn't get here. Some sub-subparser failed to include a 'func' argument.
parser.print_help()
# Shouldn't be here. Some weird dispatch function.
raise TypeError(f"Unexpected dispatch function signature: {sig}")
if __name__ == "__main__":

21
src/weectllib/__init__.py
View File

@@ -92,3 +92,24 @@ def dispatch(namespace):
if hasattr(namespace, 'dry_run') and namespace.dry_run:
print("This was a dry run. Nothing was actually done.")
log.info("This was a dry run. Nothing was actually done.")
def dispatch_with_args(namespace, extra_args):
"""All weectl commands come here. This function reads the configuration file, sets up logging,
then dispatches to the actual action.
"""
config_path, config_dict, log = weeutil.startup.start_app('weectl',
__name__,
namespace.config,
None)
# Note a dry-run, if applicable:
if hasattr(namespace, 'dry_run') and namespace.dry_run:
print("This is a dry run. Nothing will actually be done.")
log.info("This is a dry run. Nothing will actually be done.")
# Call the specified action:
namespace.action_func(config_dict, namespace, extra_args)
if hasattr(namespace, 'dry_run') and namespace.dry_run:
print("This was a dry run. Nothing was actually done.")
log.info("This was a dry run. Nothing was actually done.")

6
src/weectllib/extension_cmd.py
View File

@@ -78,7 +78,7 @@ def add_subparser(subparsers):
help="Don't ask for confirmation. Just do it.")
install_extension_parser.add_argument('--verbosity', type=int, default=1, metavar='N',
help="How much information to display (0|1|2|3).")
install_extension_parser.set_defaults(func=weectllib.dispatch)
install_extension_parser.set_defaults(func=weectllib.dispatch_with_args)
install_extension_parser.set_defaults(action_func=install_extension)
# ---------- Action uninstall' ----------
@@ -111,9 +111,9 @@ def list_extensions(config_dict, _):
ext.enumerate_extensions()
def install_extension(config_dict, namespace):
def install_extension(config_dict, namespace, extra_args=None):
ext = _get_extension_engine(config_dict, namespace.dry_run, namespace.verbosity)
ext.install_extension(namespace.source, no_confirm=namespace.yes)
ext.install_extension(namespace.source, no_confirm=namespace.yes, extra_args=extra_args)
def uninstall_extension(config_dict, namespace):