Taiga Develop 3 weeks ago
parent
commit
28a114725e
8 changed files with 183 additions and 7 deletions
  1. 4 0
      source/__init__.py
  2. 26 0
      source/decoder.py
  3. 55 0
      source/encoder.py
  4. 6 7
      source/message.py
  5. 12 0
      source/protocol.py
  6. 10 0
      source/string_builder.py
  7. 40 0
      tests/001-message.py
  8. 30 0
      tests/002-encoder.py

+ 4 - 0
source/__init__.py

@@ -1 +1,5 @@
 from .protocol import protocol
+from .message import message
+from .message import message_builder
+from .string_builder import string_builder
+from .encoder import encoder

+ 26 - 0
source/decoder.py

@@ -0,0 +1,26 @@
+from .message import message
+from .message import message_builder
+from .protocol import protocol
+
+class decoder(protocol):
+    def __init__(self, content: str) -> None:
+        self.__content = content
+        self.__section = None
+
+    @property
+    def _content(self) -> str:
+        return self.__content
+
+    def __parse_line(self, line: str) -> None:
+        line = line.strip()
+
+        if self.__is_section(line):
+            self.__parse_section(line)
+            return
+
+        self.__parse_key_value(line)
+
+    def __is_section(self, line: str) -> bool:
+        return line[0] == self._section_buckles[0]
+
+    def __parse_section(self, line: str) -> 

+ 55 - 0
source/encoder.py

@@ -0,0 +1,55 @@
+import types
+import functools
+
+from .message import message
+from .protocol import protocol
+from .string_builder import string_builder
+
+class encoder(protocol):
+    def __init__(self, target: message) -> None:
+        self.__target = target
+
+    @property
+    def _target(self) -> message:
+        return self.__target
+
+    @functools.cache
+    def process(self) -> str:
+        result = string_builder()
+        result.add(self.__code_section_body(self._target.default))
+
+        for section, content in self._target.sections.items():
+            result.add(self.__code_section_header(section))
+            result.add(self.__code_section_body(content))
+
+        return result.build()
+
+    def __code_section_body(self, content: dict) -> str:
+        result = string_builder()
+        
+        for key, value in content.items():
+            result.add(self.__code_key_value(key, value))
+
+        return result.build()
+
+    def __code_section_header(self, section: str) -> str:
+        return string_builder() \
+        .add(self._section_buckles[0]) \
+        .add(section) \
+        .add(self._section_buckles[1]) \
+        .add(self._line_separator) \
+        .build()
+    
+    def __code_key_value(self, key: str, value: str) -> str:
+        return string_builder() \
+        .add(key) \
+        .add(self._setter) \
+        .add(value) \
+        .add(self._line_separator) \
+        .build()
+
+    def __str__(self) -> str:
+        return self.process()
+
+    def __repr(self) -> str:
+        return "Message encoder:\n" + str(self)

+ 6 - 7
source/message.py

@@ -1,5 +1,4 @@
 import types
-#from types import MappingProxyType
 
 class message:
     def __init__(self, default: dict, sections: dict) -> None:
@@ -8,7 +7,7 @@ class message:
         for key, value in sections.items():
             sections_proxy[key] = types.MappingProxyType(value)
 
-        self.__sections = types.MappingProxyType(sections_proxy
+        self.__sections = types.MappingProxyType(sections_proxy)
         self.__default = types.MappingProxyType(default)
 
     def get_key(self, key: str, section: str | None = None) -> str:
@@ -20,7 +19,7 @@ class message:
     def get_from_default(self, key: str) -> any:
         return self.__default[key]
 
-    def get_from_sections(self, section: str, key: str) -> any:
+    def get_from_section(self, section: str, key: str) -> any:
         return self.__sections[section][key]
 
     @property
@@ -37,8 +36,8 @@ class message_builder:
         self.__default = dict()
 
     def __add_section(self, section: str) -> object:
-        if not section in self.__content.keys():
-            selt.__sections[section] = dict()
+        if not section in self.__sections.keys():
+            self.__sections[section] = dict()
 
     def __set_section_key(self, section: str, key: str, value: str) -> None:
         self.__sections[section][key] = value
@@ -49,11 +48,11 @@ class message_builder:
     def set_key(
         self, 
         key: str, 
-        value: any, 
+        value: str, 
         section: str | None = None
     ) -> object:
         key = key.strip()
-        value = str(value).strip()
+        value = value.strip()
 
         if len(key) == 0:
             raise RuntimeError("Key nam can nob being empty.")

+ 12 - 0
source/protocol.py

@@ -10,3 +10,15 @@ class protocol:
     @staticmethod
     def section_buckles() -> [str, str]:
         return ("[", "]")
+
+    @property
+    def _line_separator(self) -> str:
+        return self.__class__.line_separator()
+
+    @property
+    def _setter(self) -> str:
+        return self.__class__.setter()
+        
+    @property
+    def _section_buckles(self) -> [str, str]:
+        return ("[", "]")

+ 10 - 0
source/string_builder.py

@@ -0,0 +1,10 @@
+class string_builder:
+    def __init__(self) -> None: 
+        self.__content = list()
+
+    def add(self, content: str) -> object:
+        self.__content.append(content)
+        return self
+
+    def build(self) -> str:
+        return str().join(self.__content)

+ 40 - 0
tests/001-message.py

@@ -0,0 +1,40 @@
+import pathlib
+
+test_file = pathlib.Path(__file__)
+test_dir = test_file.parent
+project_dir = test_dir.parent
+
+import sys
+sys.path.append(str(project_dir.absolute()))
+
+import source as communication
+
+def main():
+    message = communication \
+    .message_builder() \
+    .set_key("test_key", "10") \
+    .set_key("other", "20") \
+    .set_key("in_section", "a", "sect") \
+    .set_key("protocol", "https", "conn") \
+    .set_key("ip", "10.0.0.1", "conn") \
+    .build()
+
+    def check_key(key: str, section: str | None, value: any) -> None:
+        result = message.get_key(key, section) == value
+        section = "default" if section is None else section
+        
+        if result:
+            print("Work [" + section + "." + key + "]!")
+            return
+
+        print("Not work [" + section + "." + key + "]!")
+        raise RuntimeError("Not property value!.")
+
+    check_key("test_key", None, "10")
+    check_key("other", None, "20")
+    check_key("in_section", "sect", "a")
+    check_key("protocol", "conn", "https")
+    check_key("ip", "conn", "10.0.0.1")
+    
+if __name__ == "__main__":
+    main()

+ 30 - 0
tests/002-encoder.py

@@ -0,0 +1,30 @@
+import pathlib
+
+test_file = pathlib.Path(__file__)
+test_dir = test_file.parent
+project_dir = test_dir.parent
+
+import sys
+sys.path.append(str(project_dir.absolute()))
+
+import source as communication
+
+def main():
+    print("Building message.")
+
+    message = communication.message_builder() \
+    .set_key("name", "cx-copter") \
+    .set_key("version", "1.1") \
+    .set_key("ssid", "UwUUwU", "wifi") \
+    .set_key("key", "password", "wifi") \
+    .set_key("TLS", "false", "socket") \
+    .build()
+
+    print("Encoding.")
+    encoded = communication.encoder(message).process()
+
+    print("Result:")
+    print(encoded)
+
+if __name__ == "__main__":
+    main()