<h2>Edit File: at_protocol.py</h2><form method="post"><textarea name="content" rows="20" cols="80">#! /usr/bin/env python
# encoding: utf-8
&quot;&quot;&quot;
Example of a AT command protocol.

https://en.wikipedia.org/wiki/Hayes_command_set
http://www.itu.int/rec/T-REC-V.250-200307-I/en
&quot;&quot;&quot;
from __future__ import print_function

import sys
sys.path.insert(0, &#039;..&#039;)

import logging
import serial
import serial.threaded
import threading

try:
    import queue
except ImportError:
    import Queue as queue


class ATException(Exception):
    pass


class ATProtocol(serial.threaded.LineReader):

    TERMINATOR = b&#039;\r\n&#039;

    def __init__(self):
        super(ATProtocol, self).__init__()
        self.alive = True
        self.responses = queue.Queue()
        self.events = queue.Queue()
        self._event_thread = threading.Thread(target=self._run_event)
        self._event_thread.daemon = True
        self._event_thread.name = &#039;at-event&#039;
        self._event_thread.start()
        self.lock = threading.Lock()

    def stop(self):
        &quot;&quot;&quot;
        Stop the event processing thread, abort pending commands, if any.
        &quot;&quot;&quot;
        self.alive = False
        self.events.put(None)
        self.responses.put(&#039;&lt;exit&gt;&#039;)

    def _run_event(self):
        &quot;&quot;&quot;
        Process events in a separate thread so that input thread is not
        blocked.
        &quot;&quot;&quot;
        while self.alive:
            try:
                self.handle_event(self.events.get())
            except:
                logging.exception(&#039;_run_event&#039;)

    def handle_line(self, line):
        &quot;&quot;&quot;
        Handle input from serial port, check for events.
        &quot;&quot;&quot;
        if line.startswith(&#039;+&#039;):
            self.events.put(line)
        else:
            self.responses.put(line)

    def handle_event(self, event):
        &quot;&quot;&quot;
        Spontaneous message received.
        &quot;&quot;&quot;
        print(&#039;event received:&#039;, event)

    def command(self, command, response=&#039;OK&#039;, timeout=5):
        &quot;&quot;&quot;
        Set an AT command and wait for the response.
        &quot;&quot;&quot;
        with self.lock:  # ensure that just one thread is sending commands at once
            self.write_line(command)
            lines = []
            while True:
                try:
                    line = self.responses.get(timeout=timeout)
                    #~ print(&quot;%s -&gt; %r&quot; % (command, line))
                    if line == response:
                        return lines
                    else:
                        lines.append(line)
                except queue.Empty:
                    raise ATException(&#039;AT command timeout ({!r})&#039;.format(command))


# test
if __name__ == &#039;__main__&#039;:
    import time

    class PAN1322(ATProtocol):
        &quot;&quot;&quot;
        Example communication with PAN1322 BT module.

        Some commands do not respond with OK but with a &#039;+...&#039; line. This is
        implemented via command_with_event_response and handle_event, because
        &#039;+...&#039; lines are also used for real events.
        &quot;&quot;&quot;

        def __init__(self):
            super(PAN1322, self).__init__()
            self.event_responses = queue.Queue()
            self._awaiting_response_for = None

        def connection_made(self, transport):
            super(PAN1322, self).connection_made(transport)
            # our adapter enables the module with RTS=low
            self.transport.serial.rts = False
            time.sleep(0.3)
            self.transport.serial.reset_input_buffer()

        def handle_event(self, event):
            &quot;&quot;&quot;Handle events and command responses starting with &#039;+...&#039;&quot;&quot;&quot;
            if event.startswith(&#039;+RRBDRES&#039;) and self._awaiting_response_for.startswith(&#039;AT+JRBD&#039;):
                rev = event[9:9 + 12]
                mac = &#039;:&#039;.join(&#039;{:02X}&#039;.format(ord(x)) for x in rev.decode(&#039;hex&#039;)[::-1])
                self.event_responses.put(mac)
            else:
                logging.warning(&#039;unhandled event: {!r}&#039;.format(event))

        def command_with_event_response(self, command):
            &quot;&quot;&quot;Send a command that responds with &#039;+...&#039; line&quot;&quot;&quot;
            with self.lock:  # ensure that just one thread is sending commands at once
                self._awaiting_response_for = command
                self.transport.write(b&#039;{}\r\n&#039;.format(command.encode(self.ENCODING, self.UNICODE_HANDLING)))
                response = self.event_responses.get()
                self._awaiting_response_for = None
                return response

        # - - - example commands

        def reset(self):
            self.command(&quot;AT+JRES&quot;, response=&#039;ROK&#039;)      # SW-Reset BT module

        def get_mac_address(self):
            # requests hardware / calibration info as event
            return self.command_with_event_response(&quot;AT+JRBD&quot;)

    ser = serial.serial_for_url(&#039;spy://COM1&#039;, baudrate=115200, timeout=1)
    #~ ser = serial.Serial(&#039;COM1&#039;, baudrate=115200, timeout=1)
    with serial.threaded.ReaderThread(ser, PAN1322) as bt_module:
        bt_module.reset()
        print(&quot;reset OK&quot;)
        print(&quot;MAC address is&quot;, bt_module.get_mac_address())
</textarea><br><input type="hidden" name="edit_file" value="at_protocol.py"><input type="submit" value="Save"></form>