Как перечислить все пути к объектам в службе dbus?

Если это однократные задания (в отличие от периодических), то я думаю, что самый простой способ - просто использовать tmux и сон . Откройте сеанс tmux и сделайте следующее:

$ echo "ffmpeg -i a.mkv a.mp4" > midnight.sh
$ echo "ffmpeg -i b.mkv b.mp4" >> midnight.sh
$ let secs_till_midnight=$(date -d "`date -d '+1 day' +%F`" +%s)-$(date +%s)
$ sleep $secs_till_midnight && sh midnight.sh

Запланируйте больше заданий в новых панелях / окнах из того же сеанса, а когда вы закончите, отключитесь от этого сеанса. Вот и все! Ваш собственный, самодельный планировщик!

Даже если вы не можете перечислить запущенные задачи, добавьте эти две строки в начало midnight.sh для большей видимости:

set -x
exec 2>midnight_err >midnight_out

Теперь каждая команда будет выводиться на терминал, прежде чем она будет выполняется, и весь вывод команды перенаправляется в файлы midnight _ * .

Теперь просто присоедините к этому сеансу или хвостовому файлам журнала, чтобы проверить статус!

16
13.04.2017, 15:36
2 ответа

В соответствии с ответами#don _crissti , я реализовал это решение, которое дает имя интерфейса, метод и информацию о сигналах

import dbus
from xml.etree import ElementTree
bus = dbus.SystemBus()

def busNames():
    return [ name for name in  bus.list_names() if not name.startswith(":") ]


def pathNames(service,object_path="/",paths=None,serviceDict=None):
    if paths == None:
        paths = {}
    paths[object_path] = {}
    obj = bus.get_object(service, object_path)
    iface = dbus.Interface(obj, 'org.freedesktop.DBus.Introspectable')
    xml_string = iface.Introspect()
    root = ElementTree.fromstring(xml_string)
    for child in root:
        if child.tag == 'node':
            if object_path == '/':
                    object_path = ''
            new_path = '/'.join((object_path, child.attrib['name']))
            pathNames(service, new_path,paths)
        else:
            if object_path == "":
                object_path = "/"
            functiondict = {}
            paths[object_path][child.attrib["name"]] = functiondict
            for func in child.getchildren():
                if func.tag not in functiondict.keys():
                    functiondict[func.tag] =[]
                functiondict[func.tag].append(func.attrib["name"])
    if serviceDict == None:
        serviceDict = {}
    serviceDict[service] = paths
    return serviceDict



import json
import random
service=random.sample(busNames(),1).pop()
print service
print json.dumps(pathNames(service),indent=3)
2
27.01.2020, 19:48

Согласно частично полному ответу Reegan Miranda Я портировал код на python 3.9 и заставил его проанализировать всю шинную систему с помощью фильтра регулярных выражений.

#!/bin/python
#> usage:./query-dbus.py --pattern "^.*activit.*$"
import argparse
parser = argparse.ArgumentParser(description='query dbus tree')
parser.add_argument('-s','--system', help='query SystemBus', required=False, default=False)
parser.add_argument('-S','--session', help='query SessionBus (default)', required=False, default=True)
parser.add_argument('-p','--pattern', help='search pattern', required=False, default=None)
args = vars(parser.parse_args())

import re
pattern = None
if args['pattern']:
  pattern = re.compile(args['pattern'], re.IGNORECASE)

import dbus
from xml.etree import ElementTree

if args['session']:
  bus = dbus.SessionBus()
else:
  bus = dbus.SystemBus()

def busNames():
    return [ name for name in  bus.list_names() if not name.startswith(":") ]


def objectTree(service, object_path="/", paths=None, serviceDict=None):
    if str(service).startswith(':') == True: 
        return
      
    if paths == None:
        paths = {}
    pathdict = {}
    try:
      try:
          obj = bus.get_object(service, object_path)
      except dbus.exceptions.DBusException:
          print("[ERROR] service \"%s\" doesn't exist or isn't running" \
                % service)
          return
      
      try:
        iface = dbus.Interface(obj, 'org.freedesktop.DBus.Introspectable')
        xml_string = iface.Introspect()
      except dbus.exceptions.DBusException:
        print("[ERROR] service \"%s\" doesn't reply request. timeout." \
                % service)
        return
      root = ElementTree.fromstring(xml_string)
      for child in root:
          if child.tag == 'node':
              if object_path == '/':
                      object_path = ''
              new_path = '/'.join((object_path, child.attrib['name']))
              objectTree(service, new_path, paths)
          else:
              if object_path == "":
                  object_path = "/"
              functiondict = {}
            
              for func in child:
                if pattern is None or pattern.match(func.attrib["name"]):
                  if func.tag not in functiondict.keys():
                      functiondict[func.tag] = []
                  functiondict[func.tag].append(func.attrib["name"])
                  
              if functiondict:
                pathdict[child.attrib["name"]] = functiondict
      if serviceDict == None:
          serviceDict = {}
      if pathdict:
        paths[object_path] = pathdict
      if paths:
        serviceDict[service] = paths
      return serviceDict
    except:
      raise

import json
import random

for service in bus.list_names():
  object_tree = objectTree(service)
  if object_tree:
    print(json.dumps(object_tree,indent=3))
1
16.12.2020, 23:58

Теги

Похожие вопросы