doorbell: telemetrix_local.py comparison

comparison telemetrix_local.py @ 9:7e19dffb767b default tip

rm more unused code and logging

author	drewp@bigasterisk.com
date	Mon, 06 Feb 2023 11:53:42 -0800
parents	5441581186f8
children

comparison

equal deleted inserted replaced

-:5441581186f8
+:7e19dffb767b
 # import struct
 import sys
 import time
 import logging
 log = logging.getLogger('tele')
+cmdlog = logging.getLogger('command')
 # noinspection PyPackageRequirementscd
 from serial.serialutil import SerialException
 # noinspection PyPackageRequirements
 from serial.tools import list_ports
 """
 # noinspection PyPep8,PyPep8
 def __init__(self, com_port=None,
-arduino_instance_id=1, arduino_wait=4,
+arduino_instance_id=1,
-sleep_tune=0.0001, autostart=True,
+autostart=True, shutdown_on_exception=True,
-loop=None, shutdown_on_exception=True,
-close_loop_on_shutdown=True,
 ip_address=None, ip_port=31335):
 """
 If you have a single Arduino connected to your computer,
 then you may accept all the default values.
 :param com_port: e.g. COM3 or /dev/ttyACM0.
 :param arduino_instance_id: Must match value in the Telemetrix4Arduino sketch
-:param arduino_wait: Amount of time to wait for an Arduino to
-fully reset itself.
-:param sleep_tune: A tuning parameter (typically not changed by user)
 :param autostart: If you wish to call the start method within
 your application, then set this to False.
-:param loop: optional user provided event loop
 :param shutdown_on_exception: call shutdown before raising
 a RunTimeError exception, or
 receiving a KeyboardInterrupt exception
-:param close_loop_on_shutdown: stop and close the event loop loop
-when a shutdown is called or a serial
-error occurs
 """
 # check to make sure that Python interpreter is version 3.8.3 or greater
 python_version = sys.version_info
 if python_version[:3] < (3, 8, 3):
 "required for use of this program.")
 # save input parameters
 self.com_port = com_port
 self.arduino_instance_id = arduino_instance_id
-self.arduino_wait = arduino_wait
-self.sleep_tune = sleep_tune
 self.autostart = autostart
 self.ip_address = ip_address
 self.ip_port = ip_port
 # if tcp, this variable is set to the connected socket
 self.sock = None
-# set the event loop
-if loop is None:
-self.loop = asyncio.get_event_loop()
-else:
-self.loop = loop
 self.shutdown_on_exception = shutdown_on_exception
-self.close_loop_on_shutdown = close_loop_on_shutdown
 # dictionaries to store the callbacks for each pin
 self.analog_callbacks = {}
 self.digital_callbacks = {}
 self.dht_count = 0
 # serial port in use
 self.serial_port: Optional[TelemetrixAioSerial] = None
-self.the_task = None
+self.report_task = None
-# flag to indicate we are in shutdown mode
-self.shutdown_flag = False
 # reported features
 self.reported_features = 0
 # To add a command to the command dispatch table, append here.
 print(f'TelemetrixAIO Version: {PrivateConstants.TELEMETRIX_AIO_VERSION}')
 print(f'Copyright (c) 2018-2021 Alan Yorinks All rights reserved.\n')
 if autostart:
-self.loop.run_until_complete(self.start_aio())
+raise NotImplementedError("you run start_aio")
 async def start_aio(self):
 """
 This method may be called directly, if the autostart
 parameter in __init__ is set to false.
 if self.shutdown_on_exception:
 await self.shutdown()
 raise RuntimeError('No Arduino Found or User Aborted Program')
 # using tcp/ip
 else:
-self.sock = TelemetrixAioSocket(self.ip_address, self.ip_port, self.loop)
+self.sock = TelemetrixAioSocket(self.ip_address, self.ip_port)
 await self.sock.start()
 # self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 # self.sock.connect((self.ip_address, self.ip_port))
 # print(f'Successfully connected to: {self.ip_address}:{self.ip_port}')
 # get arduino firmware version and print it
 firmware_version = await self._get_firmware_version()
-log.debug(f'start_aio p2 {firmware_version=}')
 if not firmware_version:
 log.info('*** Firmware Version retrieval timed out. ***')
 log.info('\nDo you have Arduino connectivity and do you have the ')
 log.info('Telemetrix4Arduino sketch uploaded to the board and are connected')
 log.info('to the correct serial port.\n')
 '"list_serial_ports" in your console.')
 if self.shutdown_on_exception:
 await self.shutdown()
 raise RuntimeError
 else:
-log.debug(f'p3')
 if firmware_version[1] < 5:
 raise RuntimeError('Please upgrade the server firmware to version 5.0.0 or greater')
 log.info(f'Telemetrix4Arduino Version Number: {firmware_version[1]}.'
 f'{firmware_version[2]}.{firmware_version[3]}')
 # start the command dispatcher loop
 command = [PrivateConstants.ENABLE_ALL_REPORTS]
-log.debug('start_aio send ENABLE_ALL_REPORTS')
 await self._send_command(command)
-if not self.loop:
+self.report_task = asyncio.create_task(self._arduino_report_dispatcher())
-log.debug('start_aio new loop')
+log.debug(f'create task for _arduino_report_dispatcher done: {self.report_task}')
-self.loop = asyncio.get_event_loop()
-self.the_task = self.loop.create_task(self._arduino_report_dispatcher())
-log.debug('create task for _arduino_report_dispatcher')
-log.debug(f'create task for _arduino_report_dispatcher done: {self.the_task}')
 # get the features list
 command = [PrivateConstants.GET_FEATURES]
-log.debug('send get_features')
 await self._send_command(command)
 # time.sleep(.5)
 # Have the server reset its data structures
 command = [PrivateConstants.RESET]
-log.debug('send reset')
 await self._send_command(command)
-log.debug(f'start_aio bye\n')
-async def get_event_loop(self):
-"""
-Return the currently active asyncio event loop
-:return: Active event loop
-"""
-return self.loop
 async def _find_arduino(self):
 """
 This method will search all potential serial ports for an Arduino
 containing a sketch that has a matching arduino_instance_id as
 serial_ports.append(self.serial_port)
 # display to the user
 print('\t' + port.device)
-# wait for arduino to reset
-print('\nWaiting {} seconds(arduino_wait) for Arduino devices to '
-'reset...'.format(self.arduino_wait))
-await asyncio.sleep(self.arduino_wait)
 print('\nSearching for an Arduino configured with an arduino_instance = ',
 self.arduino_instance_id)
 for serial_port in serial_ports:
 self.serial_port = serial_port
 log.info(f'Searching for correct arduino_instance_id: {self.arduino_instance_id}')
 while True:
 # if port is not found, a serial exception will be thrown
 log.info('Opening {} ...'.format(self.com_port))
 self.serial_port = TelemetrixAioSerial(self.com_port, 115200)
-# log.debug('await open')
 await self.serial_port.open()
-# print('Waiting {} seconds for the Arduino To Reset.'
-#       .format(self.arduino_wait))
-# await asyncio.sleep(self.arduino_wait)
 command = [PrivateConstants.ARE_U_THERE]
 try:
 i_am_here = await asyncio.wait_for(self._send_command(command, has_response=True), timeout=0.5)
 except asyncio.TimeoutError:
 log.info(f'no answer- retrying')
 This method retrieves the Arduino4Telemetrix firmware version
 :returns: Firmata firmware version
 """
 command = [PrivateConstants.GET_FIRMWARE_VERSION]
-log.debug('GET_FIRMWARE_VERSION')
 firmware_version = await self._send_command(command, has_response=True)
-log.debug(f'{firmware_version=}\n')
 return firmware_version
 async def analog_write(self, pin, value):
 """
 Set the specified pin to the specified value.
 """
 This method attempts an orderly shutdown
 If any exceptions are thrown, they are ignored.
 """
-self.shutdown_flag = True
 # stop all reporting - both analog and digital
 try:
 if self.serial_port:
 command = [PrivateConstants.STOP_ALL_REPORTS]
 await self._send_command(command)
-time.sleep(.5)
-if self.close_loop_on_shutdown:
-self.loop.stop()
 elif self.sock:
 command = [PrivateConstants.STOP_ALL_REPORTS]
 await self._send_command(command)
-self.the_task.cancel()
+self.report_task.cancel()
-time.sleep(.5)
-if self.close_loop_on_shutdown:
-self.loop.stop()
 except (RuntimeError, SerialException):
 pass
 async def disable_all_reporting(self):
 """
 if command:
 # the length of the list is added at the head
 command.insert(0, len(command))
 # print(command)
 send_message = bytes(command)
-log.debug(f'sending {CommandName[command[1]]} {command[2:]}')
+cmdlog.debug(f'sending {CommandName[command[1]]} {command[2:]}')
 if not self.ip_address:
 await self.serial_port.write(send_message)
 else:
 await self.sock.write(send_message)
 # await asyncio.sleep(.1)
 return None
 else:
 header = await self.serial_port.read(1)
 packet_length = header[0]
 pkt = await self.serial_port.read(packet_length)
-log.debug(f'got response {list(pkt)}')
+cmdlog.debug(f'got response {list(pkt)}')
 return pkt

Mercurial > code > home > repos > doorbell

comparison telemetrix_local.py @ 9:7e19dffb767b default tip