The idea here is to send push notifications from server at anytime, by keeping a connection open.
Last active
October 27, 2024 12:52
-
-
Save Cheaterman/6b19b99002b5197f5693014a05ed720e to your computer and use it in GitHub Desktop.
Kivy/Flask data streaming example
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import socket | |
import queue | |
import threading | |
from kivy.app import App | |
from kivy.clock import Clock | |
from kivy.lang import Builder | |
import requests | |
KV = r''' | |
ScrollView: | |
Label: | |
id: label | |
text: 'You are example_name\n' | |
size_hint_y: None | |
height: self.texture_size[1] | |
''' | |
class Stream(App): | |
def build(self): | |
self.queue = queue.Queue() | |
Clock.schedule_interval(self.update_stream, 0) | |
threading.Thread( | |
target=self.stream_worker, | |
daemon=True, | |
).start() | |
root = Builder.load_string(KV) | |
self.label = root.ids.label | |
return root | |
def update_stream(self, *args): | |
try: | |
data = self.queue.get(block=False) | |
except queue.Empty: | |
return | |
self.label.text += f'You won {data} points!\n' | |
def stream_worker(self): | |
self.response = requests.get( | |
'http://localhost:5000/example_name', | |
stream=True | |
) | |
while True: | |
for data in self.response.iter_lines( | |
chunk_size=1, | |
decode_unicode=True | |
): | |
self.queue.put(data) | |
Stream().run() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import time | |
import queue | |
from flask import Flask, Response, request | |
app = Flask(__name__) | |
queues = {} | |
def stream(name): | |
queues[name] = stream_queue = queue.Queue() | |
while True: | |
try: | |
yield f'{stream_queue.get()}\n' | |
except GeneratorExit: | |
break | |
del queues[name] | |
@app.route('/<name>') | |
def index(name): | |
return Response(stream(name), mimetype='text/event-stream') | |
@app.route('/give', methods=('GET', 'POST')) | |
def give(): | |
if request.method == 'POST' and request.form['name'] in queues: | |
queues[request.form['name']].put(request.form['points']) | |
return ( | |
'<form method="post">' | |
' Name to give to: <input name="name"><br>' | |
' Points to give: <input name="points"><br>' | |
' <input type="submit">' | |
'</form>' | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hello Cheaterman, thanks for your code. It looks exactly what i need for my project but i don't know why this example not works in my case.

I started kivy part with
python client.py
and flask part withexport FLASK_APP=server.py & flask run
.When i open in broswer url
http://127.0.0.1:5000/give
i see your form, but when i fill fields and submit it, nothing happend. Kivy client is still in state "You are example_name" (in attachment).Problem is probably on my side, but i don't know what i am doing wrong :/
Please, can you help me how to start your gist?
Thank you.