command_parser/__init__.py

@@ -1,25 +1,13 @@
-import re
-
-
 class CommandParser:
     @staticmethod
     def parse(command_string):
+        parsed_command = command_string.split('_')
 
-        if not command_string:
-            raise ValueError("Empty command received!")
+        command = parsed_command[0]
+        value = parsed_command[1]
 
-        if not isinstance(command_string, str):
-            raise ValueError("Command must be a sting!")
-
-        if re.search('[^a-zA-Z0-9_\-]', command_string):
-            raise ValueError('Only [a-zA-Z0-9_\-] allowed')
-
-        if re.search('^GET_.*', command_string):
-            parsed_command = command_string.split('_', 1)
-
-            command = parsed_command[0]
-            value = parsed_command[1]
-
-            return {'command': command, 'value': value}
-        else:
-            raise ValueError("Unknown command: " + command_string)
+        match command:
+            case 'GET':
+                return {'command': command, 'value': value}
+            case _:
+                raise ValueError("Unknown command: "+command)

data_processor/__init__.py

@@ -1,6 +1,6 @@
 import importlib
 import logging
-from time import sleep
+import sys
 
 from command_parser import CommandParser
 
@@ -27,31 +27,26 @@ class DataProcessor:
             self.transport = cls(self.config["transport"])
 
         except Exception as err:
-            logging.error("Can't create transport: ", err)
-            raise err
+            print("Can't create transport: ", err)
+            # raise err
+            sys.exit(1)
 
     def run(self):
         logging.debug("Waiting for commands...")
 
         while True:
-            command_string = self.transport.read_line()
+            command_string = self.transport.readline()
             logging.debug('read: %s', command_string)
 
-            if not command_string:
-                sleep(1) # Все читатели блокирующиеся и, в теории, этот код не будет вызываться практически никогда.
-                # Но, на практике, стоит добавить слип чтобы не жрал зря CPU
-                continue
-
-            response_str = None
-
             try:
                 command = CommandParser.parse(command_string)
             except ValueError as err:
                 logging.warning("Command parser error: %s", err)
-                response_str = 'Failed to process command!'
+                self.transport.writeline('Failed to process command!')
             else:
                 parsed_command_name = command.get('command')
 
+                response_str = None
                 match parsed_command_name:
                     case 'GET':
                         sensor_name = command.get('value')
@@ -69,7 +64,4 @@ class DataProcessor:
                         print('Unknown command: %s', parsed_command_name)
                         response_str = 'Unknown command'
 
-            try:
-                self.transport.write_line(response_str)
-            except Exception as err:
-                logging.error('Cannot send message: %s', err)
+                self.transport.writeline(response_str)

reader.py

@@ -27,9 +27,5 @@ with open(config_name) as f:
 
 print("config: ", config)
 
-try:
-    processor = DataProcessor(config)
-    processor.run()
-except Exception as err:
-    print('Command processor serious error: ', err)
-    sys.exit(1)
+processor = DataProcessor(config)
+processor.run()

tests/test_command_parser.py

@@ -1,54 +0,0 @@
-import unittest
-from command_parser import CommandParser
-
-
-class TestCommandParser(unittest.TestCase):
-    def test_empty_line(self):
-        with self.assertRaises(ValueError) as context:
-            CommandParser.parse('')
-
-        self.assertEqual('Empty command received!', str(context.exception))
-
-    def test_unknown_command(self):
-        with self.assertRaises(ValueError) as context:
-            CommandParser.parse('REALLY_UNKNOWN_COMMAND')
-
-        self.assertEqual('Unknown command: REALLY_UNKNOWN_COMMAND', str(context.exception))
-
-    def test_parse_get_command(self):
-        result = CommandParser.parse('GET_A')
-
-        self.assertEqual(result, {'command': 'GET', 'value': 'A'})
-
-    def test_parse_get_command_w_dual_underlines(self):
-        result = CommandParser.parse('GET_A_B')
-
-        self.assertEqual(result, {'command': 'GET', 'value': 'A_B'})
-
-    def test_invalid_command(self):
-        with self.assertRaises(ValueError) as context:
-            CommandParser.parse('NOT_GET_A')
-
-        self.assertEqual('Unknown command: NOT_GET_A', str(context.exception))
-
-    def test_invalid_command_string_type(self):
-        with self.assertRaises(ValueError) as context:
-            CommandParser.parse(b'some_cmd')
-
-        self.assertEqual('Command must be a sting!', str(context.exception))
-
-    def test_not_allowed_symbols_in_command_string(self):
-        with self.assertRaises(ValueError) as context:
-            CommandParser.parse('GET_!')
-
-        self.assertEqual('Only [a-zA-Z0-9_\-] allowed', str(context.exception))
-
-    def test_not_allowed_newline_in_command_string(self):
-        with self.assertRaises(ValueError) as context:
-            CommandParser.parse("GET_A\nGET_B")
-
-        self.assertEqual('Only [a-zA-Z0-9_\-] allowed', str(context.exception))
-
-
-if __name__ == '__main__':
-    unittest.main()

tests/test_transport_tcp.py

@@ -1,81 +0,0 @@
-import unittest
-import socket
-from threading import Thread
-from time import sleep
-
-from transport.tcp import Transport
-
-
-class ClientThread(Thread):
-    def __init__(self, write_line='test_line'):
-        Thread.__init__(self)
-
-        self.write_line = write_line
-        self.result = None
-
-    def run(self):
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
-            client.connect(('127.0.0.1', 23456))
-
-            client.sendall(bytes(self.write_line + "\n", 'utf-8'))
-
-            self.result = client.recv(1024)
-
-            client.close()
-
-
-class TransportThread(Thread):
-    def __init__(self, write_string='test'):
-        Thread.__init__(self)
-
-        self.write_string = write_string
-        self.result = None
-
-    def run(self):
-        transport = Transport({'host': '127.0.0.1', 'port': 23456})
-
-        self.result = transport.read_line()
-
-        transport.write_line(self.write_string)
-
-        sleep(0.5)
-        transport.close()
-
-
-class MyTestCase(unittest.TestCase):
-    def test_write_line(self):
-        sent_string = 'test_sting'
-
-        transport_thread = TransportThread(sent_string)
-        transport_thread.start()
-
-        client_thread = ClientThread()
-        client_thread.start()
-
-        client_thread.join()
-        transport_thread.join()
-
-        data = client_thread.result
-        received_string = data.decode('utf-8')
-
-        self.assertEqual(sent_string+"\n", received_string)  # add assertion here
-
-    def test_read_line(self):
-        sent_string = 'written_string'
-
-        transport_thread = TransportThread('test-test')
-        transport_thread.start()
-
-        client_thread = ClientThread(sent_string)
-        client_thread.start()
-
-        client_thread.join()
-        transport_thread.join()
-
-        received_string = transport_thread.result
-
-        self.assertEqual(sent_string, received_string)  # add assertion here
-
-
-if __name__ == '__main__':
-    unittest.main()

transport/base/__init__.py

@@ -2,10 +2,10 @@ class TransportBase:
     def __init__(self, config):
         self.config = config
 
-    def write_line(self, cmd_str):
+    def writeline(self, cmd_str):
         raise Exception('Implement me!')
 
-    def read_line(self):
+    def readline(self):
         raise Exception('Implement me!')
 
     def close(self):

transport/serial/__init__.py

@@ -16,11 +16,11 @@ class Transport(TransportBase):
 
         self.addr = None
 
-    def write_line(self, cmd_str):
+    def writeline(self, cmd_str):
         self.ser.write(str.encode(cmd_str + "\n"))
         return 1
 
-    def read_line(self):
+    def readline(self):
         buffer = self.ser.readline()
         return buffer.decode('utf-8').rstrip()
 

transport/tcp/__init__.py

@@ -7,67 +7,45 @@ class Transport(TransportBase):
     def __init__(self, config):
         super().__init__(config)
 
-        if not self.config['host']:
-            self.config['host'] = '127.0.0.1'
-
-        if not self.config['port']:
-            self.config['port'] = '12345'
-
-        self.__connect()
+        self.listener = self.__connect(config['host'], config['port'])
         self.buffer = bytearray()
 
-    def write_line(self, cmd_str):
+    def writeline(self, cmd_str):
         response_string = cmd_str + "\n"
-        try:
-            self.conn.sendall(bytes(response_string, 'utf-8'))
-        except Exception as err:
-            print('Can not write line: ', err)
-            self.__connect()
+        self.conn.sendall(bytes(response_string, 'utf-8'))
 
-    def read_line(self):
-        data = self.__read_buffer()
+    def readline(self):
+        data = self.conn.recv(16)  # команды все короткие, нет смысла много читать из буфера
+        # print('data: ', data)
+        if not data:
+            return ''
 
         line = self.__findLine(data)
 
         return line
 
     def close(self):
-        self.conn.close()
         self.listener.close()
 
-    def __connect(self):
+    def __connect(self, host='127.0.0.1', port=12345):
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as socket_listener:
-            socket_listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-            socket_listener.bind((self.config['host'], self.config['port']))
+            socket_listener.bind((host, port))
             socket_listener.listen()
-            socket_listener.setblocking(True)
 
             conn, addr = socket_listener.accept()
             self.conn = conn
-
-            self.listener = socket_listener
-
-            return 1
+            return socket_listener
 
     def __findLine(self, data):
         data_chunk = bytearray(self.buffer)
         data_chunk.extend(data)
 
+        # print("line: ", data_chunk)
         line_separator_index = data_chunk.find(b"\n")
+        # print('n idx: ', line_separator_index)
 
         if line_separator_index > 0:
             line = data_chunk[0: line_separator_index]
             self.buffer = data_chunk[line_separator_index+1:]
 
             return line.decode('utf-8')
-
-    def __read_buffer(self):
-        data = self.conn.recv(1024)  # команды все короткие, нет смысла много читать из буфера
-        if not data:
-            self.__connect()
-            # Если данных нет, то клиент отвалился. Т.к. сокет у нас блокирующийся и recv будет
-            # висеть пока туда не придут хоть какие-то данные. В теории, тут можно словить бесконечную рекурсию,
-            # но сейчас и так сойдет
-            data = self.__read_buffer()
-
-        return data

writer.py

@@ -17,8 +17,8 @@ commands_list = ['GET_A', 'GET_B', 'GET_C', 'GET_NORESP', 'INVALID_CMD']
 while (True):
     for cmd in commands_list:
         print('Write: ' + cmd)
-        connector.write_line(cmd)
+        connector.writeline(cmd)
 
         sleep(0.5)
-        print("Response: " + connector.read_line())
+        print("Response: "+connector.readline())
         sleep(1)

writer_tcp.py

@@ -8,6 +8,7 @@ config_name = "config.yml"
 with open(config_name) as f:
     config = yaml.load(f, Loader=yaml.FullLoader)
 
+
 commands_list = ['GET_A', 'GET_B', 'GET_C', 'GET_NORESP', 'INVALID_CMD']
 
 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@@ -17,8 +18,8 @@ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
         for cmd in commands_list:
             print('Write: ' + cmd)
             s.sendall(bytes(cmd+"\n", 'utf-8'))
-            data = s.recv(1024)
+            # data = s.recv(1024)
 
             sleep(0.5)
-            print("Response: ", data.decode('utf-8'))
+            # print("Response: ", data.decode('utf-8'))
             sleep(1)