Bonjour,
Je ne souhaite pas installer de librairies et je les ai donc téléchargé et placé dans un répertoire lib de mon projet.
Cependant je suis obligé de renseigner le path à chaque reprises et dans chacune de mes classes.
Auriez-vous une solution pour gérer cela ?
J'ai remarqué dans certains projets que les gens géraient les imports dans les __init__ de chaque packages.
Est ce une bonne pratique et cela me permettrait t'il de faire une seule fois le path et ensuite d'appeler via from . import ou from .. import ?
Ma seconde question ...
Mon programme possède deux classes qui héritent de Thread.
Ma première classe est un serveur web sous Flask.
La seconde classe est un serveur Socket.
L'idée est la suivante. Lorsque l'utilisateur réalise une action, celle-ci est transmise à tous les clients qui écoutent le serveur socket.
Pour cela mon serveur web doit pouvoir accéder aux fonctions du serveur Socket.
Cependant je ne peux pas importer mon serveur socket dans mon serveur web si non je suis en présence d'une boucle (les deux serveurs étant tous deux appelés dans un seul serveur).
Ci-joint mon code (qui n'est qu'une ébauche), j'ai trouvé une solution qui ne me plait pas... J'ai passé mon objet serveur socket à ma classe serveur Flask (partie commentée dans mon code)
Code : Sélectionner tout - Visualiser dans une fenêtre à part
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32 #!/usr/bin/env python -B # -*- coding: utf-8 -*- from utils.debugger import printer from controller.server.zWeb import ZwebServer from controller.server.zSocket import ZsocketServer class ZServeurMain(): def __init__(self, pArgs): printer('__init__()', __file__) self.args = pArgs #self.runSocketServer() self.runWebServer() print 'Démarrage des deux serveurs' self.zWeb.join() print 'Extinction du server web' #self.zSocket.set_isRunning(False) #self.zSocket.join() print 'stop all thread' exit(0) def runWebServer(self): printer('runWebServer()', __file__) #self.zWeb = ZwebServer(self.zSocket,'', self.args.get_web_port()) self.zWeb = ZwebServer(None,'', self.args.get_web_port()) self.zWeb.start() def runSocketServer(self): printer('runSocketServer()', __file__) self.zSocket = ZsocketServer('',5000) self.zSocket.start()
Code : Sélectionner tout - Visualiser dans une fenêtre à part
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84 #/usr/bin/env python # -*- coding:utf-8 -*- import os import sys import socket, select from threading import Thread from utils.debugger import printer from utils.utilsFonction import init_var from utils.globals import RECV_BUFFER, DEFAULT_SOC_PORT, DEFAULT_SOC_HOST HOST = DEFAULT_SOC_HOST PORT = DEFAULT_SOC_PORT class ZsocketServer(Thread): def __init__(self, pHost, pPort): printer('__init__()', __file__) global HOST, PORT HOST = init_var(HOST,pHost) PORT = init_var(PORT,pPort) Thread.__init__(self) self.isRunning = True self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((HOST, PORT)) self.server_socket.listen(10) self.connection_list = [] self.connection_list.append(self.server_socket) def run(self): while self.isRunning: print 'on boucle' read_sockets,write_sockets,error_sockets = select.select(self.connection_list,[],[]) print 'on boucle 1' for sock in read_sockets: if sock == self.server_socket: sockfd, addr = self.server_socket.accept() self.connection_list.append(sockfd) print "Client (%s, %s) connected" % addr self.broadcast_data(sockfd, "essai") else: try: data = sock.recv(RECV_BUFFER) if data: self.broadcast_data(sock, "\r" + '<' + str(sock.getpeername()) + '> ' + data) except: self.broadcast_data(sock, "Client (%s, %s) is offline" % addr) print "Client (%s, %s) is offline" % addr sock.close() self.connection_list.remove(sock) continue print 'fermeture socket' self.server_socket.close() def broadcast_data (self, sock, message): for socket in self.connection_list: if socket != self.server_socket and socket != sock : try : socket.send(message) except : socket.close() self.connection_list.remove(socket) def transmitte_order(self, message): for socket in self.connection_list: if socket != self.server_socket: try : socket.send(message) except: socket.close() self.connection_list.remove(socket) def get_isRunning(self): return self.isRunning def set_isRunning(self, pValue): self.isRunning = pValue print 'passage à {}'.format(pValue)
Code : Sélectionner tout - Visualiser dans une fenêtre à part
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63 #/usr/bin/env python # -*- coding:utf-8 -*- import os import sys sys.path.append(os.path.abspath('{}/../../../../lib/'.format(os.path.realpath(__file__)))) from flask import Flask, request, render_template, make_response from threading import Thread from utils.debugger import printer from utils.utilsFonction import init_var from utils.globals import DEFAULT_WEB_HOST, DEFAULT_WEB_PORT print os.path.abspath('{}/../../../view/templates'.format(os.path.realpath(__file__))) app = Flask(__name__, template_folder=os.path.abspath('{}/../../../view/templates'.format(os.path.realpath(__file__)))) HOST = DEFAULT_WEB_HOST PORT = DEFAULT_WEB_PORT SOCKET = None class ZwebServer(Thread): def __init__(self, pSocket, pHost, pPort): printer('__init__()', __file__) global HOST, PORT, SOCKET HOST = init_var(HOST,pHost) PORT = init_var(PORT,pPort) SOCKET = init_var(SOCKET,pSocket) Thread.__init__(self) def run(self): app.run(host=HOST, port=PORT) @app.route('/shutdown', methods=['GET', 'POST']) def shutdown(): shutdown_server() return 'Server shutting down...' @app.route('/', methods=['GET', 'POST']) def pg_accueil(): return 'Accueil !' @app.route('/acc') def pg_acc(): return render_template('accueil.html') @app.route('/transmit', methods=['GET', 'POST']) def pg_transmitte(): SOCKET.transmitte_order('test') return 'Transmit !' @app.errorhandler(404) def ma_page_404(error): return "Ma jolie page error 404" def shutdown_server(): func = request.environ.get('werkzeug.server.shutdown') if func is None: raise RuntimeError('Not running with the Werkzeug Server') func()
Je suis à l'écoute de toutes les critiques possibles sur mon code pour pouvoir l'améliorer !!!
En vous remerciant d'avance.
Cordialement Quentin
Partager