[Kodi Addon] MTP Connect - for Android on Raspberry Pi

Wenn du dir nicht sicher bist, in welchem der anderen Foren du die Frage stellen sollst, dann bist du hier im Forum für allgemeine Fragen sicher richtig.
Antworten
harryberlin
User
Beiträge: 227
Registriert: Donnerstag 17. Dezember 2015, 12:17

Da es im anderen Thread etwas vom topic abkommt, gibts hier nen eigenen thread.

ausgangscode war dieser:

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# sudo apt-get install jmtpfs

# sudo mkdir ~/Android
# sudo chmod 775 ~/Android
# sudo jmtpfs -o allow_other ~/Android

# sudo umount ~/Android
# sudo rmdir ~/Android


import os
import sys
import time
import xbmc
import xbmcgui
import xbmcaddon
import subprocess
from threading import Thread

__author__ = 'harryberlin'

DEVICE_NAMES = ['Android', 'userdefined']
ANDROID_PATH = os.path.expanduser("~") + "/Android"

APT_UPDATE = ['sudo', 'apt-get', 'update']
INSTALL_JMTPFS = ['sudo', 'apt-get', 'install', '-y', 'jmtpfs']
REMOVE_JMTPFS = ['sudo', 'apt-get', 'remove', '-y', 'jmtpfs']
MAKE_DIR = ['sudo', 'mkdir', '%s' % ANDROID_PATH]
REMOVE_DIR = ['sudo', 'rmdir', '%s' % ANDROID_PATH]
MOUNT_DEVICE = ['sudo', 'jmtpfs', '-o allow_other', '%s' % ANDROID_PATH]
UNMOUNT_DEVICE = ['sudo', 'umount', '%s' % ANDROID_PATH]


def ok(message1, message2=" ", message3=" "):
    xbmcgui.Dialog().ok(heading="MTP Connect", line1=message1, line2=message2, line3=message3)


def note(header, message=" ", time=3000):
    __icon__ = xbmcaddon.Addon().getAddonInfo('path') + '/icon.png'
    xbmcgui.Dialog().notification(heading=' %s' % header, message=' %s' % message, icon=__icon__, time=time)
    log('NOTIFICATION: ' + header + ' - ' + message)


def log(message):
    xbmc.log('plugin.script.mtpconnect: %s' % message)


def busy_show():
    xbmc.executebuiltin("ActivateWindow(busydialog)")


def busy_close():
    xbmc.executebuiltin("Dialog.Close(busydialog)")


def open_settings():
    xbmcaddon.Addon().openSettings()


def install():
    busy_show()
    log('update package libary')
    f = subprocess.check_output(APT_UPDATE)
    log(f)

    log('install package and add mount-path')

    f = subprocess.check_output(INSTALL_JMTPFS)
    log(f)

    f = subprocess.check_output(MAKE_DIR)
    log(f)
    busy_close()
    note("Package installed", "Path %s created" % ANDROID_PATH)


def deinstall():
    busy_show()
    log('delete directory')
    f = subprocess.check_output(REMOVE_JMTPFS)
    log(f)

    log('delete directory')
    f = subprocess.check_output(REMOVE_DIR)
    log(f)
    busy_close()
    note("Package deinstalled", "Path %s removed" % ANDROID_PATH)


def connect():
    disconnect(False)

    ok('Some Devices only share their Folders,', 'when Screen is unlocked.', 'UNLOCK YOUR DISPLAY')

    log('try to mount')
    f = subprocess.check_output(MOUNT_DEVICE)
    log(f)

    if f.find('No mtp devices found') > 0 or f.find('bad mount point') > 0 or f.find('jmtpfs: command not found') > 0:
        note("CAN'T CONNECT DEVICE")

    elif f.find('Device 0') > 0 or f.find('Device 1') > 0 or f.find('Device 2') > 0 or f.find('Device 3') > 0:
        note("Android Device connected", time=800)
        DIR_CHECKER.start()
    else:
        pass


def disconnect(shownotification=True):
    log('try to unmount')
    f = subprocess.check_output(UNMOUNT_DEVICE)
    log(f)
    if shownotification: note("Android Device disconnected")


def is_device_on():
    counter = 0
    while counter < 1:
        # log("check path %s" % counter)
        try:
            if os.listdir(ANDROID_PATH) == []:
                pass
            else:
                counter = -1
            time.sleep(1)
            counter += 1
        except:
            break
    disconnect()


def main():
    count = len(sys.argv) - 1
    if count > 0:
        given_args = sys.argv[1].split(';')
        if str(given_args[0]) == "install":
            install()
        elif str(given_args[0]) == "deinstall":
            deinstall()
        elif str(given_args[0]) == "connect":
            connect()
        elif str(given_args[0]) == "disconnect":
            disconnect(False)
        elif str(given_args[0]) == "settings":
            open_settings()
        else:
            note(header='Unknown Arguments given!', message='install, connect, disconnect are available', time=5000)

    else:
        open_settings()


DIR_CHECKER = Thread(name="dirchecker", target=is_device_on)
DIR_CHECKER.setDaemon(True)

if __name__ == '__main__':
    main()
danach hat blackjack hand angelegt und hat folgendes dazu beigetragen:
BlackJack hat geschrieben:@harryberlin: Anmerkungen zum Code zusätzlich zu dem was Sirius3 schon gesagt hat:

Wenn man die Kommandozeilenargumente über ein Wörterbuch auf Funktionen abbildet, kann man die Meldung bei einem unbekannten Kommando leichter fehlerfrei hinbekommen, weil man die aus den Schlüsseln erstellen kann, statt das man immer daran denken muss bei Veränderungen bei den Kommandos auch diese Ausgabe entsprechend anzupassen.

`DEVICE_NAMES` wird nicht verwendet. `DIRCHECKER` ist keine Konstante und hat dort auf Modulebene auch nichts zu suchen. Dem Thread einen Namen zu geben, macht nicht wirklich Sinn.

Warum der Name `__icon__` in `note()`? Namen mit zwei führenden und folgenden Unterstrichen sind per Konvention für Werte vorgesehen die im Python-Ökosystem eine besondere Bedeutung haben, also entweder mit der Sprache oder mit Werkzeugen auf besondere Weise interagieren. Beispiele wären die ganzen ”magischen” Methoden die festlegen wie sich Objekte verhalten (Operatoren, ``for``, ``with``, …) oder `__doc__`, `__author__`, `__version__`, die von Hilfesystemen und Dokumentationsgeneratoren ausgewertet werden.

Da `busy_show()` und `busy_close()` immer in Paaren vorkommen, die Code ”umschliessen”, würde es sich anbieten da einen Contextmanager draus zu machen und ``with`` zu verwenden.

Wenn Du grundsätzlich die Ausgabe von externen Programmen protokollieren willst, dann bietet sich ja alleine deshalb schon eine Funktion dafür an. Damit man nicht jedes mal zusammen mit dem Aufruf noch einen `log()`-Aufruf schreiben muss.

Die `is_device_on()`-Funktion sieht sehr abenteuerlich aus. Erst einmal kann man den ``pass``-Zweig eleminieren. Dann sollte Ausnahmebehandlung präziser sein. Also weniger Code und nur die Ausnahme(n) umfassen, die man hier auch tatsächlich erwartet. Das ist also wohl das `os.listdir()` und eine `OSError`-Ausnahme, für den Fall das einem das Verzeichnis vor der Nase weggelöscht wird. Andererseits ist das `os.listdir()` in der Schleife die einzige Quelle für einen `OSError` und die Aktion ist das verlassen der Schleife. Also kann man die Ausnahmebehandlung auch um die Schleife legen, und damit den Sonderfall aus der Schleife herausziehen. Wenn man das gemacht hat, fällt auf, dass diese Ausnahmebehandlung dort nichts macht, also eigentlich nur dazu da ist das `disconnect()` davor zu schützen von einer Ausnahme übergangen zu werden. Also kann man ``try``/``finally`` verwenden.

Jetzt gilt es noch den gruseligen `counter` weg zu bekommen der nur Werte die Werte -1, 0, und 1 annehmen kann, aber eigentlich nur ein Wahrheitswert ist, der ungünstig ausgedrückt wurde. Denn eigentlich will man ja, dass die Schleife solange wartet, bis keine Dateien mehr in dem Pfad vorhanden sind. Das lässt sich *wesentlich* simpler ausdrücken.

Ich lande dann ungefähr hier (ungetestet):

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: utf-8 -*-
 
# sudo apt-get install jmtpfs
 
# sudo mkdir ~/Android
# sudo chmod 775 ~/Android
# sudo jmtpfs -o allow_other ~/Android
 
# sudo umount ~/Android
# sudo rmdir ~/Android

import os
import subprocess
import sys
import time
from contextlib import contextmanager
from threading import Thread

import xbmc
import xbmcaddon
import xbmcgui
 
__author__ = 'harryberlin'

ANDROID_PATH = os.path.expanduser('~/Android')
 
APT_UPDATE = ['sudo', 'apt-get', 'update']
INSTALL_JMTPFS = ['sudo', 'apt-get', 'install', '-y', 'jmtpfs']
REMOVE_JMTPFS = ['sudo', 'apt-get', 'remove', '-y', 'jmtpfs']
MAKE_DIR = ['sudo', 'mkdir', ANDROID_PATH]
REMOVE_DIR = ['sudo', 'rmdir', ANDROID_PATH]
MOUNT_DEVICE = ['sudo', 'jmtpfs', '-o', 'allow_other', ANDROID_PATH]
UNMOUNT_DEVICE = ['sudo', 'umount', ANDROID_PATH]
 
 
def log(message):
    xbmc.log('plugin.script.mtpconnect: %s' % message)


def notify(header, message=' ', display_duration=3000):
    xbmcgui.Dialog().notification(
        heading=header,
        message=message,
        icon=os.path.join(xbmcaddon.Addon().getAddonInfo('path') + 'icon.png'),
        time=display_duration,
    )
    log('NOTIFICATION: %s - %s' % (header, message))


@contextmanager
def busy_shown():
    xbmc.executebuiltin('ActivateWindow(busydialog)')
    try:
        yield
    finally:
        xbmc.executebuiltin('Dialog.Close(busydialog)')

 
def ask_confirmation(message1, message2=' ', message3=' '):
    return xbmcgui.Dialog().ok(
        heading='MTP Connect', line1=message1, line2=message2, line3=message3
    )


def open_settings():
    xbmcaddon.Addon().openSettings()


def call(command):
    try:
        try:
            output = subprocess.check_output(command, stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as error:
            output = error.output
            return error.returncode
        else:
            return 0
    finally:
        log(output)


def install():
    with busy_shown():
        log('update package libary')
        call(APT_UPDATE)
        log('install package and add mount-path')
        call(INSTALL_JMTPFS)
        call(MAKE_DIR)
    # 
    # FIXME This message is wrong if any of the calls above did not
    #   succeed.
    # 
    notify('Package installed', 'Path %s created' % ANDROID_PATH)
 
 
def deinstall():
    with busy_shown():
        log('remove package')
        call(REMOVE_JMTPFS)
        log('delete directory')
        call(REMOVE_DIR)
    # 
    # FIXME This message is wrong if any of the calls above did not
    #   succeed.
    # 
    notify('Package deinstalled', 'Path %s removed' % ANDROID_PATH)
 

def disconnect(show_notification=False):
    log('try to unmount')
    call(UNMOUNT_DEVICE)
    # 
    # FIXME This message may be wrong if any of the calls above did
    #   not succeed.
    # 
    if show_notification:
        notify('Android Device disconnected')

  
def wait_for_unmount():
    try:
        # 
        # FIXME Use a better indicator than an empty path.
        #   `os.path.ismount()` for instance.  Maybe even replace
        #   polling by `pyinotify` where available.
        # 
        while os.listdir(ANDROID_PATH):
            time.sleep(1)
    finally:
        disconnect(True)
 

def connect():
    disconnect()
    ask_confirmation(
        'Some Devices only share their Folders,',
        'when Screen is unlocked.',
        'UNLOCK YOUR DISPLAY'
    )
    log('try to mount')
    if call(MOUNT_DEVICE) == 0:
        notify('Android Device connected', 800)
        thread = Thread(target=wait_for_unmount)
        thread.daemon = True
        thread.start()
    else:
        notify("CAN'T CONNECT DEVICE")
 
 
def main():
    command_name2func = {
        'connect': connect,
        'deinstall': deinstall,
        'disconnect': disconnect,
        'install': install,
    }
    if len(sys.argv) > 1:
        command_name = sys.argv[1].partition(';')[0]
        command = command_name2func.get(command_name)
        if command:
            command()
        else:
            notify(
                'Unknown command given!',
                '{0} are available'.format(
                    ', '.join(sorted(command_name2func))
                ),
                5000
            )
    else:
        open_settings()


if __name__ == '__main__':
    main()
empty Sig
harryberlin
User
Beiträge: 227
Registriert: Donnerstag 17. Dezember 2015, 12:17

zu "wait_for_unmount():"
hier soll überwacht werden, wenn man das usb kabel trennt. da nicht autom. ummounted wird. wenn ich jetzt auf ismount umstelle, checkt es die funktion, obwohl nicht geunmounted wurde?

der counter ist drin, weil damit in nem größeren intervall nach einer vorhandenen device sozusagen "gesucht" wird. ja, dann muss man es etwas umschreiben.

__icon__ wird üblicherweise in kodi so verwendet.

und wie schon erwähnt, das log soll nicht immer drin sein.

und ja, die install usw. können auch fehlerhaft beendet werden, wo die auswertung dazu noch fehlt. das sollte ein einfaches flag setzen, wenn ungleich 0 zurück kommt, mit entsprechenden hinweise. oder gleich die funktion verlassen.
in der form etwa:

Code: Alles auswählen

def install():
    with busy_shown():
        log('update package libary')
        if call(APT_UPDATE) != 0:
            ask_confirmation('error with %s' % APT_UPDATE)
            return
        log('install package and add mount-path')
        if call(INSTALL_JMTPFS) != 0:
            ask_confirmation('error with %s' % INSTALL_JMTPFS)
            return
        if call(MAKE_DIR) != 0:
            ask_confirmation('error with %s' % MAKE_DIR)
            return
    notify('Package installed', 'Path %s created' % ANDROID_PATH)
empty Sig
BlackJack

@harryberlin: Was man am besten in `wait_for_unmount()` macht, hängt davon ab was passiert wenn man das Kabel trennt. Der USB-Treiber bekommt das ja mit und es kann durchaus sein, dass damit dann auch automatisch das Dateisystem ausgehängt wird. Falls das der Fall ist, könnte man `ismount()` verwenden. Oder `pyinotify()` damit man nicht „busy waiting“ betreiben muss.

`__icon__` wird in Kodi sicher nicht als lokaler Name für irgend etwas verwendet. Vielleicht kann man das ja auf Modulebene setzen, damit Kodi da nachschaut wenn es ein Plugin importiert hat, aber an lokale Namen kommt es sicher nicht ran.

Die `install()` die Du zeigst ist dann aber wieder genau der Grund warum man mit Ausnahmen arbeiten möchte und nicht mit Rückgabwerten die einen Fehler anzeigen. Denn das führt genau zu solchen wiederkehrenden Codemustern die man als Programmierer vermeidet. Wenn `call()` ganz einfach den `CalledProcessError` auslösen würde, kann man den in der `install()`-Funktion an *einer* Stelle für alle Aufrufe behandeln:

Code: Alles auswählen

def install():
    try:
        with busy_shown():
            log('update package libary')
            call(APT_UPDATE)
            log('install package and add mount-path')
            call(INSTALL_JMTPFS)
            call(MAKE_DIR)
    except subprocess.CalledProcessError as error:
        ask_confirmation('error with %s' % error.cmd)
    else:
        notify('Package installed', 'Path %s created' % ANDROID_PATH)
harryberlin
User
Beiträge: 227
Registriert: Donnerstag 17. Dezember 2015, 12:17

was gibt error.cmd dann aus, den gesendeten befehl?
warum eigentlich nicht finally, statt else?

bin recht prakmatisch gestrikt, beispiele helfen mir da am besten.

warum hast du die funktion eigentlich call genannt? kann ja eigentlich alles mögliche bedeuten.
empty Sig
BlackJack

@harryberlin: Pragmatisch gestrickt sein sollte nicht davon abhalten die Dokumentation zu lesen. Was das `cmd`-Attribut enthält steht dort, und man könnte das auch ganz einfach mal ausprobieren:

Code: Alles auswählen

In [67]: try: subprocess.check_output(['false'])
   ....: except Exception as error: pass
   ....: 

In [68]: error.cmd
Out[68]: ['false']

In [69]: try: subprocess.check_output(['ls', 'not existent'])
   ....: except Exception as error: pass
   ....: 
ls: cannot access not existent: No such file or directory

In [70]: error.cmd
Out[70]: ['ls', 'not existent']
Die Benachrichtigung passiert in einem ``else``-Zweig weil sie *nicht* erscheinen soll falls der Text in ihr nicht zutrifft.

`call()` weil es eine Funktion in `subprocess` gibt, die so heisst. `call_and_log()` wäre vielleicht noch ein bisschen passender. Falls sie so umgeschrieben wird, dass sie eine Ausnahme auslöst, dann wäre `check_call_and_log()` angebrachter. :-)
Antworten