Bonjour à tous,

Voilà je bosse sur une application graphique (PyQt/PySide) qui devra me permettre d'effectuer un certain nombre de test via la framework unittest ou PyUnit.

Je dois utiliser pexpect pour lancer un serveur JTAG et lui envoyer des commandes, j'ai donc écrit la classe ModTelnet qui se connecte et ses méthodes send, receive, et stop pour respectivement envoyer/recevoir des commandes et arrêter la connection serveur comme le montre le code ci-dessous :

"""
Communication with embedded board
"""
import sys 
import time
import threading
import Queue

import pexpect
import serial
import fdpexpect
 
from pexpect import EOF, TIMEOUT
            

class ModTelnet():
    def __init__(self):    

    self.is_running = False
    self.HOST = 'localhost'
    self.port = '4444'

    def connect(self):
        #connect to board with JTAG and OpenOCD
		self.is_running = True
		password = 'xxxx'
		myfile = 'openocd.cfg'
		self._ocd = pexpect.spawn('sudo openocd -f %s' % (myfile))
		i = self._ocd.expect(['password', EOF, TIMEOUT])
		if i == 0:
			self._ocd.sendline(password)
			time.sleep(1.0)
			self._connect_to_tn()
		elif i == 1:
			print ' *** OCD Connection failed *** '
			raise Disconnected()
		elif i == 2:
			print ' *** OCD Connection timeout *** '
			raise Timeout()  
        
    def _connect_to_tn(self):
         #connect to telnet session @ localhost port 4444
         self._tn = pexpect.spawn('telnet %s %s' % (self.HOST, self.port))
         condition = self._tn.expect(['>', EOF, TIMEOUT])
         if condition == 0:
			print 'Telnet opened with success'
         elif condition == 1:
            print self._tn.before
            raise Disconnected()
         elif condition == 2: 
            print self._tn.before
            raise Timeout()

   def receive(self):
        #receive data (= msg) from telnet stdout
        data = [ EOF, TIMEOUT, '>' ]
        index = self._tn.expect(data, 2)
        if index == 0:
            return 'eof', None
        elif index == 1:
            return 'timeout', None
        elif index == 2:
            print 'success', self._tn.before.split('\r\n')[1:]
            return 'success',self._tn.before

    def send(self, command):
	print 'sending command: ', command
        self._tn.sendline(command)

	
    def stop(self):
	print 'Connection stopped !'
	self._ocd.sendcontrol('c')

if __name__ =='__main__':
    try:
	tn = ModTelnet()
	tn.connect()   
    except:
	   print 'Cannot connect to board!'
	   exit(0)

C'est sûrement tout bête comme problème...mais je n'arrive pas à utiliser mes classes send, receive et stop sans générer une erreur du style

Alors comment doit-je faire? sachant qu'il me faut absolument pouvoir accéder à ces méthodes depuis un autre module ??

Merci pour toutes vos réponses !
Cordialement

self._tn.sendline(command)

Rien à faire je ne trouve pas cette fonction

En même temps si tu peux mettre le traceback complet ça arrangerait tout le monde

La fonction sendline fait partie du module pexpect et self._tn correspond au child du process lancer avec pexpect.spawn. Étais-ce ta question?


Pour le traceback ... la carte est seulement au boulot je ne peut donc pas reproduire l'erreur désolé !

Citation

La fonction sendline fait partie du module pexpect et self._tn correspond au child du process lancer avec pexpect.spawn. Étais-ce ta question?

zut j'avais pas vu ton self._tn autant pour moi

child = pexpect.spawn('scp foo myname@host.example.com:.')
child.expect ('Password:')
child.sendline (mypassword)

Edit :

import pexpect
from pexpect import EOF, TIMEOUT

ça c'est très moche!!!

Citation : fred1599

Citation

La fonction sendline fait partie du module pexpect et self._tn correspond au child du process lancer avec pexpect.spawn. Étais-ce ta question?

Oui parce-que l'on doit hériter de la classe pexpect pour avoir le droit d'écrire self.quelquechose.

Dans ton cas tu n'hérites de rien.

Il faut donc exécuter tes fonctions de ta classe pexpect normalement, par exemple,

child = pexpect.spawn('scp foo myname@host.example.com:.')
child.expect ('Password:')
child.sendline (mypassword)

Oui alors, ce que tu me dis est vrai mais comment faire pour conserver mon child d'une méthode à l'autre dans ma classe?
Est-tu sûr que cela ne fonctionne pas, car dans moc cas j'ai déja pu le tester et cela marche, mon problème ne viens pas de là je crois.

Citation : fred1599

<citation>Edit :

import pexpect
from pexpect import EOF, TIMEOUT

ça c'est très moche!!!

Effectivement j'avoue c'est moche, j'aurais simplement pu faire

from pexpect import *

, c'est un simple oubli de ma part !

J'ai édité j'avais pas vu self._tn.

Sinon la fonction expect s'utilise avec une liste?

Edit : Oui on peut, désolé je découvre pexpect

Tu ne sais pas à quel endroit se trouve ton erreur? tu ne te souviens plus de ton traceback?

Citation : fred1599

J'ai édité j'avais pas vu self._tn.

Sinon la fonction expect s'utilise avec une liste?

Désolé, je n'ai pas vu l'edit.

Oui on l'utilise avec une liste par exemple :

i = child.expect([condition1, condition2, condition3])
if i == 0:
condition1 vrai alors ...
elif i == 1:
condition2 vrai alors ...
elif i == 2:
condition3 vrai alors ...
...

Erf edit:

Je sait que mon erreur est de type AttributeError comme je l'ai dit auparavant, en réalité je suis parti du code suivant que j'ai allégé (en quelque sorte) pour ne pas utiliser de thread :

"""
Communication with embedded board
"""
import sys 
import time
import threading
import Queue   
from pexpect import *

 
            
class Disconnected(BaseException):
    pass

class Timeout(BaseException):
    pass
    
class MXComm():
    def __init__(self):
        self.is_connected = False
        self._tn_to_gui = Queue.Queue()
        self._gui_to_tn = Queue.Queue()
        self._tx_item = None
        
    def _connect(self):
        pass

    def _receive(self):
        pass

    def _send(self, command):
        pass

    def connect(self):
        self._connect()
        self.is_connected = True

    def try_send_data(self):
        try:
            #get an item from the queue, try to send it
            #if the item wasn't sent, we don't get another one
            if self._tx_item is None:
                self._tx_item = self._gui_to_tn.get_nowait()
            self._send(self._tx_item)
            self._tx_item = None #everything's ok, item sent
        except Queue.Empty:
            pass
        except Timeout:
            pass
        except Disconnected:
            self.is_connected = False

    #get data from telnet session to gui
    def try_get_data(self):
        try:
            rx_item = self._receive()
            self._tn_to_gui.put_nowait(rx_item)
        except Timeout:
            pass
        except Disconnected:
            self.is_connected = False
            

class ModTelnet(MXComm):
    def __init__(self):    
        MXComm.__init__(self)
    
    self.is_running = False
    self.HOST = 'localhost'
    self.port = '4444'

    def connect(self):
        #connect to board with JTAG and OpenOCD
		self.is_running = True
		password = 'xxxx'
		myfile = 'openocd.cfg'
		self._ocd = pexpect.spawn('sudo openocd -f %s' % (myfile))
		i = self._ocd.expect(['password', EOF, TIMEOUT])
		if i == 0:
			self._ocd.sendline(password)
			time.sleep(1.0)
			self._connect_to_tn()
		elif i == 1:
			print ' *** OCD Connection failed *** '
			raise Disconnected()
		elif i == 2:
			print ' *** OCD Connection timeout *** '
			raise Timeout()  
        
    def _connect_to_tn(self):
         #connect to telnet session @ localhost port 4444
         self._tn = pexpect.spawn('telnet %s %s' % (self.HOST, self.port))
         condition = self._tn.expect(['>', EOF, TIMEOUT])
         if condition == 0:
			print 'Telnet opened with success'
         elif condition == 1:
            print self._tn.before
            raise Disconnected()
         elif condition == 2: 
            print self._tn.before
            raise Timeout()

   def receive(self):
        #receive data (= msg) from telnet stdout
        data = [ EOF, TIMEOUT, '>' ]
        index = self._tn.expect(data, 2)
        if index == 0:
            return 'eof', None
        elif index == 1:
            return 'timeout', None
        elif index == 2:
            print 'success', self._tn.before.split('\r\n')[1:]
            return 'success',self._tn.before

    def send(self, command):
	print 'sending command: ', command
        self._tn.sendline(command)

            
       
class MODBoard(threading.Thread):
    """
    Provides communication with an ARM board
    It's using pexpect, openOCD and telnet to detect and communicate with the board.
    The class creates a thread with two queue to read/write data asynchronously.
    """
    def __init__(self, connection):
        threading.Thread.__init__(self)
        
        self._conn = connection
        
        self._dispatch = Dispatcher()
        
        self._is_running = True
        self.start()

    def __del__(self):
        self.stop()
        if self.isAlive():
            self.join()

    def stop(self):
        self._is_running = False
    
    def read_packet(self):
        try:
            return  self._get_packet()
        except Queue.Empty:
            return           
            
    def _get_packet(self):
        """ Get a message from telnet or the openOCD session (non blocking call), 
        raise queue empty if no message available """
        return self._conn._tn_to_gui.get_nowait() 
                #self._conn.openocd_to_gui.get_nowait())
            
    def send_packet(self, command):
        """Send a message to the telnet or Openocd session (non-blocking call)"""
        self._conn._gui_to_tn.put_nowait(command)
        
    def is_connected(self):
        return self._conn.is_connected

    def run(self):
        """Thread main: read and write alternatively"""
        while self._is_running:
                if not self._conn.is_connected:
                    try:
                        self._conn.connect()
                    except :
                        time.sleep(1.0)
                        
                else:
                    #-------- PC -> Board -----
                    self._conn.try_send_data()
                    #-------- Board -> PC -----
                    self._conn.try_get_data()  
                    
if __name__ =='__main__':
    try:
        tn = ModTelnet()
        #connect to telnet session
        board = MODBoard(tn)

        
    except:
        print 'Cannot connect to Board!'
        exit(0)

Le problème c'est que j'ai un asynchronisme deplus à gérer en utilisant les Queues, j'ai donc préférer m'en passer.

dans ton 1er code en ligne 19 t'as un problème d'indentation

en ligne 54 aussi il te manque un espace

dans ton 2eme code, ligne 68 --> indentation

ligne 103, il manque un espacement pour une bonne indentation

Salut, et merci fred pour ton aide !

J'ai finalement trouver mon erreur :

Class ModTelnet():

Cela venais de la déclaration de ma classe, lorsque l'on écrit sa propre classe sans héritage on écrit normalement :

Class ModTelnet:

Ce que je trouve bizarre c'est que python me retourne une erreur sur l'instanciation de mes méthodes alors que le problème venait de l'instanciation même de ma classe...

Bref problème résolu !

J’espère que sa servira à d'autres en guise d'exemples