pyzmq REQ/REP with asyncio await for variable

Question

I'm playing for the first time with asyncio in python and trying to combine it with ZMQ.

Basically my issue is that I have a REP/REQ system, in an async def with a function I need to await. how the value is not updated. Here's a snippet of the code to illustrate that:

#Declaring the zmq context
context = zmq_asyncio.Context()
REP_server_django = context.socket(zmq.REP)
REP_server_django.bind("tcp://*:5558")

I send this object to a class and get it back in this function

async def readsonar(self, trigger_pin, REP_server_django):
        i= 0
        while True:

            ping_from_view = await REP_server_django.recv()  # line.1
            value = await self.board.sonar_read(trigger_pin) # line.2
            print(value)                                     # line.3
            json_data = json.dumps(value)                    # line.4
            #json_data = json.dumps(i)                       # line.4bis
            REP_server_django.send(json_data.encode())       # line.5
            i+=1                                             # line.6
            await asyncio.sleep(1/1000)                      # line.7

the sonar_read, is using pymata_express to read an ultrasonic sensor. If I comment line.2 and line.4 I get the right value for i. If I comment line.1 and line.5 the print(value) prints the correct value from sonar_read. However, when I run it as shown here, the value is not updated.

Am I missing something?

EDIT :
Edited a type regarding the line comments. What I meant is that if I only read the sonar and print the value. It works fine. If I only .recv() and .send(json.dumps(i).encode()), it works. But if I try to send the value from the sonar. It locks to a given value which is not updated

EDIT2 : (answer to Alan Yorinks): here is the MWE, it considers what you sent regarding the declaration of zmq in the class. It is taken from the pymata_express example concurrent_tasks.py

To reproduce the error, run these two scripts in two different terminals. You will need an arduino board with Frimata_express installed. If all runs well, PART A. should only spit out the same value on the mve_req.py end. You may edit the diffrent blocks (PARTS A, B or C) to see the behaviour.

mve_rep.py

#ADAPTED FROM PYMATA EXPRESS EXAMPLE CONCURRENTTAKS
#https://github.com/MrYsLab/pymata-express/blob/master/examples/concurrent_tasks.py
import asyncio
import zmq
import json
import zmq.asyncio as zmq_asyncio
from pymata_express.pymata_express import PymataExpress


class ConcurrentTasks:

    def __init__(self, board):


        self.loop = board.get_event_loop()
        self.board = board

        self.ctxsync = zmq.Context()
        self.context = zmq.asyncio.Context()
        self.rep = self.context.socket(zmq.REP)
        self.rep.bind("tcp://*:5558")

        self.trigger_pin = 53
        self.echo_pin = 51

        loop.run_until_complete(self.async_init_and_run())

    async def readsonar(self):
        i = 0
        while True:


            #PART. A. WHAT I HOPE COULD WORK
            rep_recv = await self.rep.recv()                       # line.1
            value = await self.board.sonar_read(self.trigger_pin)  # line.2
            print(value)                                           # line.3
            json_data = json.dumps(value)                          # line.4
            # json_data = json.dumps(i)                            # line.4bis
            await self.rep.send(json_data.encode())                # line.5
            i += 1                                                 # line.6
            await asyncio.sleep(1 / 1000)                          # line.7


            '''
            #PART. B. WORKS FINE IN UPDATING THE SONAR_RAED VALUE AND PRINTING IT
            value = await self.board.sonar_read(self.trigger_pin)  # line.2
            print(value)                                           # line.3
            json_data = json.dumps(value)                          # line.4
            i += 1                                                 # line.6
            await asyncio.sleep(1 / 1000)                          # line.7
            '''

            '''
            #PART. C. WORKS FINE IN SENDING THE i VALUE OVER ZMQ
            rep_recv = await self.rep.recv()                       # line.1
            json_data = json.dumps(i)                              # line.4bis
            await self.rep.send(json_data.encode())                # line.5
            i += 1                                                 # line.6
            await asyncio.sleep(1 / 1000)                          # line.7
            '''



    async def async_init_and_run(self):

        await self.board.set_pin_mode_sonar(self.trigger_pin, self.echo_pin)

        readsonar = asyncio.create_task(self.readsonar())
        await readsonar

        # OTHER CREATED_TASK GO HERE, (removed them in the MVE, but they work fine)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    my_board = PymataExpress()
    try:
        ConcurrentTasks(my_board)
    except (KeyboardInterrupt, RuntimeError):
        loop.run_until_complete(my_board.shutdown())
        print('goodbye')
    finally:
        loop.close()

mve_req.py

import zmq
import time
import json

def start_zmq():
    context = zmq.Context()
    REQ_django  = context.socket(zmq.REQ)
    REQ_django.connect("tcp://localhost:5558")

    return REQ_django, context

def get_sonar(REQ_django):
    REQ_django.send(b"server_django")
    ping_from_server_django = REQ_django.recv()
    return ping_from_server_django.decode()

if __name__ == '__main__':

    data = {"sensors":{}}

    REQ_django, context = start_zmq()
    while REQ_django:

            data['sensors']['sonar'] = get_sonar(REQ_django)
            json_data = json.dumps(data)
            print(data)

            #DO OTHER WORK
            time.sleep(1)

    REQ_django.close()
    context.term()

Alan Yorinks · Accepted Answer · 2019-07-28T18:00:46.693

In full disclosure, I am the author of pymata-express and python-banyan. The OP requested that I post this solution, so this is not meant to be a shameless plug.

I have been developing with asyncio since it was first introduced in Python 3. When asyncio code works, asyncio (IMHO) can simplify concurrency and the code. However, when things go awry, it can be frustrating to debug and understand the cause of the issues.

I apologize ahead of time, since this may be a little lengthy, but I need to provide some background information so that the example will not seem like some random bit of code.

The python-banyan framework was developed to provide an alternative to threading, multi-processing, and asyncio. Simply put, a Banyan application consists of small targeted executables that communicate with one another using protocol messages that are shared over a LAN. At its core it uses Zeromq. It was not designed to have traffic move over the WAN, but to use a LAN as a "software backplane." In some ways, Banyan is similar to MQTT, but it is much faster when used within a LAN. It does have the capability to connect to an MQTT network if that is desireable.

Part of Banyan is a concept called OneGPIO. It is a protocol messaging specification that abstracts GPIO functionality to be independent of any hardware implementation. To implement the hardware specifics, specialized Banyan components, called Banyan Hardware Gateways were developed. There are gateways available for the Raspberry Pi, Arduino, ESP-8266 and Adafruit Crickit Hat. A GPIO application publishes the generic OneGPIO messages that any or all of the gateways can elect to receive. To move from one hardware platform to another, the hardware associated gateway is launched, and without modification, the control component (which is the code shown below) is launched. To go from one hardware platform to another, there are no code modifications necessary for any of the components, neither the control component nor the gateway is modified. Variables, such as pin numbers may be specificied through command line options when launching the control component. For the Arduino Gateway, pymata-express is used to control the GPIO of the Arduino. Pymata-express is an asyncio implementation of a StandardFirmata client. The thing to note that the code below is not asyncio. The Banyan framework allows one to develop using the tools that fit the problem, yet allow decoupling of parts of the solution, and in this case, the application allows the mixing the of asyncio with non-asyncio without any of the headaches normally encountered in doing so.

In the code provided, all the code below the class definition is used to provide support for command-line configuration options.

import argparse
import signal
import sys
import threading
import time

from python_banyan.banyan_base import BanyanBase


class HCSR04(BanyanBase, threading.Thread):
    def __init__(self, **kwargs):
        """
        kwargs contains the following parameters
        :param back_plane_ip_address: If none, the local IP address is used
        :param process_name: HCSR04
        :param publisher_port: publishing port
        :param subscriber_port: subscriber port
        :param loop_time: receive loop idle time
        :param trigger_pin: GPIO trigger pin number
        :param echo_pin: GPIO echo pin number
        """

        self.back_plane_ip_address = kwargs['back_plane_ip_address'],
        self.process_name = kwargs['process_name']
        self.publisher_port = kwargs['publisher_port']
        self.subscriber_port = kwargs['subscriber_port'],
        self.loop_time = kwargs['loop_time']
        self.trigger_pin = kwargs['trigger_pin']
        self.echo_pin = kwargs['echo_pin']
        self.poll_interval = kwargs['poll_interval']

        self.last_distance_value = 0

        # initialize the base class
        super(HCSR04, self).__init__(back_plane_ip_address=kwargs['back_plane_ip_address'],
                                     subscriber_port=kwargs['subscriber_port'],
                                     publisher_port=kwargs['publisher_port'],
                                     process_name=kwargs['process_name'],
                                     loop_time=kwargs['loop_time'])

        threading.Thread.__init__(self)
        self.daemon = True

        self.lock = threading.Lock()

        # subscribe to receive messages from arduino gateway
        self.set_subscriber_topic('from_arduino_gateway')

        # enable hc-sr04 in arduino gateway
        payload = {'command': 'set_mode_sonar', 'trigger_pin': self.trigger_pin,
                   'echo_pin': self.echo_pin}
        self.publish_payload(payload, 'to_arduino_gateway')

        # start the thread
        self.start()

        try:
            self.receive_loop()
        except KeyboardInterrupt:
            self.clean_up()
            sys.exit(0)

    def incoming_message_processing(self, topic, payload):
        print(topic, payload)
        with self.lock:
            self.last_distance_value = payload['value']

    def run(self):
        while True:
            with self.lock:
                distance = self.last_distance_value
            payload = {'distance': distance}
            topic = 'distance_poll'
            self.publish_payload(payload, topic)
            time.sleep(self.poll_interval)


def hcsr04():
    parser = argparse.ArgumentParser()
    # allow user to bypass the IP address auto-discovery.
    # This is necessary if the component resides on a computer
    # other than the computing running the backplane.
    parser.add_argument("-b", dest="back_plane_ip_address", default="None",
                        help="None or IP address used by Back Plane")
    parser.add_argument("-i", dest="poll_interval", default=1.0,
                        help="Distance polling interval")
    parser.add_argument("-n", dest="process_name", default="HC-SRO4 Demo",
                        help="Set process name in banner")
    parser.add_argument("-p", dest="publisher_port", default="43124",
                        help="Publisher IP port")
    parser.add_argument("-s", dest="subscriber_port", default="43125",
                        help="Subscriber IP port")
    parser.add_argument("-t", dest="loop_time", default=".1",
                        help="Event Loop Timer in seconds")
    parser.add_argument("-x", dest="trigger_pin", default="12",
                        help="Trigger GPIO pin number")
    parser.add_argument("-y", dest="echo_pin", default="13",
                        help="Echo GPIO pin number")

    args = parser.parse_args()

    if args.back_plane_ip_address == 'None':
        args.back_plane_ip_address = None
    kw_options = {'back_plane_ip_address': args.back_plane_ip_address,
                  'publisher_port': args.publisher_port,
                  'subscriber_port': args.subscriber_port,
                  'process_name': args.process_name,
                  'loop_time': float(args.loop_time),
                  'trigger_pin': int(args.trigger_pin),
                  'echo_pin': int(args.echo_pin),
                  'poll_interval': int(args.poll_interval)
                  }

    # replace with the name of your class
    HCSR04(**kw_options)


# signal handler function called when Control-C occurs
def signal_handler(sig, frame):
    print('Exiting Through Signal Handler')
    raise KeyboardInterrupt


# listen for SIGINT
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

if __name__ == '__main__':
    hcsr04()

Maybe I missed something, but why did you need to include `threading.Thread` in the class? — will.mendil, Jul 27 '19 at 16:51
I was playing around and left it in by mistake, however, I just modified the script above to use threading to allow you to poll for the latest value. You can set the polling time by using the -i option. A new Banyan message is published each time the polling time is achieved. — Alan Yorinks, Jul 28 '19 at 18:03

user3666197 · Answer 2 · 2019-07-24T20:31:27.200

_{( The O/P MCVE-problem definition creeps further - yet the problem of coordination, be it prioritised or not, of {sensors|actors}-control-systems, the more the systems designed with a use of distributed autonomous agents is professionally very complex and easy to make defective "shortcuts" or get into a system-wide blocking-state

Best first read at least this about ZeroMQ Hierarchy in Less Than Five Seconds and this about mutual dead-lock blocking

Reading the fabulous Pieter HINTJENS' book "Code Connected: Volume 1" is of immense value for any system designer )}

_{"...seams quite interesting as it implements the async already, so I could just add the async zmq as I did. am I wrong?"}

Yes, there is no "just add async" shortcut, control-systems are very interesting discipline, but rather a complex one. Always. Sorry to have to hear that straight. Some complexities might get hidden from user in schoolbook examples or in trivial makers-projects. The hammer then comes right on trying steps to extend them, by just adding one or a few more trivial features. The complexity suddenly gets to the surface, not seen before.

The formal map of the O/P multi-agent-`[A,B,C,D]`-system code^(as-is)

Put the formal map on a full-screen editor so as to see the bigger picture of all mutually conflicting dependencies and competing loops-of-control. Latency is the easy part. Several places of risks of un-resolvable deadlock blocking is the core one. ZeroMQ, since v2.x had tooling for avoiding some of these, software designer has the responsibility to properly mitigate all the others. Control systems ( robotics or other ) have to proof such robustness and resilience to errors and safely "survive" also all "external" mishaps.

The best point to start with is the old golden rule as expressed in assembly language directive on row 1:

;ASSUME NOTHING

and fight to carefully design all the rest.

multi-agent-[A,B,C,D]-system coordination
             | | | |
             +-|-|-|--------------------- python while   ~ 100 [ms] GIL-lock enforced quota for pure-[SERIAL]-ised code-execution, imposed on all python-threads ( be it voluntarily or involuntarily interruped by the python GIL-lock mechanics, O/S-specific )
               +-|-|--------------------- hardware ~  64 - 147 [ms] self.board proxy-driven, responding to python code
                 +-|--------------------- python asynchronous, strict sequence of remote/local events dependent ZeroMQ dFSA, distributed among local-code operated REP and remote-code operated REQ-side(s) - enforcing a mutually ordered sequence of distributed behaviour as REQ/REP Scalable Formal Communication Archetype Pattern defines
                   +--------------------- python asyncio.get_event_loop() instantiated another event-loop that may permit to defer an execution(s) of some parts of otherwise imperative python-code to some later time

multi-agent-[A,B,C,D]-system code (as-is)
             : : : :
             : : : +---------------------------------------------------------+
             : : +-----------------------------------------------------------:-------------------+ - - - - - - - - - - - - - - - - -<?network?>- - - - - - - - - - - - - - +
             : +-------------------------------------------------------------:----------+        :                                                                         :
             :                                                               :          :        :                                                                         :
             :                                                               :          :        :                                                                         :
             !                                                               :          :        :                                                                         :
____PYTHON___!                                                               :          :        :                                                                         :
             !                                                               ?          ?        ?                                                                         ?
          +->!                                                              D?         B?       C?REP-1:{0:N}-remote---------------<?network?>------------------------REQ.C? dFSA-state?dependent
          ^  !                                                              D?         B?       C?REP-1:{0:N}                                                            .C?
          ^ A!: IMPERATIVE LOOP-HEAD: while True:                           D?AWAIT    B?       C?REP-1:{0:N}-distributed-Finite-State-Automaton (dFSA) BEHAVIOUR, local .C? side depends also on EVOLUTION OF A FUZZY, DYNAMIC, MULTIPARTY, network-wide dFSA-STATE(s) inside such ECOSYSTEM
          ^  !                                                              D?         B?       C?                                                                        
          ^  !                                                              D?         B?       C?                    REQ.C?-distributed-Finite-State-Automaton-STATE-REP.C?
          ^  !                                                              D?         B?       C?                       vC?                                             ^C?
          ^  !_______.SET DEFERRED:         P_D?C?_deferred_yield_ping     =D?await ...         C?REP.recv()---<--?---?--vC?-----<--<network>--------<--?remote-REQ-state-C?-( ^C?-dFSA-state && C?.recv()-blocking-mode of REQ/REP .recv()-waiting till a message, if any arrives error-free, blocks till then, just deferred via D?await )
          ^  !                                                              D?         B?                                vC?                                             ^C?
          ^  !_______.SET DEFERRED:         S_D?B?_deferred_yield_sonar    =D?await ...B?.board.sonar_read()-o-<--?-+    vC?                                             ^C?
          ^  !                                                                                               :      |    vC?                                             ^C?
          ^  !_______.GUI OUTPUT:           print( deferred_yield_sonar )  #A!->-----------------------------+->----?->---:?--->[ a last-known (if any) S_D?B?_deferred_yield_sonar value put "now" on GUI-screen ]
          ^  !                                                                                               :      ^    vC?                                             ^C?
          ^  !_______.SET TRANSFORMED:      S_D?B?_dependent_tranformed    =A!json.dumps( S_D?B? )--<--<--<--+      |    vC? <--[ a last-known (if any) S_D?B?_deferred_yield_sonar value transformed and assigned]
          ^  !                                                                                               :      |    vC?                                             ^C?
          ^  !_______.BLOCKING-MODE-SEND()  REP.send( S_D?B?_dependent_transformed.encode() )  #C? .send( S_D?B? )--?---->C?-->----<?network?>-->-------?remote-REQ-state-C?-( +C?-indeterministic and blocking-mode of REQ/REP .recv()-waiting till a message, if any arrives error-free, blocks till then )
          ^  !X:C?                                                                                                  ^    vC?                                             ^C?
          ^  !X:C?___.SET IMPERATIVE:       i += 1                                                                  | REQ.C?-distributed-Finite-State-Automaton-STATE-REP.C?
          ^  !X:C?                                                                                                  ?                                                       
          ^  !X:C?___.NOP/SLEEP() DEFERRED: await sleep( ... )             #D?AWAIT                                 ^                                                      :
          ^  !X:C?D?+0ms                                                                                            |                                                      :
          ^  !X:C?D?_.JUMP/LOOP                                                                                     ?                                                      :
          ^__!X:C?D?+0ms                                                                                            ^                                                      :
                                                                                                                    |                                                      :
                                                                                                                    |                                                      :
                                                                                                                    |                                                      :
____SONAR___________________________________________________________________________________________________________B? REQUEST T0: + EXPECT ~64 - ~147 [ms] LATENCY        :
                                                                                                                    B? hardware value acquisition latency can be masked    :
                                                                                                                       via await or other concurrency-trick )              :
                                                                                                                                                                           :
____REQ-side(s)_?{0:N} __________________________________________________________________________________________________________________________________________________^C? dFSA-remote autonomous-agent outside-of-CodeBlock-scope-of-control + behind <?network?>
_____REQ-side(s)_?{0:N} _________________________________________________________________________________________________________________________________________________^C? dFSA-remote autonomous-agent outside-of-CodeBlock-scope-of-control + behind <?network?>
______REQ-side(s)_?{0:N} ________________________________________________________________________________________________________________________________________________^C? dFSA-remote autonomous-agent outside-of-CodeBlock-scope-of-control + behind <?network?>
_______REQ-side(s)_?{0:N} _______________________________________________________________________________________________________________________________________________^C? dFSA-remote autonomous-agent outside-of-CodeBlock-scope-of-control + behind <?network?>
     ...                                                                                                                                                                 ::: ...
______...REQ-side(s)_?{0:N} _____________________________________________________________________________________________________________________________________________^C? dFSA-remote autonomous-agent outside-of-CodeBlock-scope-of-control + behind <?network?>

As the O/P's EDIT : has explained 2 hours ago,

the problem is now evident. The infinite while True:-loop instructs to hard-step through, line by line, and loop-"rotate" all the steps again, one after another, while any asyncio await-decorated functor(s) present there are being left, asynchronously independent of this "main" A: while True:-loop block of imperative code execution. The same way a B: self.board-device's external sonar-device is an independently timed device, external to the python code, having some unmanageable hardware/read/decode-latencies, the coordination of fixed-looping + C: ZeroMQ-REQ/REP-Archetype-behaviour ( again externally-coordinated with a decentralised "foreign" REQ-actor(s)/agent(s) - yes, you cannot know, how many of these are there ... - but all being outside of your scope of control and both all the REQ-side(s) and your locally-instantiated REP-side distributed-Finite-State-Machine states are being fully independent of the "framing"-python loop's will to drive steps forward and execute the next step, the next step, the next step ... ) + another, here D: asyncio.get_event_loop()-instantiated "third"-event_loop, that influences how await-decorated functors are actually permitted to defer to yield their results and deliver 'em a some later time ----- and, this is the problem of the "cross-bread"-event_loops.

_{If this problem setup has been elaborated from any Computer Science professor, she/he deserves standing ovations for making the task the best example of problems with distributed systems - almost may serve as a tribute to Mrs. Margaret HAMILTON's work on proper design of the Apollo AGC computer system, where her work has solved this class-of-problems and thus has saved the lives of crew and all the pride of the Moon landing, right 50-years ago. Great lecture, Mrs. Hamilton, great lecture.}

Trivial, yet right on the spot.

Indeed a lovely and scientifically marvelous task:

Design a strategy for a robust, failure-resilient and coordinated work of a set of independently timed and operated agents [A, B, C, D], A being an imperative interpreted python language, principally having the GIL-lock prevented zero-concurrency, but a pure [SERIAL] process-flow, C being a fuzzy set of semi-persistent network-distributed REQ/REP-agents, B being an independently operated hardware device with some limited I/O interfacing to an A-inspectable self.board-proxy and all being mutually independent and physically distributed across a given ecosystem of software, hardware and network.

Hardware diagnostics + a proposed System Architecture approach have already been proposed yesterday. Without testing the self.board-hosted sonar-device latencies no one can decide a next best step, as the realistic ( in-vivo benchmarked ) hardware response-times ( + best also the documentation down to the .board and it's peripheral sensor device(s) MUX-ed or not? PRIO-driven or MUTEX-locked or static, non-shared peripheral device, register-read-only abstracted, ... ? ) are cardinal for deciding about the possible [A, B, C, D]-coordination strategy.

The ZeroMQ part :

If you comment l.5 - REP_server_django.send(json_data.encode()) # l.5 you get into final block, as the original, strict form of the REQ/REP ZeroMQ Scalable Formal Communication Archetype Pattern cannot .recv() again, if it did not reply before that to the REQ-side after the first .recv() has been received with a .send().

This was a simple catch.

The rest is not a reproducible code.

You may want to:

verify, if self.board.sonar_read( trigger_pin ) receives any value and test a latency of doing that:

   import numpy as np
   from zmq import Stopwatch
   aClk = Stopwatch()

   def sonarBeep():
       try:
            a_value   = -1
            aClk.start()
            a_value   = self.board.sonar_read( trigger_pin )
            a_time_us = aClk.stop()
       except:
            try:
                aClk.stop()
            finally:
                a_time_us = -1
       finally:
           return( a_value, a_time_us )

and run a series of 100 sonar-tests, to get min, Avg, StDev, MAX readings about latency times all in [us] as these values are cardinal to know, in case some control-loops are to be designed w.r.t the SONAR-sensor data.

[ aFun( [ sonarBeep()[1] for _    in range( 100 ) ]
        )                for aFun in ( np.min, np.mean, np.std, np.max )
  ]

The System Architecture and sub-systems coordination :

Last, but not least, one may let read and store sonar data, in an absolutely independent event loop, uncoordinated with any other operations and just read a state-variable from such a storage, being set in an independently working subsystem ( if not extremely saving power for doing that as independent system behaviour )

Whenever one tries to tightly coordinate a flow of independent events ( the worst in distributed systems with uncoordinated or weakly coordinated agents ) design has to grow in both robustness to errors and time mis-alignments and error-resilience. Otherwise the system may soon deadlock/livelock itself in a snap.

If in doubts, may learn from the original philosophy of the XEROX Palo Alto Research Centre MVC-separation, where the MODEL-part can ( and in the GUI-frameworks most of the time, since 198x+ does ) receive many state-variables updated all independently of other system components, that just read/use actual state-variables' data if they need them and as they need them. Similarly, SONAR can, if power budget permits, continuously scan the scene and write readings into any local-registers and let other components come and ask or get served their request for the last actual SONAR reading.

So does the ZeroMQ zen-of-zero work.

If that may help, check the zmq.CONFLATE mode of the local-side message-store working right this way.

_{A minor note: one might already have noted, that sleep( 1 / 1000 ) is quite an expensive, repetitively executed step and dangerous as it effectively does no sleep in py2.x, due to integer division.}

That's a very interesting answer, maybe a bit out of my reach. nevertheless, I've ran your latency test. and here are the value : min = 64 ; max = 147; sdtdev = 7.48; avg = 74.5. i ran it over 1000. My thoughts into this method, was to have zmq wait to receive a msg from another script to look for the sonar value, and whilst it is waiting, it could be running concurrent tasks such as enabling motors, etc... this is were pymata express seams quite interesting as it implements the async already, so I could just add the async zmq as I did. am I wrong? — will.mendil, Jul 24 '19 at 12:39
I very much appreciate the details in your answers, but I have to admit that I don't fully understand everything. Are you telling me that what I am trying to do is not possible? Is there an alternative way I could solve this issue? Basically piping the values read by sonar to another script. — will.mendil, Jul 25 '19 at 01:23

score 0 · Answer 3 · answered Jul 23 '19 at 23:36

I am not sure if this will solve your issue but I did find some potential problems.

It is not clear how readsonar is being called.
The creation of the context has a typo.
REP_server_django.send is not awaited.

Below is my rework of the code (untested):

import asyncio
import zmq
import json


class Play:
    def __init__(self):
        self.context = zmq.asyncio.Context()
        self.REP_server_django = self.context.socket(zmq.REP)
        self.REP_server_django.bind("tcp://*:5558")
        self.event_loop = asyncio.get_event_loop()
        self.event_loop.run_until_complete(self.readsonar(4))

    async def readsonar(self, trigger_pin):
        i = 0
        while True:
            ping_from_view = await self.REP_server_django.recv()  # l.1
            value = await self.board.sonar_read(trigger_pin)  # l.2
            print(value)  # l.3
            json_data = json.dumps(value)  # l.4
            # json_data = json.dumps(i) # l.4bis
            await self.REP_server_django.send(json_data.encode())  # l.5
            i += 1  # l.6
            await asyncio.sleep(1 / 1000)  # l.6

Having worked with ZeroMQ for ages, there is a whole class of problems if one tries to **safely** "cross-breed" the { synchronous (blocking) | asynchronous }-modes of ZeroMQ Context()-instance's (an async engine, w signaling/messaging/queue-storage-mgrs/multiprotocol-factories/interface-driving logic, all operated inside a completely separate pool of threads) services with any other asyncio (now py3-included) or other event-loops. Even making a try to "re-use" Tkinter-mainloop, which has always been a trick how to reuse a smart framework of uncoordinated actors/functors is hard to mix with zmq — user3666197, Jul 24 '19 at 03:40
Updated the original question to address this point with a mwe. If ZMQ is not right for the job here. How would you solve this issue. Basically I have the "server" script asking for the sonar value for this script using the REQ/REP. I wouldn't know how else to pipe this data — will.mendil, Jul 24 '19 at 12:41

score 0 · Answer 4 · answered Jul 25 '19 at 21:40

I made it work, though I have to admit, I do not understand the reason for why it works. Basically I had to make a new async def which only polls the reading from sonar_read and use asyncio.wait to return the value. here's the code:

#ADAPTED FROM PYMATA EXPRESS EXAMPLE CONCURRENTTAKS
#https://github.com/MrYsLab/pymata-express/blob/master/examples/concurrent_tasks.py
import asyncio
import zmq
import json
import zmq.asyncio as zmq_asyncio
from pymata_express.pymata_express import PymataExpress


class ConcurrentTasks:

    def __init__(self, board):


        self.loop = board.get_event_loop()
        self.board = board

        self.ctxsync = zmq.Context()
        self.context = zmq.asyncio.Context()
        self.rep = self.context.socket(zmq.REP)
        self.rep.bind("tcp://*:5558")

        self.trigger_pin = 53
        self.echo_pin = 51

        loop.run_until_complete(self.async_init_and_run())

    ### START:  NEW CODE THAT RESOLVED THE ISSUE
    async def pingsonar(self):
        value = await self.board.sonar_read(self.trigger_pin)
        return value

    async def readsonar(self):
        while True:
            rep_recv = await self.rep.recv() 
            value = await asyncio.wait([self.pingsonar()])
            valuesonar = list(value[0])[0].result()
            json_data = json.dumps(valuesonar) 
            await self.rep.send(json_data.encode()) 
            await asyncio.sleep(1 / 1000) #maybe this line isn't necessary

    ### END : NEW CODE THAT RESOLVED THE ISSUE

    async def async_init_and_run(self):

        await self.board.set_pin_mode_sonar(self.trigger_pin, self.echo_pin)

        readsonar = asyncio.create_task(self.readsonar())
        await readsonar

        # OTHER CREATED_TASK GO HERE, (removed them in the MVE, but they work fine)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    my_board = PymataExpress()
    try:
        ConcurrentTasks(my_board)
    except (KeyboardInterrupt, RuntimeError):
        loop.run_until_complete(my_board.shutdown())
        print('goodbye')
    finally:
        loop.close()

I appreciate the help nonetheless.

pyzmq REQ/REP with asyncio await for variable

4 Answers4

The formal map of the O/P multi-agent-`[A,B,C,D]`-system code^(as-is)

The ZeroMQ part :

The System Architecture and sub-systems coordination :

Linked

pyzmq REQ/REP with asyncio await for variable

4 Answers4

The formal map of the O/P multi-agent-[A,B,C,D]-system code (as-is)

The ZeroMQ part :

The System Architecture and sub-systems coordination :

Linked

The formal map of the O/P multi-agent-`[A,B,C,D]`-system code^(as-is)