SObjectizer Tales – 6. Is the stream still in progress?

Within the earlier episode, we coated the method of sending instructions and using timers. Nonetheless, we encountered a problem that impacts our picture viewer. Particularly, when the acquisition course of is halted, the picture window turns into unresponsive. Nevertheless, as soon as the acquisition resumes, it regains responsiveness and features easily.

Are you able to reproduce the difficulty? What’s occurring?

The conduct depends upon how we deal with OpenCV’s window. After displaying a picture with imshow it’s our duty to name waitKey to allow GUI-related duties like responding to mouse and keyboard interactions. This course of is carried out for every body by image_viewer. Nevertheless, when the picture stream pauses, the chance to name waitKey additionally vanishes. Consequently, the window, though nonetheless current, turns into unresponsive.

When the agent now not receives frames, it could be extra appropriate to both shut the window or proceed invoking waitKey. On this put up, we’ll discover a extra superior utilization of agent states to implement each of those methods.

Observing when the stream goes idle

The central problem revolves round detecting when the producer involves a halt. One potential resolution may contain having the image_viewer subscribe to the cease command. Nevertheless, this method has its drawbacks as a result of the producer could take a while to course of the cease command and ship the ultimate frames. Consequently, the image_viewer may doubtlessly miss the final set of photographs.

Moreover, in a extra basic state of affairs, the cease command won’t even be accessible, or the producer may stop its operation for varied different causes, together with momentary points, {hardware} errors, and the like.

Another, middle-ground resolution entails enhancing the communicativeness of producers. They may transmit notifications concerning the standing of the stream and the digicam, indicating when the acquisition commences, when it concludes, or if any issues come up. Nevertheless, this method is efficient solely when the image_viewer straight subscribes to the producer’s output. On the flip facet, if the image_viewer is employed to visualise a unique channel, such because the output of one other agent, implementing this method would necessitate further logic, as notifications won’t be a part of the output of such an agent.

Therefore, it’s crucial to make sure that the image_viewer can operate seamlessly with any picture channel.

One common method to find out the stream’s present standing is by implementing a time-based criterion: if a particular length elapses with none photographs being transmitted by way of the channel, we will infer that the stream has skilled both a definitive interruption or pause.

We may derive the brink from the digicam’s body charge, however there are conditions the place the machine is externally triggered, body by body, by a bodily entity like an encoder. This methodology not directly impacts the output charge dynamically. In such circumstances, extra superior options could contain repeatedly estimating and adjusting this threshold over time. Nevertheless, it’s essential to notice that the image_viewer is primarily supposed as a troubleshooting instrument and gained’t be ceaselessly utilized in manufacturing. Due to this fact, utilizing a static worth could suffice in the intervening time.

Strategy #1: simply closing the window

The preliminary method entails closing the window when the stream ends.

At current, the image_viewer operates solely in its default state. Upon receiving the primary body, it creates a window that’s subsequently closed when this system concludes. Subsequent frames are then displayed inside this window. As beforehand talked about, one method to establish when the stream turns into inactive entails monitoring the absence of incoming frames for a specified interval.

SObjectizer’s agent states are fairly refined and supply some utilities that is perhaps helpful for growing a working resolution. To begin with, image_viewer may be modeled as a two-state agent:

• dealing with photographs: as earlier than, simply show photographs as they arrive;
• stream down: do nothing.

The principle level is learn how to specific transitions between the 2:

• when in dealing with photographs, if no frames arrive for a sure time restrict, the agent transitions to stream down;
• when exiting from dealing with photographs, the agent closes the window;
• if a picture arrives when in stream down, the agent transitions to dealing with photographs.

Right here is the code:

class image_viewer : public so_5::agent_t
{
    so_5::state_t st_handling_images{ this };
    so_5::state_t st_stream_down{ this };
public:
    image_viewer(so_5::agent_context_t ctx, so_5::mbox_t channel)
        : agent_t(std::transfer(ctx)), m_channel(std::transfer(channel))
    {
    }

    void so_define_agent() override
    {
        st_handling_images
            .occasion(m_channel, [this](const cv::Mat& picture) {
                imshow(m_title, picture);
                cv::waitKey(25);
                st_handling_images.time_limit(500ms, st_stream_down);
            })
            .on_exit([this] { cv::destroyWindow(m_title); });

        st_stream_down
            .transfer_to_state<cv::Mat>(m_channel, st_handling_images);

        st_stream_down.activate();
    }
personal:

    so_5::mbox_t m_channel;
    std::string m_title = "picture viewer";
};

As you may see, states present features for expressing further properties and transitions.

Some particulars:

• on_exit takes a operate that’s executed when the agent leaves the state. It’s the chance to name destroyWindow;
• transfer_to_state permits the agent to modify from the present state to a different when a sure message arrives. In different phrases, whereas being in st_stream_down, if cv::Mat arrives, the agent transitions to st_handling_images and cv::Mat is dealt with there;
• time_limit permits to set a most period of time the agent could persist in a sure state.

It’s essential to know why time_limit is known as contained in the message handler. One can suppose that doing this may be equal:

st_handling_images
    .occasion(m_channel, [this](const cv::Mat& picture) {
        imshow(m_title, picture);
        cv::waitKey(25);
    })
    .time_limit(500ms, st_stream_down);
    .on_exit([this] { cv::destroyWindow(m_title); });

Nevertheless, on this case the timer is just not reset when a brand new picture arrives. The impact is seeing the window blinking whereas the stream is energetic, as a result of it will get destroyed and created each 500ms! Alternatively, when time_limit known as contained in the handler then its timer is reset after each picture (that’s the supposed conduct, as a result of we set a time restrict after the final acquired body).

As you may think about, there may be additionally on_enter that executes a operate when transitioning right into a state. Simply to your info, each on_enter and on_exit mustn’t throw exceptions.

Word that this resolution is just not bulletproof: cv::waitKey(25) suffers the exact same downside as estimating how lengthy to await between two frames. Nevertheless, as stated, image_viewer is only for troubleshooting and may be sufficient for now. Be at liberty to alter this quantity at will.

Here’s a demo:

Strategy #2: protecting the window responsive

Typically it may make sense to show the final acquired body and maintain the window responsive.

Principally, when the stream goes down, we must always carry on calling waitKey till a brand new body arrives. This could remind you of the post the place we mentioned learn how to mannequin an infinite loop with messages. Certainly, the sample fills the invoice.

Additionally, since we’ve got simply given a style of agent state utilities, we will use our new instruments to specific transitions succinctly.

We are able to nonetheless mannequin the picture viewer as a two-state machine, including a sign to the occasion. When photographs arrive – that’s st_handling_images – the agent behaves precisely as earlier than. When it enters into st_stream_down, this time the agent sends call_waitkey to itself to start out calling waitKey till cv::Mat arrives, and the state turns into st_handling_images once more. Not neglect to take away on_exit from st_handling_images since we don’t have to destroy the window anymore:

class image_viewer_live : public so_5::agent_t
{
    struct call_waitkey closing : so_5::signal_t {};
    so_5::state_t st_handling_images{ this };
    so_5::state_t st_stream_down{ this };   
public:
    image_viewer_live(so_5::agent_context_t ctx, so_5::mbox_t channel)
        : agent_t(std::transfer(ctx)), m_channel(std::transfer(channel))
    {
    }

    void so_define_agent() override
    {
        st_handling_images
            .occasion(m_channel, [this](const cv::Mat& picture) {
                imshow(m_title, picture);
                cv::waitKey(25);
                st_handling_images.time_limit(500ms, st_stream_down);
            });

        st_stream_down
            .on_enter([this] {
                so_5::ship<call_waitkey>(*this);
            })
            .occasion([this](so_5::mhood_t<call_waitkey>) {
                cv::waitKey(25);
                so_5::ship<call_waitkey>(*this);
            }).transfer_to_state<cv::Mat>(m_channel, st_handling_images);

        st_handling_images.activate();
    }
personal:
    so_5::mbox_t m_channel;
    std::string m_title = "picture viewer reside";
};

Word that we activate st_handling_images first in any other case call_waitkey is distributed with none present window (OpenCV wouldn’t complain although).

Here’s a demo:

Stream heartbeat logger

A variation of this sample may be adopted for crafting one other utility agent that logs the uptime of the stream, let’s say each 5 seconds. When the stream is energetic, we count on such logs:

[Heartbeat] Uptime: 00:00:05
[Heartbeat] Uptime: 00:00:10
[Heartbeat] Uptime: 00:00:15
[Heartbeat] Uptime: 00:00:20
...

When the stream goes down, the agent shuts up. And if the stream goes up once more, so does the agent (restarting from 0).

This time, we mix state transitions with periodic messages. Likewise iage_viewer_live, the agent sends a sign to itself, however not at breakneck velocity, as a substitute each 5 seconds:

class stream_heartbeat : public so_5::agent_t
{
    struct log_heartbeat closing : so_5::signal_t {};
    so_5::state_t st_handling_images{ this };
    so_5::state_t st_stream_down{ this };
public:
    stream_heartbeat(so_5::agent_context_t ctx, so_5::mbox_t channel)
        : agent_t(std::transfer(ctx)), m_channel(std::transfer(channel))
    {
    }

    void so_define_agent() override
    {
        st_handling_images
            .on_enter([this] {
                m_start_time = std::chrono::steady_clock::now();
                m_timer = so_5::send_periodic<log_heartbeat>(so_direct_mbox(), 5s, 5s);
            })
            .occasion(m_channel, [this](const cv::Mat&) {
                st_handling_images.time_limit(500ms, st_stream_down);
            }).occasion([this](so_5::mhood_t<log_heartbeat>) {
                std::osyncstream(std::cout) << std::format("[Heartbeat] Uptime: {:%H:%M:%S}n", std::chrono::ground<std::chrono::seconds>(std::chrono::steady_clock::now() - m_start_time));
            }).on_exit([this] {
                m_timer.launch();
            });

        st_stream_down
            .transfer_to_state<cv::Mat>(m_channel, st_handling_images);

        st_stream_down.activate();
    }
personal:
    so_5::mbox_t m_channel;
    std::chrono::time_point<std::chrono::steady_clock> m_start_time;
    so_5::timer_id_t m_timer;
};

Just a few particulars:

• log_heartbeat signifies when it’s time to log the message on the console;
• m_timer holds the periodic message and it’s launched (aka: cancelled) when the stream goes down;
• m_timer is (re)created when the stream goes up, together with m_start_time which represents the streaming begin time;
• the uptime is calculated because the variety of seconds from the beginning time rounded up.

As normal, you may play with these new brokers by including them to the cooperation. For instance:

int essential()
{
    auto ctrl_c = get_ctrl_c_token();

    const so_5::wrapped_env_t sobjectizer;
    auto main_channel = sobjectizer.atmosphere().create_mbox("essential");
    auto commands_channel = sobjectizer.atmosphere().create_mbox("instructions");

    auto dispatcher = so_5::disp::active_obj::make_dispatcher(sobjectizer.atmosphere());
    sobjectizer.atmosphere().introduce_coop(dispatcher.binder(), [&](so_5::coop_t& c) {
        c.make_agent<image_producer_recursive>(main_channel, commands_channel);
        c.make_agent<image_viewer_live>(main_channel);
        c.make_agent<remote_control>(commands_channel);
        c.make_agent<stream_heartbeat>(main_channel);
    }

    wait_for_stop(ctrl_c);
}

Here’s a demo:

Only a design digression

Since stream_heartbeat, image_viewer and image_viewer_live do the exact same interior logic, we will consider transferring that right into a devoted agent that sends stream notifications:

class stream_detector : public so_5::agent_t
{
    so_5::state_t st_handling_images{ this };
    so_5::state_t st_stream_down{ this };

public:
    struct stream_up closing : so_5::signal_t{};
    struct stream_down closing : so_5::signal_t{};

    stream_detector(so_5::agent_context_t ctx, so_5::mbox_t channel, so_5::mbox_t output)
        : agent_t(std::transfer(ctx)), m_channel(std::transfer(channel)), m_out_channel(std::transfer(output))
    {
    }

    void so_define_agent() override
    {
        st_handling_images
            .on_enter([this] {
                so_5::ship<stream_up>(m_out_channel);
            })
            .occasion(m_channel, [this](const cv::Mat&) {
                st_handling_images.time_limit(500ms, st_stream_down);
            }).on_exit([this] {
                so_5::ship<stream_down>(m_out_channel);
            });

        st_stream_down
            .transfer_to_state<cv::Mat>(m_channel, st_handling_images);

        st_stream_down.activate();
    }
personal:
    so_5::mbox_t m_channel;
    so_5::mbox_t m_out_channel;
};

At this level, stream_heartbeat boils all the way down to:

class stream_heartbeat_with_detector: public so_5::agent_t
{
    struct log_heartbeat closing : so_5::signal_t {};
public:
    stream_heartbeat_with_detector(so_5::agent_context_t ctx, so_5::mbox_t detector_channel)
        : agent_t(std::transfer(ctx)), m_channel(std::transfer(detector_channel))
    {
    }

    void so_define_agent() override
    {
        so_subscribe(m_channel).occasion([this](so_5::mhood_t<stream_detector::stream_up>) {
            m_start_time = std::chrono::steady_clock::now();
            m_timer = so_5::send_periodic<log_heartbeat>(so_direct_mbox(), 5s, 5s);
        }).occasion([this](so_5::mhood_t<stream_detector::stream_down>) {          
            m_timer.launch();
        });

        so_subscribe_self().occasion([this](so_5::mhood_t<log_heartbeat>) {
            std::osyncstream(std::cout) << std::format("[Heartbeat] Uptime: {:%H:%M:%S}n", std::chrono::ground<std::chrono::seconds>(std::chrono::steady_clock::now() - m_start_time));
        });
    }
personal:
    so_5::mbox_t m_channel;
    std::chrono::time_point<std::chrono::steady_clock> m_start_time;
    so_5::timer_id_t m_timer;
};

Right here is an instance of utilization:

int essential()
{
    auto ctrl_c = get_ctrl_c_token();

    const so_5::wrapped_env_t sobjectizer;
    auto main_channel = sobjectizer.atmosphere().create_mbox("essential");
    auto commands_channel = sobjectizer.atmosphere().create_mbox("instructions");
    auto stream_notifications = sobjectizer.atmosphere().create_mbox("stream");

    auto dispatcher = so_5::disp::active_obj::make_dispatcher(sobjectizer.atmosphere());
    sobjectizer.atmosphere().introduce_coop(dispatcher.binder(), [&](so_5::coop_t& c) {
        c.make_agent<image_producer_recursive>(main_channel, commands_channel);
        c.make_agent<stream_detector>(main_channel, stream_notifications);
        c.make_agent<stream_heartbeat_with_detector>(stream_notifications);
        c.make_agent<image_viewer_live>(main_channel);
        c.make_agent<remote_control>(commands_channel);
    }

    wait_for_stop(ctrl_c);
}

Clearly, somewhat than including a brand new channel, main_channel can host output notifications as effectively:

c.make_agent(main_channel, main_channel); // output to main_channel
c.make_agent(main_channel);

Including an additional channel is perhaps not only a matter of fashion or style. We are likely to maintain some channels separate, particularly to keep away from mixing “uncooked” knowledge with “derived” knowledge. Within the subsequent put up we’ll get again to this a bit extra.

One other design argument regards stream_heartbeat_with_detector.

It’s evident that stream_heartbeat_with_detector is much less generic than stream_heartbeat as a result of it may’t merely work with any “picture channel”, however solely with these carrying stream_up and stream_down. This creates an implicit dependency on stream_detector, and we will additional solidify this relationship by putting each stream_detector and heartbeat_with_detector into the identical cooperation.

Alternatively, stream_heartbeat_with_detector has gained two attention-grabbing options: firstly, it may now operate seamlessly with any picture kind, eliminating its earlier dependency on cv::Mat. For instance, if we introduce introduce a brand new SDKImage and combine it with stream_detector, stream_heartbeat_with_detector will proceed to work with out modification. Secondly, it now not handles the “500 milliseconds” threshold, which was beforehand used to find out if a stream turns into idle. This duty is now solely throughout the area of stream_detector (about this threshold, nevertheless, the central level mentioned earlier on this article nonetheless stands: the simplest method to find out if the stream has began and concluded is to have the producer ship applicable indicators).

It’s essential to focus on that the primary function has a wider applicability. Within the state of affairs the place we’ve got quite a few brokers designed to work solely with cv::Mat (comparable to image_viewer and image_tracer), if, in some unspecified time in the future, we introduce assist for SDKImage into this system, we will simplify the transition by introducing an adapter agent that converts SDKImage into cv::Mat (assuming it’s possible). This adaptation would allow the present brokers to seamlessly proceed their operations with out requiring any further modifications:

The excellent news is that we do have a alternative. Wanna go for placing all of the logic into a number of brokers, minimizing message trade and sort proliferation? Deal. Or do you favor fine-grained brokers and extra modularity? Deal.

Takeaway

On this episode we’ve got realized:

• agent states present features to specific transitions and specific situations that should be assured;
• on_enter executes a operate when the agent transitions to a sure state;
• on_exit executes a operate when the agent leaves a sure state;
• on_enter and on_exit handlers mustn’t throw exceptions;
• time_limit units the most time the agent may be in a sure state (and may be reset from a message handler);
• combining agent states with periodic messages may be highly effective.

As normal, calico is up to date and tagged.

What’s subsequent?

Our supervisor is extremely impressed with the venture’s developments and is keen to place it to make use of. Nevertheless, with an essential buyer demonstration on the horizon, she has a brand new request:

“The shopper shall be captivated by the system’s flexibility. May you add extra brokers and supply examples showcasing the ability of composition?”

Within the forthcoming put up, we’ll introduce a number of extra brokers, highlighting the dynamic potential of composition in motion!

Due to Yauheni Akhotnikau for having reviewed this put up.