First commit, add all files

1 files changed, 236 insertions, 0 deletions

diff --git a/py05/ex0/data_processor.py b/py05/ex0/data_processor.py
new file mode 100755
index 0000000..56166ac
--- /dev/null
+++ b/py05/ex0/data_processor.py
@@ -0,0 +1,236 @@
+#!/usr/bin/env python3
+
+from abc import ABC, abstractmethod
+import typing
+import sys
+
+
+class DataProcessor(ABC):
+    def __init__(self, name: str) -> None:
+        self.storage: typing.List[typing.Any] = []
+        self.rank: int = 0
+        self.counter: int = 0
+        self.name = name
+
+    @abstractmethod
+    def validate(self, data: typing.Any) -> bool:
+        pass
+
+    @abstractmethod
+    def ingest(self, data: typing.Any) -> None:
+        pass
+
+    def output(self) -> tuple[int, str]:
+        # extract the oldest piece of data
+        # piece of data is removed
+        position = self.rank
+        self.rank += 1
+        if isinstance(self.storage[0], dict):
+            entry = self.storage[0]
+            output_tuple: tuple = tuple(entry.values())
+            to_return = (f"{output_tuple[0]}: {output_tuple[1]}")
+            self.storage.pop(0)
+            return position, to_return
+        to_return = self.storage[0]
+        self.storage.pop(0)
+        return position, to_return
+
+
+class NumericProcessor(DataProcessor):
+
+    def validate(self, data: int | float | typing.List[int | float]) -> bool:
+        try:
+            if isinstance(data, int):
+                self.counter += 1
+                return True
+            elif isinstance(data, float):
+                self.counter += 1
+                return True
+            elif isinstance(data, typing.List):
+                type_list = "yes"
+                for item in data:
+                    self.counter += 1
+                    if not isinstance(item, int | float):
+                        type_list = "no"
+                        self.counter -= 1
+                if type_list == "yes":
+                    return True
+                else:
+                    raise ValueError()
+            else:
+                raise ValueError()
+        except ValueError:
+            return False
+
+    def ingest(self, data: int | float | typing.List[int | float]) -> None:
+        try:
+            if isinstance(data, typing.List):
+                for item in data:
+                    if isinstance(item, int | float):
+                        self.storage.append(item)
+            elif isinstance(data, int | float):
+                self.storage.append(data)
+            else:
+                raise ValueError()
+        except ValueError:
+            print("Improper numeric data", file=sys.stderr)
+
+
+def is_number(s: str) -> bool:
+    try:
+        float(s)
+        return True
+    except (ValueError, TypeError):
+        return False
+
+
+class TextProcessor(DataProcessor):
+
+    def validate(self, data: str | typing.List[str]) -> bool:
+        try:
+            if isinstance(data, str):
+                if is_number(data):
+                    raise ValueError()
+                else:
+                    self.counter += 1
+                    return True
+            elif isinstance(data, typing.List):
+                for item in data:
+                    if isinstance(item, str):
+                        if is_number(item):
+                            raise ValueError()
+                            return False
+                        elif isinstance(item, dict):
+                            raise ValueError()
+                            return False
+                        else:
+                            self.counter += 1
+                    if isinstance(item, dict):
+                        raise ValueError()
+                        return False
+                return True
+            else:
+                raise ValueError()
+        except ValueError:
+            return False
+
+    def ingest(self, data: str | typing.List[str]) -> None:
+        try:
+            if isinstance(data, str):
+                self.storage.append(data)
+            elif isinstance(data, typing.List):
+                for item in data:
+                    if is_number(item):
+                        raise ValueError()
+                    elif isinstance(item, dict):
+                        raise ValueError()
+                    else:
+                        self.storage.append(item)
+            else:
+                raise ValueError()
+        except ValueError:
+            print("Improper text data", file=sys.stderr)
+
+
+class LogProcessor(DataProcessor):
+
+    def validate(self, data: dict | typing.List[dict]) -> bool:
+        try:
+            if isinstance(data, dict):
+                return True
+            if isinstance(data, typing.List):
+                for item in data:
+                    if isinstance(item, dict):
+                        self.counter += 1
+                    else:
+                        raise ValueError()
+                        return False
+                return True
+            else:
+                raise ValueError()
+        except ValueError:
+            return False
+
+    def ingest(self, data: dict | typing.List[dict]) -> None:
+        try:
+            if isinstance(data, dict):
+                for item in data:
+                    self.storage.append(item)
+            elif isinstance(data, typing.List):
+                for item in data:
+                    if isinstance(item, dict):
+                        self.storage.append(item)
+            else:
+                raise ValueError()
+        except ValueError:
+            print("Improper log data", file=sys.stderr)
+
+
+def testNumericProcessor() -> None:
+    is_num = NumericProcessor("Numeric Processor")
+    test_num: typing.List = [42, 42.0, 'FortyTwo']
+    for item in test_num:
+        print(f"Trying to validate input {item}: ", end='')
+        print(is_num.validate(item))
+    print("Testing invalid ingestion of string 'foo' without prior validation")
+    print("Got exception: ", end='')
+    # flush buffer
+    sys.stdout.flush()
+    # will raise a mypy error
+    is_num.ingest('foo')
+    is_num.ingest([1, 2, 3, 4, 5])
+    print(f"Processing data: {is_num.storage}")
+    print(f"Extracting {len(is_num.storage)} values...")
+    while is_num.storage:
+        position, value = is_num.output()
+        print(f"Numeric value {position} : {value}")
+
+
+def testTextProcessor() -> None:
+    is_text = TextProcessor("Text Processor")
+    # test_text: typing.List = ['42', '42.0', 'FortyTwo', ['More', '43']]
+    test_text: typing.List = ['42', '42.0', 'FortyTwo']
+    for item in test_text:
+        print(f"Trying to validate input {item}: ", end='')
+        print(is_text.validate(item))
+    test_list_text: typing.List = ['Hello', 'Nexus', 'FortyTwo']
+    print(f"Trying to validate '{test_list_text}': "
+          f"{is_text.validate(test_list_text)}")
+    is_text.ingest(test_list_text)
+    print(f"Processing data: {is_text.storage}")
+    print(f"Extracting {len(is_text.storage)} values...")
+    while is_text.storage:
+        position, value = is_text.output()
+        print(f"Text value {position} : {value}")
+
+
+def testLogProcessor() -> None:
+    log = [{'log_level': 'NOTICE',
+           'log_message': 'Connection to server'},
+           {'log_level': 'ERROR',
+           'log_message': 'Unauthorized access!!'}]
+    is_log = LogProcessor("Log Processor")
+    # will raise a mypy error
+    print(f"Trying to validate input 'Hello': {is_log.validate('Hello')}")
+    is_log.ingest(log)
+    print(f"Processing data: {is_log.storage}")
+    print(f"Extracting {len(is_log.storage)} values...")
+    while is_log.storage:
+        position, value = is_log.output()
+        print(f"Log entry {position} : {value}")
+
+
+def main() -> None:
+    print("=== Code Nexus - Data Processor ===")
+    print("Testing Numeric Processor...")
+    testNumericProcessor()
+    print()
+    print("Testing Text Processor...")
+    testTextProcessor()
+    print()
+    print("Testing Log Processor...")
+    testLogProcessor()
+
+
+if __name__ == "__main__":
+    main()