course/w8/w8-s6-av-code.html at 72b3a74b90454429fb6da52a2a58fa4e0224f36c · flotpython/course · GitHub

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
<div class='clipboard'><style>
.clipboard {
    font-size: 11px;
}
</style>
<h3 style="border: 1px solid #66b; border-radius: 5px; padding: 10px; background-color: #eee;">
<b><i>bloc-note pour copier/coller le code de la vidéo</i></b>
</h3>
<hr />
<blockquote>
<p>Prendre un nouveau fichier sous IDLE ou votre éditeur python</p>
</blockquote>
<pre><code>from datetime import datetime
import asyncio
import time

###############################################################################
#
#  pour remplacer 'from asynchelpers import start_timer, show_timer'
#
###############################################################################

# low-level API
def _start_timer():
    return datetime.now()


def _show_timer(start, *args):
    delta = datetime.now() - start
    duration = f&quot;{delta.seconds}s + {delta.microseconds // 1000:03d}ms&quot;
    print(duration, *args)


####################
# Use a module global to keep things simple

glo_start = None


def start_timer():
    global glo_start
    print(&quot;---------- zero&quot;)
    glo_start = _start_timer()


def show_timer(*args):
    global glo_start
    _show_timer(glo_start, *args)


##############################
async def sequence(*messages, delay=1):
    show_timer(&quot;&gt;&gt;&gt;&quot;, *messages)
    await asyncio.sleep(delay)
    show_timer(&quot;&lt;&lt;&lt;&quot;, *messages)


##########
def reset_loop():
    asyncio.set_event_loop(asyncio.new_event_loop())

###############################################################################
#
#   fin de 'pour remplacer from asynchelpers import start_timer, show_timer'
#
###############################################################################
</code></pre>
<blockquote>
<p>Faire Save As &quot;w8s6 Vidéo.py&quot; puis F5</p>
</blockquote>
<pre><code># pour démonstration de l'utilitaire
start_timer()

time.sleep(1)

show_timer('message')
</code></pre>
<blockquote>
<p>Reprendre le fichier sous IDLE ou votre éditeur python</p>
</blockquote>
<pre><code>async def c1():
    show_timer(&quot;&gt;&gt;&gt; c1&quot;)
    await asyncio.sleep(1)
    show_timer(&quot;forking&quot;)
    # fork
    asyncio.ensure_future(c2())
    await asyncio.sleep(1)
    show_timer(&quot;&lt;&lt;&lt; c1&quot;)

# sera forkée par c1() après une seconde

async def c2():
    show_timer(&quot;&gt;&gt;&gt; c2&quot;)
    await asyncio.sleep(2)
    show_timer(&quot;&lt;&lt;&lt; c2&quot;)

async def main():
    # Il faut maintenant utiliser get_running_loop à la place de get_event_loop.
    loop = asyncio.get_running_loop()
    # On crée un objet future
    fut = loop.create_future()
    asyncio.ensure_future(c1())
    await fut

start_timer()

# interrompre après 2s
try:
    # run dans un terminal, idle, ipython. dans un notebook remplacer par await main()
    # asyncio.run() crée toujours une nouvelle boucle d'événements et la ferme à la fin.
    # Elle doit être utilisée comme point d'entrée principal pour les programmes asynchrones, et ne doit idéalement être appelée qu'une seule fois.
    asyncio.run(main())
except KeyboardInterrupt:
    print(&quot;bye&quot;)
</code></pre>
<blockquote>
<p>Faire Ctl-s puis F5
Pour interrompre la boucle &quot;sans fin&quot;, taper Ctl-c</p>
</blockquote>
<blockquote>
<p>réinitialisation de la boucle</p>
</blockquote>
<!-- Plus necessaire asyncio.run() gére.
<pre><code>asyncio.set_event_loop(asyncio.new_event_loop()) -->
</code></pre>
<blockquote>
<p>Reprendre le fichier sous IDLE ou votre éditeur python
et remplacer (après l'utilitaire) par</p>
</blockquote>
<pre><code>async def c1_stop(fut):
    show_timer(&quot;&gt;&gt;&gt; c1&quot;)
    await asyncio.sleep(1)
    show_timer(&quot;forking&quot;)
    # fork
    asyncio.ensure_future(c2_stop(fut))
    await asyncio.sleep(1)
    show_timer(&quot;&lt;&lt;&lt; c1&quot;)

# sera forkée par c1_stop() après une seconde
async def c2_stop(fut):
    show_timer(&quot;&gt;&gt;&gt; c2&quot;)
    await asyncio.sleep(2)
    show_timer(&quot;&lt;&lt;&lt; c2&quot;)
    # On set un résultat pour l'objet future afin de pouvoir quitter la boucle.
    fut.set_result(&quot;End&quot;)

async def main():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()
    asyncio.ensure_future(c1_stop(fut))

    print(await fut)

start_timer()

# s'arrête tout seul
try:
    asyncio.run(main())
except KeyboardInterrupt:
    print(&quot;bye&quot;)

print(&quot;done&quot;)
</code></pre>
<blockquote>
<p>Faire Ctl-s puis F5</p>
</blockquote>
<blockquote>
<p>Pour interrompre la boucle &quot;sans fin&quot;, taper Ctl-c</p>
</blockquote>
</div>