improving snap / snapd installation check latency

CHANGELOG.md

@@ -23,6 +23,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Reading installed Snaps now takes around 95% less time
 - Reading Snap suggestions now takes around 75% less time
 - Reading installed Flatpaks now takes around 45% less time
+- "snap" and "snapd" installation check latency
 - Refreshing only the associated package type after a successful operation (install, uninstall, downgrade, ...)
 - When a package is installed, it will be the first element of the table after refreshing
 - Progress bar status can now be controlled by the software manager while an operation is being executed

bauh/gems/snap/controller.py

@@ -56,29 +56,35 @@ class SnapManager(SoftwareManager):
         return app
 
     def search(self, words: str, disk_loader: DiskCacheLoader, limit: int = -1) -> SearchResult:
-        installed = self.read_installed(disk_loader).installed
+        if snap.is_snapd_running():
+            installed = self.read_installed(disk_loader).installed
 
-        res = SearchResult([], [], 0)
+            res = SearchResult([], [], 0)
 
-        for app_json in snap.search(words):
+            for app_json in snap.search(words):
 
-            already_installed = None
+                already_installed = None
 
-            if installed:
+                if installed:
-                already_installed = [i for i in installed if i.id == app_json.get('name')]
+                    already_installed = [i for i in installed if i.id == app_json.get('name')]
-                already_installed = already_installed[0] if already_installed else None
+                    already_installed = already_installed[0] if already_installed else None
 
-            if already_installed:
+                if already_installed:
-                res.installed.append(already_installed)
+                    res.installed.append(already_installed)
-            else:
+                else:
-                res.new.append(self.map_json(app_json, installed=False, disk_loader=disk_loader))
+                    res.new.append(self.map_json(app_json, installed=False, disk_loader=disk_loader))
 
-        res.total = len(res.installed) + len(res.new)
+            res.total = len(res.installed) + len(res.new)
-        return res
+            return res
+        else:
+            return SearchResult([], [], 0)
 
     def read_installed(self, disk_loader: DiskCacheLoader, limit: int = -1, only_apps: bool = False, pkg_types: Set[Type[SoftwarePackage]] = None) -> SearchResult:
-        installed = [self.map_json(app_json, installed=True, disk_loader=disk_loader) for app_json in snap.read_installed()]
+        if snap.is_snapd_running():
-        return SearchResult(installed, None, len(installed))
+            installed = [self.map_json(app_json, installed=True, disk_loader=disk_loader) for app_json in snap.read_installed()]
+            return SearchResult(installed, None, len(installed))
+        else:
+            return SearchResult([], None, 0)
 
     def downgrade(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher) -> bool:
         return ProcessHandler(watcher).handle(SystemProcess(subproc=snap.downgrade_and_stream(pkg.name, root_password), wrong_error_phrase=None))
@@ -136,7 +142,7 @@ class SnapManager(SoftwareManager):
         pass
 
     def list_warnings(self) -> List[str]:
-        if snap.get_snapd_version() == 'unavailable':
+        if not snap.is_snapd_running():
             return [self.i18n['snap.notification.snapd_unavailable']]
 
     def _fill_suggestion(self, pkg_name: str, priority: SuggestionPriority, out: List[PackageSuggestion]):

bauh/gems/snap/resources/locale/en

@@ -1,3 +1,3 @@
-snap.notification.snapd_unavailable=snapd seems not to be enabled. snap packages will not be available.
+snap.notification.snapd_unavailable=snapd seems not to be installed or enabled. snap packages will not be available.
 snap.action.refresh.status=Refreshing
 snap.action.refresh.label=Refresh

bauh/gems/snap/resources/locale/es

@@ -7,6 +7,6 @@ snap.info.revision=revisión
 snap.info.tracking=tracking
 snap.info.installed=instalado
 snap.info.publisher=publicador
-snap.notification.snapd_unavailable=snapd no parece estar habilitado. los paquetes snap estarán indisponibles.
+snap.notification.snapd_unavailable=snapd no parece estar instalado o habilitado. los paquetes snap estarán indisponibles.
 snap.action.refresh.status=Actualizando
 snap.action.refresh.label=Actualizar

bauh/gems/snap/resources/locale/pt

@@ -7,6 +7,6 @@ snap.info.revision=revisão
 snap.info.tracking=tracking
 snap.info.installed=instalado
 snap.info.publisher=publicador
-snap.notification.snapd_unavailable=snapd não parece estar habilitado. os pacotes snap estarão indisponíveis.
+snap.notification.snapd_unavailable=snapd não parece estar instalado ou habilitado. os pacotes snap estarão indisponíveis.
 snap.action.refresh.status=Atualizando
 snap.action.refresh.label=Atualizar

bauh/gems/snap/snap.py

@@ -1,7 +1,6 @@
 import logging
 import re
 import subprocess
-import time
 from typing import List
 
 from bauh.commons.system import new_root_subprocess, run_cmd, new_subprocess
@@ -11,28 +10,19 @@ BASE_CMD = 'snap'
 
 
 def is_installed():
-    version = get_snapd_version()
+    res = run_cmd('which snap')
-    return False if version is None else True
+    return res and not res.strip().startswith('which ')
 
 
-def get_version():
+def is_snapd_running() -> bool:
-    res = run_cmd('{} --version'.format(BASE_CMD), print_error=False)
+    services = new_subprocess(['systemctl', 'list-units'])
-    return res.split('\n')[0].split(' ')[-1].strip() if res else None
 
+    for o in new_subprocess(['grep', 'snapd.socket'], stdin=services.stdout).stdout:
+        if o:
+            if ' running ' in o.decode():
+                return True
 
-def get_snapd_version():
+    return False
-    res = run_cmd('{} --version'.format(BASE_CMD), print_error=False)
-
-    if not res:
-        return None
-    else:
-        lines = res.split('\n')
-
-        if lines and len(lines) >= 2:
-            version = lines[1].split(' ')[-1].strip()
-            return version if version and version.lower() != 'unavailable' else None
-        else:
-            return None
 
 
 def app_str_to_json(app: str) -> dict: