Collectives™ on Stack Overflow

Find centralized, trusted content and collaborate around the technologies you use most.

Teams

Q&A for work

Connect and share knowledge within a single location that is structured and easy to search.

Currently, I am producing a WebSocket client in python using asyncio that connects to a server every second until I interrupt it with my keyboard and outputs the response from the server to a .csv file. When I run the script, It successfully runs, but I get this in my terminal:

'coroutine' object is not callable
When I press Ctrl + c to abort the code, I get the following:
sys:1: RuntimeWarning: coroutine 'test' was never awaited
What is the issue here and how do I resolve it? Mypython code is given below:
import asyncio
import websockets
import logging
import datetime
import time
starttime = time.time() # start value for timed data acquisition
logger = logging.getLogger('websockets')
logger.setLevel(logging.INFO)  #Switch to DEBUG for full error information
logger.addHandler(logging.StreamHandler())
class Timer: #class for asynchronous (non-blocking) counter
    def __init__(self, interval, first_immediately, callback):
        self._interval = interval
        self._first_immediately = first_immediately
        self._callback = callback
        self._is_first_call = True
        self._ok = True
        self._task = asyncio.ensure_future(self._job())
        print("init timer done")
    async def _job(self):
            while self._ok:
                if not self._is_first_call or not self._first_immediately:
                    await asyncio.sleep(self._interval)
                await self._callback(self)
                self._is_first_call = False
        except Exception as ex:
            print(ex)
    def cancel(self):
        self._ok = False
        self._task.cancel()
async def test():
    async with websockets.connect("ws://198.162.1.177:80/", ping_interval=None) as websocket:
        await websocket.send(str(1.001))  #send a message to the websocket server
        response = await websocket.recv() #wait to get a response from the server
        print(response)
        dataline_pv1 = datetime.datetime.today().isoformat() + "," + str(response) + "," + str(0) + "\n" # format and assemble data line
        file_name_pv1 = '{:%Y%m%d}'.format(datetime.datetime.today()) + "_flow.csv" # generate file name
        with open(file_name_pv1, "a") as etherm_file1: # append dataline to file
            etherm_file1.write(dataline_pv1)
#asyncio.get_event_loop().run_forever(test()) # run until test() is finished while True:
timer = Timer(interval=1, first_immediately=True, callback=test())
loop = asyncio.get_event_loop()
    asyncio.ensure_future(test())
    loop.run_forever()
except KeyboardInterrupt:
    timer.cancel()
finally:
    print("Closing Loop")
    loop.close()
You have error in your Timer initialization. You places coroutine object, but you need callable function instead.
Change initialization from
timer = Timer(interval=1, first_immediately=True, callback=test())
timer = Timer(interval=1, first_immediately=True, callback=test)
It seems like you are calling test rather than passing the coroutine test in places where you expecting a value. The result of test() (calling test) is a coroutine object that is never awaited. Therefore, you should treat test like a value rather than a function which you are calling.
Relevant lines to look at:
timer = Timer(interval=1, first_immediately=True, callback=test)
#                                                             ^
                so the line asyncio.ensure_future(test()) should then be changed to asyncio.ensure_future(test) too?
– tjsmert44
                Jul 15, 2021 at 8:08
You should pass test instead of test() to callback kwarg otherwise, test() returns a coroutine object that is never awaited (thus the RuntimeWarning you are getting at the Timer(interval=1, first_immediately=True, callback=test()) line and the 'coroutine' object is not callable error you are getting due to the await self._callback(self) line)
        
Thanks for contributing an answer to Stack Overflow!
Please be sure to answer the question. Provide details and share your research!
But avoid …
Asking for help, clarification, or responding to other answers.
Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.