usb_cdc.py

00001 #
00002 # DAPLink Interface Firmware
00003 # Copyright (c) 2016-2016, ARM Limited, All Rights Reserved
00004 # SPDX-License-Identifier: Apache-2.0
00005 #
00006 # Licensed under the Apache License, Version 2.0 (the "License"); you may
00007 # not use this file except in compliance with the License.
00008 # You may obtain a copy of the License at
00009 #
00010 # http://www.apache.org/licenses/LICENSE-2.0
00011 #
00012 # Unless required by applicable law or agreed to in writing, software
00013 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
00014 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
00015 # See the License for the specific language governing permissions and
00016 # limitations under the License.
00017 #
00018 
00019 import usb.util
00020 
00021 
00022 class USBCdc (object):
00023     """Wrapper class for a CDC usb device"""
00024 
00025     # Communication commands documented in
00026     # PSTN120 inside CDC1.2_WMC1.1_012011
00027 
00028     CLASS_CDC_DATA = 0xa
00029     CLASS_CDC_COMM = 0x2
00030 
00031     FORMAT_STOP_BITS_1_0 = 0
00032     FORMAT_STOP_BITS_1_5 = 1
00033     FORMAT_STOP_BITS_2_0 = 2
00034 
00035     PARITY_NONE = 0
00036     PARITY_ODD = 1
00037     PARITY_EVEN = 2
00038     PARITY_MARK = 3
00039     PARITY_SPACE = 4
00040 
00041     DATA_BITS_5 = 5
00042     DATA_BITS_6 = 6
00043     DATA_BITS_7 = 7
00044     DATA_BITS_8 = 8
00045     DATA_BITS_16 = 16
00046 
00047     SEND_BREAK_ON = 0xFFFF
00048     SEND_BREAK_OFF = 0x0000
00049 
00050     def __init__(self, device):
00051         self._dev  = device
00052         self._if_data  = None
00053         self._if_comm  = None
00054         self.ep_data_out  = None
00055         self.ep_data_in  = None
00056         self._locked  = False
00057         self.timeout  = 10000
00058 
00059         # Find interfaces
00060         for interface in device.get_active_configuration():
00061             if interface.bInterfaceClass == USBCdc.CLASS_CDC_DATA:
00062                 assert self._if_data  is None
00063                 self._if_data  = interface
00064             if interface.bInterfaceClass == USBCdc.CLASS_CDC_COMM:
00065                 assert self._if_comm  is None
00066                 self._if_comm  = interface
00067         assert self._if_data  is not None
00068         assert self._if_comm  is not None
00069 
00070         # Find endpoints
00071         for endpoint in self._if_data :
00072             if endpoint.bEndpointAddress & 0x80:
00073                 assert self.ep_data_in  is None
00074                 self.ep_data_in  = endpoint
00075             else:
00076                 assert self.ep_data_out  is None
00077                 self.ep_data_out  = endpoint
00078         assert self.ep_data_in  is not None
00079         assert self.ep_data_out  is not None
00080 
00081     def lock (self):
00082         """Acquire exclisive access to CDC"""
00083         assert not self._locked 
00084 
00085         for interface in (self._if_data , self._if_comm ):
00086             num = interface.bInterfaceNumber
00087             try:
00088                 if self._dev .is_kernel_driver_active(num):
00089                     self._dev .detach_kernel_driver(num)
00090             except NotImplementedError:
00091                 pass
00092             except usb.core.USBError:
00093                 pass
00094             usb.util.claim_interface(self._dev , num)
00095         self._locked  = True
00096 
00097     def unlock (self):
00098         """Release exclusive access to CDC"""
00099         assert self._locked 
00100 
00101         for interface in (self._if_data , self._if_comm ):
00102             num = interface.bInterfaceNumber
00103             usb.util.release_interface(self._dev , num)
00104             try:
00105                 self._dev .attach_kernel_driver(num)
00106             except NotImplementedError:
00107                 pass
00108             except usb.core.USBError:
00109                 pass
00110         self._locked  = False
00111 
00112     def set_line_coding (self, baud, fmt=FORMAT_STOP_BITS_1_0,
00113                         parity=PARITY_NONE, databits=DATA_BITS_8):
00114         """Send the SetLineCoding CDC command"""
00115         assert self._locked 
00116 
00117         data = bytearray(7)
00118         data[0] = (baud >> (8 * 0)) & 0xFF
00119         data[1] = (baud >> (8 * 1)) & 0xFF
00120         data[2] = (baud >> (8 * 2)) & 0xFF
00121         data[3] = (baud >> (8 * 3)) & 0xFF
00122         data[4] = fmt
00123         data[5] = parity
00124         data[6] = databits
00125 
00126         request_type = 0x21
00127         request = 0x20                              # SET_LINE_CODING
00128         value = 0                                   # Always 0 for this request
00129         index = self._if_comm .bInterfaceNumber      # Communication interface
00130         self._dev .ctrl_transfer(request_type, request, value, index, data,
00131                                 self.timeout )
00132 
00133     def get_line_coding (self):
00134         """Send the GetLineCoding CDC command
00135 
00136         Returns a tuple containing
00137         baud, fmt, parity, databits
00138         """
00139         assert self._locked 
00140 
00141         request_type = 0xA1
00142         request = 0x21                              # GET_LINE_CODING
00143         value = 0                                   # Always 0 for this request
00144         index = self._if_comm .bInterfaceNumber      # Communication interface
00145         resp = self._dev .ctrl_transfer(request_type, request, value, index, 7,
00146                                        self.timeout )
00147         baud = (((resp[0] & 0xFF) << (8 * 0)) |
00148                 ((resp[1] & 0xFF) << (8 * 1)) |
00149                 ((resp[2] & 0xFF) << (8 * 2)) |
00150                 ((resp[3] & 0xFF) << (8 * 3)))
00151         fmt = resp[4]
00152         parity = resp[5]
00153         databits = resp[6]
00154         return (baud, fmt, parity, databits)
00155 
00156     def send_break (self, break_time):
00157         """Send the SendBreak CDC command"""
00158         assert self._locked 
00159         assert break_time & ~0xFFFF == 0, "Value outside of supported range"
00160 
00161         request_type = 0x21
00162         request = 0x23                              # SEND_BREAK
00163         value = break_time                          # Duration of break in ms
00164         index = self._if_comm .bInterfaceNumber      # Communication interface
00165         self._dev .ctrl_transfer(request_type, request, value, index, None,
00166                                 self.timeout )
00167 
00168     def read (self, size, timeout=None):
00169         """Read from the CDC data endpoint"""
00170         assert self._locked 
00171 
00172         if timeout is None:
00173             timeout = self.timeout 
00174         return self.ep_data_in .read(size, timeout)
00175 
00176     def write (self, data, timeout=None):
00177         """Write to the CDC data endpoint"""
00178         assert self._locked 
00179 
00180         if timeout is None:
00181             timeout = self.timeout 
00182         self.ep_data_out .write(data, self.timeout )