Agordejo/engine/findprograms.py

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Copyright 2020, Nils Hilbricht, Germany ( https://www.hilbricht.net )

This file is part of the Laborejo Software Suite ( https://www.laborejo.org ),

This is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""

import logging; logger = logging.getLogger(__name__); logger.info("import")

import pathlib
import configparser
import subprocess
import os
import stat

from engine.start import PATHS

def nothing(*args): pass

class SupportedProgramsDatabase(object):
    """Find all binaries with NSM support. Resources are:
    * Argodejo internal program list of known working programs.
    * Internal blacklist of known redundant programs (such as non-daw) or nonsense entries, like Argodejo itself    
    * A search through the users path to find stray programs that contain NSM announce messages
    * Finally, local to a users system: User whitelist for any program, user blacklist.
    Those two have the highest priority.

    We generate the same format as configParser does with .desktop files as _sections dict.

    Example:
     { 'categories': 'AudioVideo;Audio;X-Recorders;X-Multitrack;X-Jack;',
       'comment': 'Easy to use pattern sequencer for JACK and NSM',
       'comment[de]': 'Einfach zu bedienender Pattern-Sequencer',
       'exec': 'patroneo',
       'genericname': 'Sequencer',
       'icon': 'patroneo',
       'name': 'Patroneo',
       'startupnotify': 'false',
       'terminal': 'false',
       'type': 'Application',
       'x-nsm-capable': 'true'}

    In case there is a file in PATH or database but has no .desktop we create our own entry with
    missing data.

    """

    def __init__(self):        
        self.progressHook = nothing #prevents the initial programstart from sending meaningless messages for the cached data. Set and reverted in self.build
        
        self.grepexcluded = (pathlib.Path(PATHS["share"], "grepexcluded.txt")) #created by hand. see docstring
        #assert self.grepexcluded.exists()
        self.blacklist = ("nsmd", "non-daw", "carla", "argodejo", "adljack") #only programs that have to do with audio and music. There is another general blacklist that speeds up discovery
        self.whiteList = ("thisdoesnotexisttest", "patroneo", "vico",
                    "fluajho", "carla-rack", "carla-patchbay", "carla-jack-multi", "ardour5",
                    "ardour6", "nsm-data", "jackpatch", "nsm-proxy", "ADLplug", "ams",
                    "drumkv1_jack", "synthv1_jack", "samplv1_jack", "padthv1_jack",
                    "luppp", "non-mixer", "non-timeline", "non-sequencer", "non-midi-mapper", "non-mixer-noui",
                    "OPNplug", "qmidiarp", "qtractor", "zynaddsubfx", "jack_mixer",
                    "hydrogen", "mfp", "shuriken", "laborejo", "guitarix", "radium"                    
                    ) #shortcut list and programs not found by buildCache_grepExecutablePaths because they are just shellscripts and do not contain /nsm/server/announce.
        self.userWhitelist = () #added dynamically to morePrograms. highest priority
        self.userBlacklist = () #added dynamically to blacklist. highest priority
        self.knownDesktopFiles = { #shortcuts to the correct desktop files. Reverse lookup binary->desktop creates false entries, for example ZynAddSubFx and Carla.
            "zynaddsubfx": "zynaddsubfx-jack.desktop", #value will later get replaced with the .desktop entry
            "carla-jack-multi" : "carla.desktop",
        }
        self.programs = [] #list of dicts. guaranteed keys: argodejoExec, name, argodejoFullPath. And probably others, like description and version.
        self.nsmExecutables = set() #set of executables for fast membership, if a GUI wants to know if they are available. Needs to be build "manually" with self.programs. no auto-property for a list. at least we don't want to do the work.
        #.build needs to be called from the api/GUI.
        #self.unfilteredExecutables = self.buildCache_unfilteredExecutables() #This doesn't take too long. we can start that every time. It will get updated in build as well.
        self.unfilteredExecutables = None #in build()
        #self.build() #fills self.programs and

    def buildCache_grepExecutablePaths(self)->list:
        """return a list of executable names in the path (not the path itself)
        Grep explained:
        -s silent. No errors, eventhough subprocess uses stdout only
        -R recursive with symlinks. We don't want to look in subdirs because that is not allowed by
        PATH and nsm, but we want to follow symlinks

        If you have a custom user path that does not mean that all its executables will
        automatically show up here. They still need to contain /nsm/server/announce
        Your binaries will be in unfilteredExecutables though
        """
        result = []
        testpaths = os.environ["PATH"].split(os.pathsep) + ["/bin", "/sbin"]
        executablePaths = set([pathlib.Path(p).resolve() for p in os.environ["PATH"].split(os.pathsep)]) #resolve filters out symlinks, like arches /sbin and /bin. set() makes it unique

        excludeFromProcessingSet = set(self.blacklist + self.userBlacklist)
        whiteSet = set(self.whiteList + self.userWhitelist)
        excludeFromProcessingSet.update(whiteSet)
        
        for path in executablePaths:
            self.progressHook(f"{path}")
            command = f"grep --exclude-from {self.grepexcluded}  -iRsnl {path} -e /nsm/server/announce"           
            #command = f"grep -iRsnl {path} -e /nsm/server/announce"           
                        
            #Py>=3.7 completedProcess = subprocess.run(command, capture_output=True, text=True, shell=True)
            completedProcess = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, shell=True) #universal_newlines is an alias for text, which was deprecated in 3.7 because text is more understandable.  capture_output replaces the two PIPEs in 3.7
            for fullPath in completedProcess.stdout.split():
                self.progressHook(f"{fullPath}")
                exe = pathlib.Path(fullPath).relative_to(path)               
                if not str(exe) in excludeFromProcessingSet: #skip over any known file, good or bad
                    result.append((str(exe), str(fullPath)))

        for prg in whiteSet:
            self.progressHook(f"{prg}")
            for path in executablePaths:
                if pathlib.Path(path, prg).is_file(): #check if this actually exists
                    result.append((str(prg), str(pathlib.Path(path, prg))))
                    break #inner loop
        
        return list(set(result)) #make unique

    def buildCache_DesktopEntries(self):
        """Go through all dirs including subdirs"""
        xdgPaths = (
            pathlib.Path("/usr/share/applications"),
            pathlib.Path("/usr/local/share/applications"),
            pathlib.Path(pathlib.Path.home(), ".local/share/applications"),
            )

        config = configparser.ConfigParser()
        allDesktopEntries = []
        for basePath in xdgPaths:
            for f in basePath.glob('**/*'):
                self.progressHook(f"{f}")
                if f.is_file() and f.suffix == ".desktop":
                    config.clear()
                    try:
                        config.read(f)
                        entryDict = dict(config._sections["Desktop Entry"])
                        if f.name == "zynaddsubfx-jack.desktop":
                            self.knownDesktopFiles["zynaddsubfx"] = entryDict
                        elif f.name == "carla.desktop":
                            self.knownDesktopFiles["carla-jack-multi"] = entryDict
                        #in any case:
                        allDesktopEntries.append(entryDict)  #_sections   'DesktopEntry':{dictOfActualData)
                    except: #any bad config means skip
                        logger.warning(f"Bad desktop file. Skipping: {f}")                        
        return allDesktopEntries

    def loadPrograms(self, listOfDicts):
        """Qt Settings will send us this"""
        self.programs = listOfDicts
        self.nsmExecutables = set(d["argodejoExec"] for d in self.programs)

    def build(self, progressHook=None):
        """Can be called at any time by the user to update after installing new programs"""

        if progressHook:
            #receives one string which indicates what files is currently parsed.
            #Just the pure path, this will not get translated!
            #The purpose is to show a "we are not frozen!" feedback to the user.
            #It doesn't really matter what is reported back as long as it changes often            
            self.progressHook = progressHook 

        logger.info("Building launcher database. This might take a minute")
        self.progressHook("")
        self.programs = self._build()
        self.unfilteredExecutables = self.buildCache_unfilteredExecutables()
        self.nsmExecutables = set(d["argodejoExec"] for d in self.programs)
        self._buildWhitelist()
        self.progressHook("")
        self.progressHook = nothing
        logger.info("Building launcher database done.")

    def _exeToDesktopEntry(self, exe:str)->dict:
        """Assumes self.desktopEntries is up to date
        Convert one exe (not full path!) to one dict entry. """
        if exe in self.knownDesktopFiles: #Shortcut through internal database
            entry = self.knownDesktopFiles[exe]
        else: #Reverse Search desktop files.
            for entry in self.desktopEntries:
                if "exec" in entry and exe.lower() in entry["exec"].lower():
                    return entry
            #else: #Foor loop ended. Did not find any matching desktop file

        return None


    def _build(self):
        self.executables = self.buildCache_grepExecutablePaths()
        self.desktopEntries = self.buildCache_DesktopEntries()

        leftovers = set(self.executables)
        matches = [] #list of dicts
        for exe, fullPath in self.executables:
            self.progressHook(f"{fullPath}")
            entry = self._exeToDesktopEntry(exe)
            if entry: #Found match!
                entry["argodejoFullPath"] = fullPath
                #We don't want .desktop syntax like "qmidiarp %F"
                entry["argodejoExec"] = exe
                matches.append(entry)

                try:
                    leftovers.remove((exe,fullPath))
                except KeyError:
                    pass #Double entries like zyn-jack zyn-alsa etc.

            """
            if exe in self.knownDesktopFiles: #Shortcut through internal database
                entry = self.knownDesktopFiles[exe]
            else: #Reverse Search desktop files.
                for entry in self.desktopEntries:
                    if "exec" in entry and exe.lower() in entry["exec"].lower():
                        break #found entry. break inner loop to keep it
                else: #Foor loop ended. Did not find any matching desktop file
                    #If we omit continue it will just write exe and fullPath in any desktop file.
                    continue

            #Found match!
            entry["argodejoFullPath"] = fullPath
            #We don't want .desktop syntax like "qmidiarp %F"
            entry["argodejoExec"] = exe
            matches.append(entry)

            try:
                leftovers.remove((exe,fullPath))
            except KeyError:
                pass #Double entries like zyn-jack zyn-alsa etc.
               """

        for exe,fullPath in leftovers:
            pseudoEntry = {"name": exe.title(), "argodejoExec":exe, "argodejoFullPath":fullPath}
            matches.append(pseudoEntry)
        return matches

    def buildCache_unfilteredExecutables(self):
        def isexe(path):
            """executable by owner"""
            return path.is_file() and stat.S_IXUSR & os.stat(path)[stat.ST_MODE] == 64
        result = []
        executablePaths = [pathlib.Path(p) for p in os.environ["PATH"].split(os.pathsep)]
        for path in executablePaths:
            self.progressHook(f"{path}")            
            result += [str(pathlib.Path(f).relative_to(path)) for f in path.glob("*") if isexe(f)]

        return sorted(list(set(result)))

    def _buildWhitelist(self):
        """For reliable, fast and easy selection this is the whitelist.
        It will be populated from a template-list of well-working clients and then all binaries not
        in the path are filtered out. This can be presented to the user without worries."""
        #Assumes to be called only from self.build
        startexecutables = set(self.whiteList + self.userWhitelist)
        for prog in self.programs:
            prog["whitelist"] = prog["argodejoExec"] in startexecutables

        """
        matches = []
        for exe in startexecutables:
            entry = self._exeToDesktopEntry(exe)
            if entry: #Found match!
                #We don't want .desktop syntax like "qmidiarp %F"
                entry["argodejoExec"] = exe
                matches.append(entry)
        return matches
        """

programDatabase = SupportedProgramsDatabase()