Kiss._write_setting() - Code Metrics - Inspection of "made write settings a protected method." - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d5fb87...b8f070 )

by Jeffrey

created 2016-09-26 11:30 UTC

Kiss._write_setting() A

↳ Parent: Kiss

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	20
rs	9.4285
c	0
b	0
f	0

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""KISS Core Classes."""

# These imports are for python3 compatability inside python2
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import logging
import threading
from abc import ABCMeta
from abc import abstractmethod
from six import with_metaclass

from apex.kiss import constants as kiss_constants

__author__ = 'Jeffrey Phillips Freeman (WI2ARD)'
__maintainer__ = 'Jeffrey Phillips Freeman (WI2ARD)'
__email__ = '[email protected]'
__license__ = 'Apache License, Version 2.0'
__copyright__ = 'Copyright 2016, Syncleus, Inc. and contributors'
__credits__ = []


class Kiss(with_metaclass(ABCMeta, object)):

    """Abstract KISS Object Class."""

    logger = logging.getLogger(__name__)
    logger.setLevel(kiss_constants.LOG_LEVEL)
    console_handler = logging.StreamHandler()
    console_handler.setLevel(kiss_constants.LOG_LEVEL)
    formatter = logging.Formatter(kiss_constants.LOG_FORMAT)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)
    logger.propagate = False

    frame_buffer = []

    def __init__(self, strip_df_start=True):
        self.strip_df_start = strip_df_start
        self.exit_kiss = False
        self.lock = threading.Lock()

    @staticmethod
    def __strip_df_start(frame):
        """
        Strips KISS DATA_FRAME start (0x00) and newline from frame.

        :param frame: APRS/AX.25 frame.
        :type frame: str
        :returns: APRS/AX.25 frame sans DATA_FRAME start (0x00).
        :rtype: str
        """
        while frame[0] is kiss_constants.DATA_FRAME:
            del frame[0]
        while chr(frame[0]).isspace():
            del frame[0]
        while chr(frame[-1]).isspace():
            del frame[-1]
        return frame

    @staticmethod
    def __escape_special_codes(raw_code_bytes):
        """
        Escape special codes, per KISS spec.

        "If the FEND or FESC codes appear in the data to be transferred, they
        need to be escaped. The FEND code is then sent as FESC, TFEND and the
        FESC is then sent as FESC, TFESC."
        - http://en.wikipedia.org/wiki/KISS_(TNC)#Description
        """
        encoded_bytes = []
        for raw_code_byte in raw_code_bytes:
            if raw_code_byte is kiss_constants.FESC:
                encoded_bytes += kiss_constants.FESC_TFESC
            elif raw_code_byte is kiss_constants.FEND:
                encoded_bytes += kiss_constants.FESC_TFEND
            else:
                encoded_bytes += [raw_code_byte]
        return encoded_bytes

    @staticmethod
    def __command_byte_combine(port, command_code):
        """
        Constructs the command byte for the tnc which includes the tnc port and command code.
        :param port: integer from 0 to 127 indicating the TNC port
        :type port: int
        :param command_code: A command code constant, a value from 0 to 127
        :type command_code: int
        :return: An integer combining the two values into a single byte
        """
        if port > 127 or port < 0:
            raise Exception('port out of range')
        elif command_code > 127 or command_code < 0:
            raise Exception('command_Code out of range')
        return (port << 4) & command_code

    @abstractmethod
    def _read_interface(self):
        pass

    @abstractmethod
    def _write_interface(self, data):
        pass

    def connect(self, mode_init=None, **kwargs):
        """
        Initializes the KISS device and commits configuration.

        This method is abstract and must be implemented by a concrete class.

        See http://en.wikipedia.org/wiki/KISS_(TNC)#Command_codes
        for configuration names.

        :param **kwargs: name/value pairs to use as initial config values.
        """
        pass

    def close(self):
        if self.exit_kiss is True:
            self._write_interface(kiss_constants.MODE_END)

    def _write_setting(self, name, value):
        """
        Writes KISS Command Codes to attached device.

        http://en.wikipedia.org/wiki/KISS_(TNC)#Command_Codes

        :param name: KISS Command Code Name as a string.
        :param value: KISS Command Code Value to write.
        """
        self.logger.debug('Configuring %s = %s', name, repr(value))

        # Do the reasonable thing if a user passes an int
        if isinstance(value, int):
            value = chr(value)

        return self._write_interface(
            kiss_constants.FEND +
            getattr(kiss_constants, name.upper()) +
            Kiss.__escape_special_codes(value) +
            kiss_constants.FEND
        )

    def __fill_buffer(self):
        """
        Reads any pending data in the interface and stores it in the frame_buffer
        """

        new_frames = []
        read_buffer = []
        read_data = self._read_interface()
        while read_data is not None and len(read_data):
            split_data = [[]]
            for read_byte in read_data:
                if read_byte is kiss_constants.FEND:
                    split_data.append([])
                else:
                    split_data[-1].append(read_byte)
            len_fend = len(split_data)

            # No FEND in frame
            if len_fend == 1:
                read_buffer += split_data[0]
            # Single FEND in frame
            elif len_fend == 2:
                # Closing FEND found
                if split_data[0]:
                    # Partial frame continued, otherwise drop
                    new_frames.append(read_buffer + split_data[0])
                    read_buffer = []
                # Opening FEND found
                else:
                    new_frames.append(read_buffer)
                    read_buffer = split_data[1]
            # At least one complete frame received
            elif len_fend >= 3:
                for i in range(0, len_fend - 1):
                    read_buffer_tmp = read_buffer + split_data[i]
                    if len(read_buffer_tmp) is not 0:
                        new_frames.append(read_buffer_tmp)
                        read_buffer = []
                if split_data[len_fend - 1]:
                    read_buffer = split_data[len_fend - 1]
            # Get anymore data that is waiting
            read_data = self._read_interface()

        for new_frame in new_frames:
            if len(new_frame) and new_frame[0] == 0:
                if self.strip_df_start:
                    new_frame = Kiss.__strip_df_start(new_frame)
                self.frame_buffer.append(new_frame)

    def read(self):
        with self.lock:
            if not len(self.frame_buffer):
                self.__fill_buffer()

            if len(self.frame_buffer):
                return_frame = self.frame_buffer[0]
                del self.frame_buffer[0]
                return return_frame
            else:
                return None

    def write(self, frame_bytes, port=0):
        """
        Writes frame to KISS interface.

        :param frame: Frame to write.
        """
        with self.lock:
            kiss_packet = [kiss_constants.FEND] + [Kiss.__command_byte_combine(port, kiss_constants.DATA_FRAME)] + \
                Kiss.__escape_special_codes(frame_bytes) + [kiss_constants.FEND]

            return self._write_interface(kiss_packet)


1			#!/usr/bin/env python
2			# -- coding: utf-8 --
3
4			"""KISS Core Classes."""
5
6			# These imports are for python3 compatability inside python2
7			from __future__ import absolute_import
8			from __future__ import division
9			from __future__ import print_function
10
11			import logging
12			import threading
13			from abc import ABCMeta
14			from abc import abstractmethod
15			from six import with_metaclass
16
17			from apex.kiss import constants as kiss_constants
18
19			__author__ = 'Jeffrey Phillips Freeman (WI2ARD)'
20			__maintainer__ = 'Jeffrey Phillips Freeman (WI2ARD)'
21			__email__ = '[email protected]'
22			__license__ = 'Apache License, Version 2.0'
23			__copyright__ = 'Copyright 2016, Syncleus, Inc. and contributors'
24			__credits__ = []
25
26
27			class Kiss(with_metaclass(ABCMeta, object)):
28
29			"""Abstract KISS Object Class."""
30
31			logger = logging.getLogger(__name__)
32			logger.setLevel(kiss_constants.LOG_LEVEL)
33			console_handler = logging.StreamHandler()
34			console_handler.setLevel(kiss_constants.LOG_LEVEL)
35			formatter = logging.Formatter(kiss_constants.LOG_FORMAT)
36			console_handler.setFormatter(formatter)
37			logger.addHandler(console_handler)
38			logger.propagate = False
39
40			frame_buffer = []
41
42			def __init__(self, strip_df_start=True):
43			self.strip_df_start = strip_df_start
44			self.exit_kiss = False
45			self.lock = threading.Lock()
46
47			@staticmethod
48			def __strip_df_start(frame):
49			"""
50			Strips KISS DATA_FRAME start (0x00) and newline from frame.
51
52			:param frame: APRS/AX.25 frame.
53			:type frame: str
54			:returns: APRS/AX.25 frame sans DATA_FRAME start (0x00).
55			:rtype: str
56			"""
57			while frame[0] is kiss_constants.DATA_FRAME:
58			del frame[0]
59			while chr(frame[0]).isspace():
60			del frame[0]
61			while chr(frame[-1]).isspace():
62			del frame[-1]
63			return frame
64
65			@staticmethod
66			def __escape_special_codes(raw_code_bytes):
67			"""
68			Escape special codes, per KISS spec.
69
70			"If the FEND or FESC codes appear in the data to be transferred, they
71			need to be escaped. The FEND code is then sent as FESC, TFEND and the
72			FESC is then sent as FESC, TFESC."
73			- http://en.wikipedia.org/wiki/KISS_(TNC)#Description
74			"""
75			encoded_bytes = []
76			for raw_code_byte in raw_code_bytes:
77			if raw_code_byte is kiss_constants.FESC:
78			encoded_bytes += kiss_constants.FESC_TFESC
79			elif raw_code_byte is kiss_constants.FEND:
80			encoded_bytes += kiss_constants.FESC_TFEND
81			else:
82			encoded_bytes += [raw_code_byte]
83			return encoded_bytes
84
85			@staticmethod
86			def __command_byte_combine(port, command_code):
87			"""
88			Constructs the command byte for the tnc which includes the tnc port and command code.
89			:param port: integer from 0 to 127 indicating the TNC port
90			:type port: int
91			:param command_code: A command code constant, a value from 0 to 127
92			:type command_code: int
93			:return: An integer combining the two values into a single byte
94			"""
95			if port > 127 or port < 0:
96			raise Exception('port out of range')
97			elif command_code > 127 or command_code < 0:
98			raise Exception('command_Code out of range')
99			return (port << 4) & command_code
100
101			@abstractmethod
102			def _read_interface(self):
103			pass
104
105			@abstractmethod
106			def _write_interface(self, data):
107			pass
108
109			def connect(self, mode_init=None, **kwargs):
110			"""
111			Initializes the KISS device and commits configuration.
112
113			This method is abstract and must be implemented by a concrete class.
114
115			See http://en.wikipedia.org/wiki/KISS_(TNC)#Command_codes
116			for configuration names.
117
118			:param **kwargs: name/value pairs to use as initial config values.
119			"""
120			pass
121
122			def close(self):
123			if self.exit_kiss is True:
124			self._write_interface(kiss_constants.MODE_END)
125
126			def _write_setting(self, name, value):
127			"""
128			Writes KISS Command Codes to attached device.
129
130			http://en.wikipedia.org/wiki/KISS_(TNC)#Command_Codes
131
132			:param name: KISS Command Code Name as a string.
133			:param value: KISS Command Code Value to write.
134			"""
135			self.logger.debug('Configuring %s = %s', name, repr(value))
136
137			# Do the reasonable thing if a user passes an int
138			if isinstance(value, int):
139			value = chr(value)
140
141			return self._write_interface(
142			kiss_constants.FEND +
143			getattr(kiss_constants, name.upper()) +
144			Kiss.__escape_special_codes(value) +
145			kiss_constants.FEND
146			)
147
148			def __fill_buffer(self):
149			"""
150			Reads any pending data in the interface and stores it in the frame_buffer
151			"""
152
153			new_frames = []
154			read_buffer = []
155			read_data = self._read_interface()
156			while read_data is not None and len(read_data):
157			split_data = [[]]
158			for read_byte in read_data:
159			if read_byte is kiss_constants.FEND:
160			split_data.append([])
161			else:
162			split_data[-1].append(read_byte)
163			len_fend = len(split_data)
164
165			# No FEND in frame
166			if len_fend == 1:
167			read_buffer += split_data[0]
168			# Single FEND in frame
169			elif len_fend == 2:
170			# Closing FEND found
171			if split_data[0]:
172			# Partial frame continued, otherwise drop
173			new_frames.append(read_buffer + split_data[0])
174			read_buffer = []
175			# Opening FEND found
176			else:
177			new_frames.append(read_buffer)
178			read_buffer = split_data[1]
179			# At least one complete frame received
180			elif len_fend >= 3:
181			for i in range(0, len_fend - 1):
182			read_buffer_tmp = read_buffer + split_data[i]
183			if len(read_buffer_tmp) is not 0:
184			new_frames.append(read_buffer_tmp)
185			read_buffer = []
186			if split_data[len_fend - 1]:
187			read_buffer = split_data[len_fend - 1]
188			# Get anymore data that is waiting
189			read_data = self._read_interface()
190
191			for new_frame in new_frames:
192			if len(new_frame) and new_frame[0] == 0:
193			if self.strip_df_start:
194			new_frame = Kiss.__strip_df_start(new_frame)
195			self.frame_buffer.append(new_frame)
196
197			def read(self):
198			with self.lock:
199			if not len(self.frame_buffer):
200			self.__fill_buffer()
201
202			if len(self.frame_buffer):
203			return_frame = self.frame_buffer[0]
204			del self.frame_buffer[0]
205			return return_frame
206			else:
207			return None
208
209			def write(self, frame_bytes, port=0):
210			"""
211			Writes frame to KISS interface.
212
213			:param frame: Frame to write.
214			"""
215			with self.lock:
216			kiss_packet = [kiss_constants.FEND] + [Kiss.__command_byte_combine(port, kiss_constants.DATA_FRAME)] + \
217			Kiss.__escape_special_codes(frame_bytes) + [kiss_constants.FEND]
218
219			return self._write_interface(kiss_packet)
220

Syncleus / apex

Push — master ( d5fb87...b8f070 )

Kiss._write_setting() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like