Kiss.__init__() - Code Metrics - Inspection of "Made some OOP improvements." - Syncleus/apex - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#13)

by Jeffrey

created 2016-09-24 16:25 UTC

Kiss.init() A

↳ Parent: Kiss

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	1
c	3
b	0
f	0
dl	0
loc	3
rs	10

#!/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
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

    @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

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

Syncleus / apex

Pull Request — master (#13)

Kiss.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Kiss.init() A