Live Data Reduction

From Terminal

You can run the beamlime command from your terminal to run the application and see the real-time update of a plot.

beamlime --image-path beamlime_plot.png

This command will save the plot into beamlime_plot.png.

Jupyter Environment

You can also run the same thing in jupyter environment and use plopp matplotlib widget to see the real-time updated plot.

It uses the same ``PlotHandler`` as above, so running this application will create a file called ``beamlime_plot.png`` in the current directory.

%matplotlib widget
import scipp as sc

sc.get_logger().addHandler(sc.logging.make_widget_handler())
sc.display_logs()

from matplotlib import pyplot as plt

from beamlime.constructors import multiple_constant_providers, SingletonProvider, Factory
from beamlime.logging import BeamlimeLogger

from beamlime.executables.prototypes import collect_default_providers
from beamlime.applications.base import Application
from beamlime.applications.handlers import (
    DataReductionHandler,
    WorkflowResultUpdate,
    PlotStreamer,
    DataAssembler,
    DataReady,
)
from beamlime.applications.daemons import (
    DetectorDataReceived, FakeListener, NexusTemplatePath,
    NumFrames, DataFeedingSpeed,
    EventRate, FrameRate, RunStart
)
from beamlime.stateless_workflow import Workflow


prototype_factory = Factory(collect_default_providers())
prototype_factory.providers[FakeListener] = SingletonProvider(FakeListener)
prototype_factory.providers[PlotStreamer] = SingletonProvider(PlotStreamer)
prototype_factory.providers[DataAssembler] = DataAssembler

with multiple_constant_providers(
    prototype_factory,
    {
        BeamlimeLogger: sc.get_logger(),
        DataFeedingSpeed: DataFeedingSpeed(0.3),
        NumFrames: NumFrames(5),
        EventRate: EventRate(10_000),
        FrameRate: FrameRate(14),
        Workflow: Workflow('dummy'),
        NexusTemplatePath: NexusTemplatePath('../../tests/applications/ymir_detectors.json')
    },
):
    prototype_factory[BeamlimeLogger].setLevel("INFO")

    # Build the application
    app = prototype_factory[Application]
    # Register Handlers
    plot_saver = prototype_factory[PlotStreamer]
    app.register_handling_method(WorkflowResultUpdate, plot_saver.update_histogram)
    data_assembler = prototype_factory[DataAssembler]
    app.register_handling_method(RunStart, data_assembler.set_run_start)
    app.register_handling_method(DetectorDataReceived, data_assembler.assemble_detector_data)
    data_reduction_handler = prototype_factory[DataReductionHandler]
    app.register_handling_method(DataReady, data_reduction_handler.reduce_data)
    # Register Daemons
    app.register_daemon(prototype_factory[FakeListener])

    app.run()
    plt.close()  # Close the plot from this cell.

plot_saver.show()