@@ -693,10 +693,91 @@ The first example shows how to use a client capabilities to predict stock price
693
693
694
694
* sources - `stock-predictions.ipynb <https://github.com/influxdata/influxdb-client-python/blob/master/notebooks/stock-predictions.ipynb >`_
695
695
696
- The second example shows how to use a client capabilities to realtime visualization via `hvPlot <https://hvplot.pyviz.org >`_, `Streamz <https://streamz.readthedocs.io/en/latest/ >`_:
696
+ The second example shows how to use a client capabilities to realtime visualization via `hvPlot <https://hvplot.pyviz.org >`_, `Streamz <https://streamz.readthedocs.io/en/latest/ >`_, ` RxPY < https://rxpy.readthedocs.io/en/latest/ >`_ :
697
697
698
698
* sources - `realtime-stream.ipynb <https://github.com/influxdata/influxdb-client-python/blob/master/notebooks/realtime-stream.ipynb >`_
699
699
700
+ .. code :: python
701
+
702
+ from datetime import timedelta
703
+ from typing import List
704
+
705
+ import hvplot.streamz
706
+ import pandas as pd
707
+ import rx
708
+ from rx import operators as ops
709
+
710
+ from streamz.dataframe import Random, DataFrame
711
+ from streamz import Stream
712
+ from influxdb_client import InfluxDBClient
713
+
714
+ .. code :: python
715
+
716
+ def source_data (auto_refresh : int , query : str , sink : Stream):
717
+ rx \
718
+ .interval(period = timedelta(seconds = auto_refresh)) \
719
+ .pipe(ops.map(lambda start : f ' from(bucket: "my-bucket") '
720
+ f ' |> range(start: - { auto_refresh} s, stop: now()) '
721
+ f ' { query} ' )) \
722
+ .pipe(ops.map(lambda query : client.query_api().query_data_frame(query, data_frame_index = [' _time' ]))) \
723
+ .pipe(ops.map(lambda data_frame : data_frame.drop(columns = [' result' , ' table' ]))) \
724
+ .subscribe(observer = lambda data_frame : sink.emit(data_frame), on_error = lambda error : print (error))
725
+ pass
726
+
727
+ .. code :: python
728
+
729
+ client = InfluxDBClient(url = ' http://localhost:9999' , token = ' my-token' , org = ' my-org' )
730
+
731
+ .. code :: python
732
+
733
+ cpu_query = ' |> filter(fn: (r) => r._measurement == "cpu") ' \
734
+ ' |> filter(fn: (r) => r._field == "usage_user") ' \
735
+ ' |> filter(fn: (r) => r.cpu == "cpu-total") ' \
736
+ ' |> keep(columns: ["_time", "_value"])'
737
+
738
+
739
+ cpu_sink = Stream()
740
+ cpu_example = pd.DataFrame({' _value' : []}, columns = [' _value' ])
741
+ cpu_df = DataFrame(cpu_sink, example = cpu_example)
742
+
743
+ source_data(auto_refresh = 5 , sink = cpu_sink, query = cpu_query)
744
+
745
+ .. code :: python
746
+
747
+ mem_query = ' |> filter(fn: (r) => r._measurement == "mem") ' \
748
+ ' |> filter(fn: (r) => r._field == "available" or r._field == "free" or r._field == "total" or r._field == "used") ' \
749
+ ' |> map(fn: (r) => ({ r with _value: r._value / 1024 / 1024 }))' \
750
+ ' |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")' \
751
+ ' |> keep(columns: ["_time", "used", "total", "free", "available"])'
752
+
753
+ mem_sink = Stream()
754
+ mem_example = pd.DataFrame({' used' : [], ' total' : [], ' free' : [], ' available' : []}, columns = [' available' , ' free' , ' total' , ' used' ])
755
+ mem_df = DataFrame(mem_sink, example = mem_example)
756
+
757
+ source_data(auto_refresh = 5 , sink = mem_sink, query = mem_query)
758
+
759
+ .. code :: python
760
+
761
+ from bokeh.models.formatters import DatetimeTickFormatter
762
+
763
+ # Time formatter
764
+ formatter = DatetimeTickFormatter(
765
+ microseconds = [" %H:%M:%S" ],
766
+ milliseconds = [" %H:%M:%S" ],
767
+ seconds = [" %H:%M:%S" ],
768
+ minsec = [" %H:%M:%S" ],
769
+ minutes = [" %H:%M:%S" ],
770
+ hourmin = [" %H:%M:%S" ],
771
+ hours = [" %H:%M:%S" ],
772
+ days = [" %H:%M:%S" ],
773
+ months = [" %H:%M:%S" ],
774
+ years = [" %H:%M:%S" ],
775
+ )
776
+
777
+ cpu_df.hvplot(width = 450 , backlog = 50 , title = ' CPU % u sage' , xlabel = ' Time' , ylabel = ' %' , xformatter = formatter) + \
778
+ mem_df.hvplot.line(width = 450 , backlog = 50 , title = ' Memory' , xlabel = ' Time' , ylabel = ' MiB' , xformatter = formatter, legend = ' top_left' )
779
+
780
+ .. image :: docs/images/realtime-result.gif
700
781
701
782
702
783
Advanced Usage
0 commit comments