9
votes

Capturer la sortie d'impression de la bibliothèque partagée appelée de Python avec le module CTTYPES

Je travaille avec une bibliothèque partagée qui est appelée via le module CTTYPES. Je voudrais rediriger le stdout associé à ce module à une variable ou à un fichier que je peux accéder à mon programme. Cependant, CTTYPES utilise un stdout séparé de SYS.STDOUT.

Je vais démontrer le problème que j'ai avec Libc. Si quelqu'un est copié et collant le code, il peut être nécessaire de modifier le nom de fichier sur la ligne 2. xxx

est là que je peux capturer le stdout qui est associé aux CTTPES chargés Bibliothèque partagée?

python

0 commentaires

3 Réponses :

6
votes

Nous pouvons utiliser os.dup2 () code> et os.pipe () code> pour remplacer l'intégralité du descripteur de fichier STDOUT (FD 1) avec un tuyau que nous pouvons lire de nous-mêmes . Vous pouvez faire la même chose avec stardr (fd 2).

Cet exemple utilise Sélectionnez.select () code> pour voir si le tuyau (notre faux stdout) a des données en attente d'être écrits, alors nous peut l'imprimer en toute sécurité sans bloquer l'exécution de notre script. p>

Comme nous remplacons complètement le descripteur de fichier STDOUT pour ce processus et toutes les sous-processus, cet exemple peut même capturer la production des processus enfants. P>

import os, sys, select

# the pipe would fail for some reason if I didn't write to stdout at some point
# so I write a space, then backspace (will show as empty in a normal terminal)
sys.stdout.write(' \b')
pipe_out, pipe_in = os.pipe()
# save a copy of stdout
stdout = os.dup(1)
# replace stdout with our write pipe
os.dup2(pipe_in, 1)

# check if we have more to read from the pipe
def more_data():
        r, _, _ = select.select([pipe_out], [], [], 0)
        return bool(r)

# read the whole pipe
def read_pipe():
        out = ''
        while more_data():
                out += os.read(pipe_out, 1024)

        return out

# testing print methods
import ctypes
libc = ctypes.CDLL('libc.so.6')

print 'This text gets captured by myStdOut'
libc.printf('This text fails to be captured by myStdOut\n')

# put stdout back in place 
os.dup2(stdout, 1)
print 'Contents of our stdout pipe:'
print read_pipe()

0 commentaires

2
votes

exemple le plus simple, parce que cette question dans Google Top. XXX

0 commentaires

0
votes

Si les données que le processus natif écrit sont volumineux (plus grand que le tampon de tuyau), le programme natif bloquerait jusqu'à ce que vous fassiez de l'espace dans le tuyau en le lisant.

La solution de Lunixbochs nécessite cependant le processus natif. pour finir avant qu'il ne commence à lire le tuyau. J'ai amélioré la solution afin qu'elle lit le tuyau en parallèle à partir d'un fil séparé. De cette façon, vous pouvez capturer la sortie de n'importe quelle taille. P>

Cette solution est également inspirée par https://stackoverflow.com / A / 16571630/1076564 capture à la fois STDOUT et STDERR: P>

class CtypesStdoutCapture(object):
    def __enter__(self):
        self._pipe_out, self._pipe_in = os.pipe()
        self._err_pipe_out, self._err_pipe_in = os.pipe()
        self._stdout = os.dup(1)
        self._stderr = os.dup(2)
        self.text = ""
        self.err = ""
        # replace stdout with our write pipe
        os.dup2(self._pipe_in, 1)
        os.dup2(self._err_pipe_in, 2)
        self._stop = False
        self._read_thread = threading.Thread(target=self._read, args=["text", self._pipe_out])
        self._read_err_thread = threading.Thread(target=self._read, args=["err", self._err_pipe_out])
        self._read_thread.start()
        self._read_err_thread.start()
        return self

    def __exit__(self, *args):
        self._stop = True
        self._read_thread.join()
        self._read_err_thread.join()
        # put stdout back in place
        os.dup2(self._stdout, 1)
        os.dup2(self._stderr, 2)
        self.text += self.read_pipe(self._pipe_out)
        self.err += self.read_pipe(self._err_pipe_out)

    # check if we have more to read from the pipe
    def more_data(self, pipe):
        r, _, _ = select.select([pipe], [], [], 0)
        return bool(r)

    # read the whole pipe
    def read_pipe(self, pipe):
        out = ''
        while self.more_data(pipe):
            out += os.read(pipe, 1024)

        return out

    def _read(self, type, pipe):
        while not self._stop:
            setattr(self, type, getattr(self, type) + self.read_pipe(pipe))
            sleep(0.001)

    def __str__(self):
        return self.text

# Usage:

with CtypesStdoutCapture as capture:
  lib.native_fn()

print(capture.text)
print(capture.err)

0 commentaires