I'm trying to get python ReactiveX stream (using RxPy library) to be sent to a javascript on Web UI component, but I can't seem to find a way to do so. Also, I might need to get the data stream coming into the Javascript into a RxJS Observable of sorts for further processing. Could you please help me understand how to achieve this? I'm still getting a grip on ReactiveX so maybe there are some fundamental concepts I'm missing, but I'm struggling to find anything similar to this around the net.
This issue has come up as I'm working on a desktop app that takes data from a csv or a zeromq endpoint, and streams it to a UI where the data will be plotted dynamically (updated the plot as new data comes in). I'm using Electron to build my app, using python as my backend code. Python is a must as I will be extending the app with some TensorFlow models.
Following fyears really well made example as an initial structure, I have written some sample code to play with but I can't seem to get it to work. I manage to get from the UI button all the way to the python scripts, but I get stuck in the return of the PricesApi.get_stream(...) method.
The front end is straight forward.
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Electron Application</title>
</head>
<body>
<button id="super-button">Trigger Python Code</button>
<div id="py-output">
</div>
</body>
<script src="renderer.js" ></script>
</html>
The ZeroRPC server file is like the one in the above mentioned link.
import gevent
import json
import signal
import zerorpc
from core_operator import stream
class PricesApi(object):
def get_stream(self, filename):
return stream(filename)
def stop(self):
print('Stopping strategy.')
def echo(self, text):
"""echo any text"""
return text
def load_settings():
with open('settings.json') as json_settings:
settings_dictionary = json.load(json_settings)
return settings_dictionary
def main():
settings = load_settings()
s = zerorpc.Server(PricesApi())
s.bind(settings['address'])
print(f"Initialising server on {settings['address']}")
s.run()
if __name__ == '__main__':
main()
This is the file were the major logic will sit to get prices from zeroMQ subscription, but currently just creates an Observable from a csv.
import sys
import rx
from csv import DictReader
def prepare_csv_timeseries_stream(filename):
return rx.from_(DictReader(open(filename, 'r')))
def stream(filename):
price_observable = prepare_csv_timeseries_stream(filename)
return price_observable
finally, the javascript that should be receiving the stream:
const zerorpc = require('zerorpc');
const fs = require('fs')
const settings_block = JSON.parse(fs.readFileSync('./settings.json').toString());
let client = new zerorpc.Client();
client.connect(settings_block['address']);
let button = document.querySelector('#super-button');
let pyOutput = document.querySelector('#py-output');
let filename = '%path-to-file%'
button.addEventListener('click', () => {
let line_to_write = '1'
console.log('button click received.')
client.invoke('get_stream', filename, (error, result) => {
var messages = pyOutput;
message = document.createElement('li'),
content = document.createTextNode(error.data);
message.appendChild(content);
messages.appendChild(message);
if(error) {
console.error(error);
} else {
var messages = pyOutput;
message = document.createElement('li'),
content = document.createTextNode(result.data);
message.appendChild(content);
messages.appendChild(message);
}
})
})
I have been looking into using WebSockets, but failed in understanding how to implement it. I did find some examples using Tornado server, however I am trying to keep it as pure as possible and, also, it feels odd that having already a client/server structure from Electron, I'm not able to use that directly. Also I'm trying to maintain the entire system a PUSH structure as the data requirements don't allow for a PULL type of pattern, with regular pollings etc.
Thank you very much in advance for any time you can dedicate to this, and please let me know if you require any further details or explanations.
I found a solution by using an amazing library called Eel (described as "A little Python library for making simple Electron-like HTML/JS GUI apps"). Its absolute simplicity and intuitiveness allowed me to achieve what I wanted a few simple lines.
import sys
import rx
from csv import DictReader
def prepare_csv_timeseries_stream(filename):
return rx.from_(DictReader(open(filename, 'r')))
def process_logic():
return pipe(
ops.map(lambda p: print(p)), # just to view what's flowing through
ops.map(lambda p: eel.receive_price(p)), # KEY FUNCTION in JS file, exposed via eel, is called for each price.
)
@eel.expose # Decorator so this function can get triggered from JavaScript
def stream(filename):
price_observable = prepare_csv_timeseries_stream(filename)
price_observable.pipe(process_logic()).subscribe() # apply the pipe and subscribe to trigger stream
eel.init('web')
eel.start('main.html') # look at how beautiful and elegant this is!
let button = document.querySelector('#super-button');
let pyOutput = document.querySelector('#py-output' );
let filename = '%path-to-file%'
console.log("ready to receive data!")
eel.expose(receive_price); // Exposing the function to Python, to process each price
function receive_price(result) {
var messages = pyOutput;
message = document.createElement('li');
content = document.createTextNode(result);
message.appendChild(content);
messages.appendChild(message);
// in here you can add more functions to process data, e.g. logging, charting and so on..
};
button.addEventListener('click', () => {
console.log('Button clicked magnificently! Bloody good job')
eel.stream(filename); // calling the Python function exposed through Eel to start stream.
})
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Let's try Eel</title>
</head>
<body>
<h1>Eel-saved-my-life: the App!</h1>
<button id="super-button">Trigger Python Code</button>
<div id="py-output">
</div>
</body>
<script type="text/javascript" src="/eel.js"></script>
<script type="text/javascript" src="price_processing.js"></script>
</html>
I hope this can help anyone struggling with the same problem.