New update_fsl_package script

Dedicated script for updating FSL conda packages. Does not support installation of new packages at this point in time.

Also added the fslipython wrapper for $FSLDIR/bin/ipython

share/fsl/sbin/update_fsl_package

 #!/usr/bin/env fslpython
-"""Install/update one or more FSL packages using conda.
+"""Update one or more FSL packages using conda.
 
 This script is only intended to be used within conda-based FSL installations
 that have been created with the fslinstaller.py script. It is not intended to
 be used within conda environments that have had FSL packages installed into
 them - in this scenario, conda should be used directly.
 
+Currently this script cannot be used to install new packages - it is limited to
+updating packages that are already installed. To install new packages, conda
+should be used directly.
+
 The script performs the following steps:
 
  1. Parses command line arguments (primarily the list of packages to
-    install/update).
+    update).
 
  2. Queries the FSL conda channel(s) to find the latest available versions of
     the requested packages.
 
- 4. Runs "conda install" to install/update the packages.
+ 4. Runs "conda install" to update the packages.
 """
 # Note: Conda does have a Python API:
 #
@@ -42,7 +46,7 @@ import urllib.parse    as     urlparse
 import urllib.request  as     urlrequest
 from   collections     import defaultdict
 
-from typing import Dict, List, Union, Tuple, Optional, Sequence
+from typing import Dict, List, Union, Tuple, Optional, Sequence, Any
 
 
 log = logging.getLogger(__name__)
@@ -52,17 +56,23 @@ PUBLIC_FSL_CHANNEL   = 'https://fsl.fmrib.ox.ac.uk/fsldownloads/fslconda/public/
 INTERNAL_FSL_CHANNEL = 'https://fsl.fmrib.ox.ac.uk/fsldownloads/fslconda/internal/'
 
 
-def conda(cmd : str) -> str:
+def conda(cmd : str, capture_output=True, **kwargs) -> str:
     """Runs the conda command with the given arguments, returning its standard
     output as a string.
+
+    Keyword arguments are passed through to subprocess.run.
     """
     fsldir   = os.environ['FSLDIR']
     condabin = op.join(fsldir, 'bin', 'conda')
 
     log.debug(f'Running {condabin} {cmd}')
 
+    kwargs['check']          = False
+    kwargs['text']           = True
+    kwargs['capture_output'] = capture_output
+
     cmd    = [condabin] + shlex.split(cmd)
-    result = sp.run(cmd, check=False, capture_output=True, text=True)
+    result = sp.run(cmd, **kwargs)
 
     log.debug(f'Exit code: {result.returncode}')
 
@@ -94,16 +104,41 @@ def identify_platform() -> str:
     key    = (system, cpu)
 
     if key not in platforms:
-        supported = ', '.join(['[{}, {}]' for s, c in platforms])
-        raise Exception('This platform [{}, {}] is unrecognised or '
-                        'unsupported! Supported platforms: {}'.format(
-                            system, cpu, supported))
+        supported = ', '.join([f'[{s}, {c}]' for s, c in platforms])
+        raise Exception(f'This platform [{system}, {cpu}] is unrecognised or '
+                        f'unsupported! Supported platforms: {supported}')
 
     log.debug(f'Detected platform: {platforms[key]}')
 
     return platforms[key]
 
 
+def http_request(url      : str,
+                 username : str = None,
+                 password : str = None) -> Any:
+    """Download JSON data from the given URL. """
+
+    if username is not None:
+        urlbase = urlparse.urlparse(url).netloc
+        pwdmgr = urlrequest.HTTPPasswordMgrWithDefaultRealm()
+        pwdmgr.add_password(None, urlbase, username, password)
+        handler = urlrequest.HTTPBasicAuthHandler(pwdmgr)
+        opener = urlrequest.build_opener(handler)
+        opener.open(url)
+        urlrequest.install_opener(opener)
+
+    log.debug(f'Downloading {url} ...')
+
+    request  = urlrequest.Request(url, method='GET')
+    response = urlrequest.urlopen(request)
+    payload  = response.read()
+
+    if len(payload) == 0: payload = {}
+    else:                 payload = json.loads(payload)
+
+    return payload
+
+
 @ft.total_ordering
 class Version:
     """Class for parsing/comparing version strings. """
@@ -140,56 +175,6 @@ class Version:
         return self.parsed_version == other.parsed_version
 
 
-def http_request(url      : str,
-                 username : str = None,
-                 password : str = None):
-    """Download JSON data from the given URL. """
-
-    if username is not None:
-        urlbase = urlparse.urlparse(url).netloc
-        pwdmgr = urlrequest.HTTPPasswordMgrWithDefaultRealm()
-        pwdmgr.add_password(None, urlbase, username, password)
-        handler = urlrequest.HTTPBasicAuthHandler(pwdmgr)
-        opener = urlrequest.build_opener(handler)
-        opener.open(url)
-        urlrequest.install_opener(opener)
-
-    log.debug(f'Downloading {url} ...')
-
-    request  = urlrequest.Request(url, method='GET')
-    response = urlrequest.urlopen(request)
-    payload  = response.read()
-
-    if len(payload) == 0: payload = {}
-    else:                 payload = json.loads(payload)
-
-    return payload
-
-
-@ft.lru_cache
-def query_installed_packages() -> Dict[str, str]:
-    """Uses conda to find out the versions of all packages installed in
-    $FSLDIR, which are sourced from the FSL conda channels.
-    """
-
-    channels = [PUBLIC_FSL_CHANNEL  .rstrip('/'),
-                INTERNAL_FSL_CHANNEL.rstrip('/')]
-
-    # conda info returns a list of dicts,
-    # one per package. We re-arrange this
-    # into a dict of {pkgname : version}
-    # mappings.
-    fsldir = os.environ['FSLDIR']
-    info   = json.loads(conda(f'list -p {fsldir} --json'))
-    pkgs   = {}
-
-    for pkg in info:
-        if pkg['base_url'].rstrip() in channels:
-            pkgs[pkg['name']] = pkg['version']
-
-    return pkgs
-
-
 @ft.total_ordering
 @dataclasses.dataclass
 class Package:
@@ -202,34 +187,27 @@ class Package:
     name : str
     """Package name."""
 
+    version : str
+    """Package version string."""
+
     channel : str
     """URL of the channel which hosts the package."""
 
     platform : str
     """Platform identifier."""
 
-    version : str
-    """Package version string."""
-
-    dependencies : List[str]
+    dependencies : List[str] = None
     """References to all packages which this package depends on. Stored as
     "package[ version-constraint]" strings.
     """
 
+
     @property
     def development(self) -> bool:
         """Return True if this is a development version of the package. """
         return 'dev' in self.version
 
 
-    @property
-    def installed_version(self) -> str:
-        """Return the version of this package which is currently installed, or '-'
-        if not installed.
-        """
-        return query_installed_packages().get(self.name, '-')
-
-
     def __lt__(self, pkg):
         """Only valid when comparing another Package with the same name and
         platform.
@@ -244,6 +222,39 @@ class Package:
         return Version(self.version) == Version(pkg.version)
 
 
+@ft.lru_cache
+def query_installed_packages() -> Dict[str, Package]:
+    """Uses conda to find out the versions of all packages installed in
+    $FSLDIR, which are sourced from the FSL conda channels.
+
+    Returns a dict of {pkgname : Package} mappings. The "dependencies"
+    attributes of the package objects are not populated.
+    """
+
+    channels = [PUBLIC_FSL_CHANNEL  .rstrip('/'),
+                INTERNAL_FSL_CHANNEL.rstrip('/')]
+
+    # conda info returns a list of dicts,
+    # one per package. We re-arrange this
+    # into a dict of {pkgname : version}
+    # mappings.
+    fsldir = os.environ['FSLDIR']
+    info   = json.loads(conda(f'list -p {fsldir} --json'))
+    pkgs   = {}
+
+    # We are only interested in packages
+    # hosted on the FSL conda channels
+    for pkg in info:
+        if pkg['base_url'].rstrip() in channels:
+            pkgs[pkg['name']] = Package(pkg['name'],
+                                        pkg['version'],
+                                        pkg['channel'],
+                                        pkg['platform'])
+
+    return pkgs
+
+
+@ft.lru_cache
 def download_channel_metadata(channel_url : str, **kwargs) -> Tuple[Dict, Dict]:
     """Downloads information about packages hosted at the given conda channel.
 
@@ -296,14 +307,19 @@ def download_channel_metadata(channel_url : str, **kwargs) -> Tuple[Dict, Dict]:
     return chandata, platdata
 
 
-def identify_packages(channeldata : List[Tuple[Dict, Dict]],
-                      pkgnames    : Sequence[str],
-                      development : bool) -> Dict[str, List[Package]]:
+def identify_packages(
+        channeldata : List[Tuple[Dict, Dict]],
+        pkgnames    : Sequence[str],
+        development : bool
+) -> Dict[str, List[Package]]:
     """Return metadata about the requested packages.
 
     Loads channel and platform metadata from the conda channels. Parses the
-    metadata, and creates a Package object for every requested package. Returns
-    a dict of {name : Package} mappings.
+    metadata, and creates a Package object for every requested package.
+
+    Returns a dict of {name : [Package]} mappings, where each entry contains
+    Package objects for all available versions of the packae, ordered from
+    oldest (first) to newest (last).
 
     channeldata:  Sequence of channel data from one or more conda channels, as
                   returned by the download_channel_metadata function.
@@ -338,7 +354,7 @@ def identify_packages(channeldata : List[Tuple[Dict, Dict]],
             for pkgfile in pdata[platform][pkgname]:
                 version   = pkgfile['version']
                 depends   = pkgfile['depends']
-                pkg       = Package(pkgname, curl, platform, version, depends)
+                pkg       = Package(pkgname, version, curl, platform, depends)
                 if pkg.development and (not development):
                     continue
                 bisect.insort(packages[pkgname], pkg)
@@ -346,15 +362,57 @@ def identify_packages(channeldata : List[Tuple[Dict, Dict]],
     return packages
 
 
+def filter_packages(packages : Dict[str, List[Package]]) -> List[Package]:
+    """Identifies the versions of packages that should be installed.
+
+    Removes packages that are not installed, or that are already up to date.
+
+    Returns a list of Package objects representing the packages/versions to
+    install.
+    """
+
+    filtered = []
+
+    for pkgname, pkgs in packages.items():
+
+        # Find the Package object corresponding
+        # to the installed version
+        installed = query_installed_packages().get(pkgname, None)
+
+        if not installed:
+            log.debug(f'Package {pkgname} is not installed - ignoring')
+            continue
+
+        # select the newest available
+        # version of the package as
+        # the installation candidate
+        pkg = pkgs[-1]
+
+        if pkg <= installed:
+            log.debug(f'{pkg.name} is already up to date (available: '
+                      f'{pkg.version}, installed: {installed.version}) '
+                      '- ignoring.')
+        else:
+            filtered.append(pkg)
+
+    return filtered
+
+
 def confirm_installation(packages : Sequence[Package], yes : bool) -> bool:
     """Asks the user for confirmation, before installing/updating the requested
     packages.
     """
+
     rows = [('Package name', 'Currently installed', 'Updating to'),
             ('------------', '-------------------', '-----------')]
 
+    # Currently all requested packages should already
+    # be installed, so a package should never be "n/a".
+    # This might change in the future if this script
+    # is enhanced to allow installation of new packages.
     for pkg in packages:
-        rows.append((pkg.name, pkg.installed_version, pkg.version))
+        installed = query_installed_packages().get(pkg.name, 'n/a')
+        rows.append((pkg.name, installed.version, pkg.version))
 
     len0 = max(len(r[0]) for r in rows)
     len1 = max(len(r[1]) for r in rows)
@@ -383,22 +441,22 @@ def install_packages(packages : Sequence[Package]):
     cmd      = f'install --no-deps -p {fsldir} -y ' + ' '.join(packages)
 
     print('\nInstalling packages...')
-    conda(cmd)
+    conda(cmd, False)
 
 
 def parse_args(argv : Optional[Sequence[str]]) -> argparse.Namespace:
     """Parses command-line arguments, returning an argparse.Namespace object. """
 
     parser = argparse.ArgumentParser(
-        'update_fsl_package', description='Install/update FSL packages')
+        'update_fsl_package', description='Update FSL packages')
     parser.add_argument('package', nargs='*',
-                        help='Package[s] to install/update')
+                        help='Package[s] to update')
     parser.add_argument('-d', '--development', action='store_true',
                         help='Install development versions of packages if available')
     parser.add_argument('-y', '--yes', action='store_true',
                         help='Install package[s] without prompting for confirmation')
     parser.add_argument('-a', '--all', action='store_true',
-                        help='Install/update all installed FSL packages')
+                        help='Update all installed FSL packages')
 
     parser.add_argument('--internal', help=argparse.SUPPRESS)
     parser.add_argument('--username', help=argparse.SUPPRESS)
@@ -408,8 +466,8 @@ def parse_args(argv : Optional[Sequence[str]]) -> argparse.Namespace:
     args = parser.parse_args(argv)
 
     if len(args.package) == 0 and not args.all:
-        parser.error('Specify at least one package, or use --all to update '
-                     'all installed FSL packages.')
+        parser.error('Specify at least one package, or use --all '
+                     'to update all installed FSL packages.')
 
     return args
 
@@ -427,7 +485,8 @@ def main(argv : Sequence[str] = None):
         argv = sys.argv[1:]
     args = parse_args(argv)
 
-    logging.basicConfig()
+    logging.basicConfig(format='%(funcName)s:%(lineno)d: %(message)s')
+    logging.getLogger().setLevel(logging.CRITICAL)
     if args.verbose: log.setLevel(logging.DEBUG)
     else:            log.setLevel(logging.INFO)
 
@@ -442,37 +501,22 @@ def main(argv : Sequence[str] = None):
             username=args.username,
             password=args.password))
 
-    if args.all:
-        packages = list(query_installed_packages().keys())
-    else:
-        packages = args.package
+    print('Building FSL package list ...')
+    if args.all: packages = list(query_installed_packages().keys())
+    else:        packages = args.package
 
     # Identify the versions that are
     # available for the packages the
     # user has requested.
-    packages = identify_packages(channeldata,
-                                 packages,
-                                 args.development)
-
-    to_install = []
-    for pkgs in packages.values():
-        # select the newest available
-        # versions of all packages
-        pkg = pkgs[-1]
-
-        if Version(pkg.version) <= Version(pkg.installed_version):
-            log.debug(f'{pkg.name} is already up to date (available: '
-                      f'{pkg.version}, installed: {pkg.installed_version}) '
-                      '- ignoring.')
-        else:
-            to_install.append(pkg)
+    packages = identify_packages(channeldata, packages, args.development)
+    packages = filter_packages(packages)
 
-    if len(to_install) == 0:
+    if len(packages) == 0:
         print('\nNo packages need updating.')
         sys.exit(0)
 
-    if confirm_installation(to_install, args.yes):
-        install_packages(to_install)
+    if confirm_installation(packages, args.yes):
+        install_packages(packages)
     else:
         print('Aborting update')