Discussion :

[vekev]

Bonjour à tous,

Je travaille actuellement sur un script dont le but est de transmettre des instructions à un serveur, avec websocket.
Voici comment j'ai vu les choses :

1. Je démarre mon script, qui demande l'exécution d'une première instruction (demande de sauvegarde)
2. La connexion au serveur est initialisée, une boucle infinie transmet les instructions et récupère les résultats. On demande d'afficher le contenu de la console présente sur le serveur afin de pouvoir récupérer les résultats des instructions.
3. La connexion au serveur est faite, la première instruction de mon script est transmise (demande de sauvegarde) et attend les résultats
4. La boucle infinie récupère l'instruction, la transmet au serveur qui l'interprète et retourne les résultats
5. Le résultat est retournée à la fonction qui a demandé la sauvegarde
6. [...]

Mon problème est que je bloque à l'étape 2, le script étant paralysé par le thread contenant la boucle infinie.
Pourquoi cela fait ça en sachant que j'ai justement mis la fonction concernée dans un nouveau thread (module threading) ?

La gestion de la connexion au serveur (avec un nouveau thread)

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
 
    def start_ws_client(self):
        if not self.is_started_ws_client:
            self.stop_ws_client = False
            thread_ws = threading.Thread(target= asyncio.run(self.ws_client()))
            thread_ws.start()
 
    async def ws_client(self):
        url = self.get_websocket_url()
        cc = self.get_console_connection()
 
        async with websockets.connect(url, additional_headers= cc.get_headers(), origin=cc.get_https_url_server()) as ws:
            command = json.dumps({
                'event': 'auth',
                'args': [self.websockets_token]
            })
            await ws.send(command)
 
            response = await ws.recv()
            auth_response = json.loads(response)
            if not auth_response['event'] == 'auth success':
                raise TypeError('Auth failed')
 
            self.is_started_ws_client = True
 
            while not self.stop_ws_client:
                response = await ws.recv()
                msg = json.loads(response)
 
                if msg['event'] == 'console output':
                    self.logs.append(msg['args'][0])
 
            self.is_started_ws_client = False

Mon thread principal appele "start_ws_client()", qui lance la tâche asynchrone dans un nouveau thread.
En l'état actuel, la connexion au serveur est bien faite, je récupère des données sur la santé du serveur mais pour pouvoir transmettre mes instructions, je dois pouvoir exécuter du code en dehors de la boucle, tout en sachant qu'elle doit rester active (d'où l'usage d'un nouveau thread).

Avez-vous une idée ?
Merci.

[wiztricks]

Salut,

Avez-vous une idée ?

La programmation avec des threads et asyncio n'est pas affaire de débutant.
Dit autrement, soit vous maîtrisez et êtes capable de résoudre les soucis rencontrés tout seul, soit il va être compliqué de pouvoir vous aider sans fournir un code minimal permettant de reproduire ce que vous constatez (et pas des morceaux choisis par vous).

- W

[vekev]

J'avoue être peu habitué à python, ayant plus l'habitude du C# et de l'environnement .Net.

Voici le code utilisé :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
from sources.consoleConnection import ConsoleConnection
 
cc = ConsoleConnection()
 
if cc.send_save_all():
    cc.send_broadcast("Sauvegarde du serveur en cours")
 
cc.stop_websockets_connection()

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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
from enum import Enum
 
import requests
import json
 
from sources import commandConsoleResult
 
class ServerCommand(Enum):
    SAVE_ALL = 0
    SAVE_ON = 1
    SAVE_OFF = 2
    BROADCAST = 3
 
class ConsoleConnection:
    statusCode_success = 204
    properties_json: str = ''
    websockets_connection = None
 
    """
    Returns headers for all connections
    """
    def get_headers(self):
        return {
            'Accept': 'application/json',
            'Content-type': 'application/json',
            'Authorization': 'Bearer {0}'.format(self.get_apikey())
        }
 
    """
    Returns API key
    """
    def get_apikey(self):
        return self.properties_json['apikey']
 
    """
    Returns the contents of ./private/properties.json
    """
    def get_properties_json(self):
        if len(self.properties_json) == 0:
            f = open("../private/properties.json", 'r')
            self.properties_json = json.loads(f.read())
            f.close()
 
        return self.properties_json
 
    """
    Send a command with args
    """
    def send_command(self, command, args):
        srv_cmd = self.get_command(self, command)
 
        if not args is None and len(args) > 0:
            srv_cmd = '{0} {1}'.format(srv_cmd, args)
 
        url = '{0}/servers/{1}/command'.format(self.get_https_url_server_api_client(), self.get_server_id())
        response = requests.post(
            url,
            headers= self.get_headers(),
            json = {
                'command': srv_cmd
            }
        )
 
        # 204 = successful
        return response.status_code == self.statusCode_success
 
    """
    Try to send a command.
    Returns true if success, else an error.
    """
    def try_send_command(self, command, message=None):
        is_success = self.send_command(command, message)
 
        if not is_success:
            raise TypeError('Cannot execute {0}'.format(command))
 
        return is_success
 
    """
    Send a broadcast command with a message
    """
    def send_broadcast(self, message):
        return self.try_send_command(ServerCommand.BROADCAST, message)
 
    """
    Returns a websockets connection instance
    """
    def get_websockets_connection(self):
        from sources.webSocketsConnection import WebSocketsConnection
 
        if self.websockets_connection is None:
            self.websockets_connection = WebSocketsConnection()
 
        return self.websockets_connection
 
    """
    Stop websockets connection
    """
    def stop_websockets_connection(self):
        if not self.websockets_connection is None:
            self.websockets_connection.stop_ws_client()
 
    """
    Send a save_all command
    """
    def send_save_all(self):
        ws_connection = self.get_websockets_connection()
 
        ws_connection.start_ws_client()
 
        # get latest "save_all" line in logs
        latest_line_from_server = ws_connection.get_latest_line_logs()
 
        if self.try_send_command(ServerCommand.SAVE_ALL):
            # check if console show expected messages
            result_method = ws_connection.check_logs_to_last_line(
                latest_line_from_server,
                commandConsoleResult.get_all_message_save_all()
            )
            return result_method.get_value()
        else:
            return False
 
    """
    Returns the command string corresponding to the content of the argument  
    """
    @staticmethod
    def get_command(command):
        if not isinstance(command, ServerCommand):
            raise TypeError('command must be an instance of ServerCommand Enum')
 
        if command == ServerCommand.BROADCAST:
            return 'broadcast'
        elif command == ServerCommand.SAVE_OFF:
            return 'save-off'
        elif command == ServerCommand.SAVE_ALL:
            return 'save-all'
        else:
            return 'save-on'
 
    """
    Returns server ID
    """
    def get_server_id(self):
        self.properties_json = self.get_properties_json()
 
        if len(self.properties_json['server_id']) == 0:
            url = self.get_https_url_server_api_client()
            response = requests.get(url, headers= self.get_headers())
 
            if response.status_code == 200:
                data = response.json()
                self.properties_json['server_id'] = data["data"][0]["attributes"]["identifier"]
            else:
                return None
 
        return self.properties_json['server_id']
 
    """
    Returns the url for accessing the client APIs 
    """
    def get_https_url_server_api_client(self):
        self.properties_json = self.get_properties_json()
        return '{0}/api/client'.format(self.get_https_url_server())
 
    """
    Returns the server url 
    """
    def get_https_url_server(self):
        self.properties_json = self.get_properties_json()
        return 'https://{0}'.format(self.properties_json['server_url'])

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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import asyncio
import json
import threading
 
import websockets
import requests
 
from sources.consoleConnection import ConsoleConnection
from sources.resultMethod import ResultMethod
 
class WebSocketsConnection:
    websockets_token: str = ''
    websockets_url_socket: str = ''
    url_instance: str = ''
 
    logs = []
    stop_ws_client = False
    is_started_ws_client = False
    consoleConnection = None
    threadWs = None
 
    """
    Get an ConsoleConnection instance
    """
    def get_console_connection(self):
        if self.consoleConnection is None:
            self.consoleConnection = ConsoleConnection()
 
        return self.consoleConnection
 
    """
    Define token and url for websockets
    """
    def get_websocket_credentials(self):
        cc = self.get_console_connection()
 
        self.websockets_token = ''
        self.websockets_url_socket = ''
 
        url = f"{cc.get_https_url_server_api_client()}/servers/{cc.get_server_id()}/websocket"
        response = requests.get(url, headers= cc.get_headers())
 
        if response.status_code == 200:
            json_object = json.loads(response.content)
 
            self.websockets_token = json_object['data']['token']
            self.websockets_url_socket = json_object['data']['socket']
            return True
        else:
            return False
 
    """
    Check if token is valid
    """
    def is_valid_credentials(self):
        if len(self.websockets_token) == 0:
            self.get_websocket_credentials()
 
        return len(self.websockets_token) == 0
 
    """
    Check if url sockets is valid and returns it
    """
    def get_websocket_url(self):
        if len(self.websockets_url_socket) == 0:
            self.get_websocket_credentials()
 
        return self.websockets_url_socket
 
    """
    WebSocket client. Make connection and save all logs.
    """
    async def ws_client(self):
        url = self.get_websocket_url()
        cc = self.get_console_connection()
 
        async with websockets.connect(url, additional_headers= cc.get_headers(), origin=cc.get_https_url_server()) as ws:
            command = json.dumps({
                'event': 'auth',
                'args': [self.websockets_token]
            })
            await ws.send(command)
 
            response = await ws.recv()
            auth_response = json.loads(response)
            if not auth_response['event'] == 'auth success':
                raise TypeError('Auth failed')
 
            self.is_started_ws_client = True
 
            while not self.stop_ws_client:
                response = await ws.recv()
                msg = json.loads(response)
 
                if msg['event'] == 'console output':
                    self.logs.append(msg['args'][0])
 
            self.is_started_ws_client = False
 
    """
    Start websocket client
    """
    def start_ws_client(self):
        if not self.is_started_ws_client:
            self.stop_ws_client = False
            thread_ws = threading.Thread(target= asyncio.run(self.ws_client()))
            thread_ws.start()
 
    """
    Stop websocket client
    """
    def stop_ws_client(self):
        self.stop_ws_client = True
 
    """
    Return latest line present in logs.
    If logs are empty, return an empty string.
    """
    def get_latest_line_logs(self):
        if len(self.logs) > 0:
            return self.logs.pop()
        else:
            return ''
 
    """
    Check if lines are present in logs, from an specific line
    """
    def check_logs_to_last_line(self, from_line, contains_lines):
        if len(self.logs) > 0:
            # return all values from
            all_logs = self.logs
 
            try:
                start = all_logs.index(from_line)
 
                for i in range(start, len(all_logs)):
                    current_line_logs = all_logs[i]
 
                    for j in range(len(contains_lines)):
                        expected_line = contains_lines[j]
 
                        if expected_line in current_line_logs:
                            return ResultMethod(True, expected_line)
 
                return ResultMethod(False, '')
            except ValueError:
                return ResultMethod(False, '')
        else:
            return ResultMethod(False, '')

[wiztricks]

Salut,

Ce que vous avez posté est loin d'être un code "minimal".... et à par "lire", impossible de reproduire quoi que ce soit facilement (en recopiant ce que vous avez fourni chez moi). Néanmoins j'espère (pour vous) que d'autres seront plus inspirés.

- W

[vekev]

Malheureusement je ne peux pas aller plus loin, mis à part fournir également l'url de connexion au serveur et la clé API qui va avec.

En tout cas, le programme se contente de boucler la portion de code (dans "WebSocketsConnection") entre les lignes 91 et 96 (ce qui est plutôt logique puisque c'est ce que je demande) mais tout en paralysant le thread principal alors que j'ai demandé à ce que ce soit exécuté dans un autre thread.

Merci tout de même d'avoir jeté un oeil

[papajoker]

bonjour

Il faut partir d'un code minimal et non de ton app.
Puisque tu utilises asyncio, existe cette lib

2 minis fichiers qui simulent ton besoin (?), tirés de cette lib et légèrement modifiés pour une boucle particulière dans le client

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
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = [ "websockets>=15" ]
# ///
 
"""Client using the asyncio API."""
 
import asyncio
import random
from websockets.asyncio.client import connect
 
 
async def process(all, msg):
    tempo = random.randint(0, 1)
    await asyncio.sleep(tempo)  # PAS bloquer l'exécution des autres tâches asyncio.
    print(f"< {msg}")
    all.append(msg)
    return all
 
 
async def hello():
    content = []
    async with connect("ws://localhost:8888") as websocket:
        await websocket.send("send login/pass")
        message = await websocket.recv()
        print(message)
 
        # while True:
        async for message in websocket:
            content = await process(content, message)
            if len(content) == 5:
                print("traiter ces datas json:", "[" + ",".join(content) + "]")  # todo? thread traiter le retour
                content = []
                break  # ou pas
 
        print("END")
 
 
if __name__ == "__main__":
    asyncio.run(hello())

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
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = [ "websockets>=15" ]
# ///
 
"""Server example using the asyncio API."""
 
import asyncio
import time
import random
from websockets.asyncio.server import serve
 
 
async def hello(websocket):
    name = await websocket.recv()
    print(f"<<< {name}")
 
    resp = f"Hello {name}!"
 
    await websocket.send(resp)
    print(f">>> {resp}")
 
    for i in range(1, 50):
        time.sleep(random.randint(1, 6) // 10)
        await websocket.send(f"{{ id: {i} }}")
 
 
async def main():
    async with serve(hello, "localhost", 8888) as server:
        await server.serve_forever()
 
 
if __name__ == "__main__":
    asyncio.run(main())