WIP: part 4 of module 6

05/ex0/stream_processor.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, Union
 from typing_extensions import override
 
 
@@ -111,8 +111,8 @@ class LogProcessor(DataProcessor):
 def class_tester() -> None:
     print("=== CODE NEXUS - DATA PROCESSOR FOUNDATION ===\n")
     print("Initializing Numeric Processor...")
-    data = [1, 2, 3, 4, 5]
+    data: Union[list[int | str], str] = [1, 2, 3, 4, 5]
-    processor = NumericProcessor()
+    processor: DataProcessor = NumericProcessor()
     res = processor.process(data)
     is_valid = processor.validate(data)
     formatted_output = processor.format_output(res)

05/ex1/data_stream.py

@@ -1,6 +1,4 @@
 from abc import ABC, abstractmethod
-from re import error
-import re
 from typing import Any, List, Optional, Dict, Union
 from typing_extensions import override
 
@@ -19,6 +17,8 @@ class DataStream(ABC):
         self, data_batch: List[Any], criteria: Optional[str] = None
     ) -> List[Any]:
         res = []
+        if not criteria:
+            return data_batch
         for data in data_batch:
             if data.split(":", 1)[0] in criteria:
                 res += [data]
@@ -31,7 +31,9 @@ class DataStream(ABC):
 
 
 class SensorStream(DataStream):
-    def __init__(self, stream_id: str, stream_type: str) -> None:
+    def __init__(
+        self, stream_id: str, stream_type: str = "Environmental Data"
+    ) -> None:
         super().__init__(stream_id, stream_type)
 
     def process_batch(self, data_batch: List[Any]) -> str:
@@ -72,7 +74,9 @@ class SensorStream(DataStream):
 
 
 class TransactionStream(DataStream):
-    def __init__(self, stream_id: str, stream_type: str) -> None:
+    def __init__(
+        self, stream_id: str, stream_type: str = "Financial Data"
+    ) -> None:
         super().__init__(stream_id, stream_type)
 
     def process_batch(self, data_batch: List[Any]) -> str:
@@ -125,7 +129,9 @@ class TransactionStream(DataStream):
 
 
 class EventStream(DataStream):
-    def __init__(self, stream_id: str, stream_type: str) -> None:
+    def __init__(
+        self, stream_id: str, stream_type: str = "System Events"
+    ) -> None:
         super().__init__(stream_id, stream_type)
 
     def process_batch(self, data_batch: List[Any]) -> str:
@@ -160,15 +166,17 @@ class EventStream(DataStream):
 
 class StreamProcessor:
     def __init__(self) -> None:
-        self.streams: Dict[
+        self.streams: Dict[str, List[DataStream]] = {
-            str, List[Union[SensorStream, TransactionStream, EventStream]]
+            "sensor": [],
-        ] = {"sensor": [], "transaction": [], "event": []}
+            "transaction": [],
+            "event": [],
+        }
 
     def process_batch(
         self,
         data_batch: List[Any],
-        stream: Union[SensorStream, TransactionStream, EventStream],
+        stream: DataStream,
-    ) -> Union[SensorStream, TransactionStream, EventStream]:
+    ) -> DataStream:
         try:
             stream.process_batch(data_batch)
             if isinstance(stream, SensorStream):
@@ -205,7 +213,7 @@ class StreamProcessor:
         return res
 
 
-def data_stream_tester():
+def data_stream_tester() -> None:
     print("=== CODE NEXUS - POLYMORPHIC STREAM SYSTEM ===\n")
     sensor_stream = SensorStream("SENSOR_001", "Environmental Data")
     print(f"Stream ID: {sensor_stream.stream_id},\
@@ -235,7 +243,7 @@ def data_stream_tester():
  events, {event_analysis['nb_error']} error detected")
 
 
-def stream_processor_tester():
+def stream_processor_tester() -> None:
     print("=== Polymorphic Stream Processing ===\n")
     processor = StreamProcessor()
     data_batch = [
@@ -243,7 +251,7 @@ def stream_processor_tester():
         ["buy:100", "sell:150"],
         ["login", "error", "logout", "login", "error"],
     ]
-    streams = [
+    streams: list[DataStream] = [
         SensorStream("SENSOR_001", "Environmental Data"),
         TransactionStream("TRANS_001", "Financial Data"),
         EventStream("EVENT_001", "System Events"),

05/ex2/nexus_pipeline.py

@@ -1,5 +1,5 @@
 from abc import ABC
-from typing import Any, List, Protocol, Union
+from typing import Any, Dict, List, Protocol, Union
 from typing_extensions import override
 
 
@@ -9,18 +9,35 @@ class ProcessingStage(Protocol):
 
 
 class InputStage:
-    def process(self, data: Any) -> Dict:
+    def process(self, data: Any) -> Dict[int, float]:
-        pass
+        res: Dict[int, float] = {}
+        i = 0
+        for n in data:
+            try:
+                if not isinstance(n, float):
+                    n = float(n)
+                res[i] = n
+                i += 1
+            except ValueError:
+                continue
+        return res
 
 
 class TransformStage:
-    def process(self, data: Any) -> Dict:
+    def process(self, data: Any) -> Dict[str, float]:
-        pass
+        res: Dict[str, Union[int, float]] = {}
+        res["processed_data"] = len(data)
+        try:
+            res["avg"] = sum(data.values()) / len(data)
+        except ZeroDivisionError:
+            res["avg"] = 0
+        return res
 
 
 class OutputStage:
     def process(self, data: Any) -> str:
-        pass
+        return f"Summary:\n\t\
+- Processed_data: {data['processed_data']}\n\t- avg temp: {data['avg']:.1f}°C"
 
 
 class ProcessingPipeline(ABC):
@@ -30,10 +47,12 @@ class ProcessingPipeline(ABC):
     def add_stage(
         self, stage: Union[InputStage, TransformStage, OutputStage]
     ) -> None:
-        pass
+        self.stages += [stage]
 
     def process(self, data: Any) -> Any:
-        pass
+        for stage in self.stages:
+            data = stage.process(data)
+        return data
 
 
 class JSONAdapter(ProcessingPipeline):
@@ -43,7 +62,10 @@ class JSONAdapter(ProcessingPipeline):
 
     @override
     def process(self, data: Any) -> Any:
-        return super().process(data)
+        res: List[Any] = []
+        for n in data:
+            res += [data[n]]
+        return super().process(res)
 
 
 class CSVAdapter(ProcessingPipeline):
@@ -53,7 +75,7 @@ class CSVAdapter(ProcessingPipeline):
 
     @override
     def process(self, data: Any) -> Any:
-        return super().process(data)
+        return super().process(data.split(","))
 
 
 class StreamAdapter(ProcessingPipeline):
@@ -63,12 +85,65 @@ class StreamAdapter(ProcessingPipeline):
 
     @override
     def process(self, data: Any) -> Any:
+        if not isinstance(data, List):
+            raise Exception
         return super().process(data)
 
 
-def tester():
+class NexusManager:
-    pass
+    def __init__(self, pipelines: List[ProcessingPipeline]) -> None:
+        self.pipelines: List[ProcessingPipeline] = pipelines
+
+    def add_pipeline(self, pipeline: ProcessingPipeline) -> None:
+        self.pipelines += [pipeline]
+
+    def process_data(self, data: Any) -> str:
+        res: str | None = None
+        for pipeline in self.pipelines:
+            pipeline.add_stage(InputStage())
+            pipeline.add_stage(TransformStage())
+            pipeline.add_stage(OutputStage())
+            try:
+                res = pipeline.process(data)
+                break
+            except Exception:
+                continue
+        if res is None:
+            return "[ERROR] Unknown format, incompatible with pipelines"
+        return res
+
+
+def tester() -> None:
+    print("=== CODE NEXUS - ENTERPRISE PIPELINE SYSTEM ===\n")
+    print("Initializing Nexus Manager...")
+    manager = NexusManager(
+        [
+            JSONAdapter("JSON_01"),
+            CSVAdapter("CSV_01"),
+            StreamAdapter("Stream_01"),
+        ]
+    )
+    print("\n=== test JSONAdapter ===")
+    data: Any = {"temp": 10, "est": "10.3", "t": 10.6, "p": "Hello"}
+    print(f"input: {data}")
+    res = manager.process_data(data)
+    print(f"Output: {res}\n")
+    print("\n=== test CSVAdapter ===")
+    data = "10,20,40,30"
+    print(f'input: "{data}"')
+    res = manager.process_data(data)
+    print(f"Output: {res}\n")
+    print("\n=== test StreamAdapter ===")
+    data = [10, 30, 0, "100"]
+    print(f"input: {data}")
+    res = manager.process_data(data)
+    print(f"Output: {res}\n")
+    print("\n=== test Invalid input ===")
+    data = 10
+    print(f"input: {data}")
+    res = manager.process_data(data)
+    print(f"Output: {res}\n")
 
 
 if __name__ == "__main__":
-    pass
+    tester()

06/alchemy/__init__.py

@@ -0,0 +1,5 @@
+from .elements import create_fire, create_water
+
+__version__ = "1.0.0"
+__author__ = "Master Pythonicus"
+__all__ = ["create_fire", "create_water"]

06/alchemy/elements.py

@@ -0,0 +1,14 @@
+def create_fire() -> str:
+    return "Fire element created"
+
+
+def create_water() -> str:
+    return "Water element created"
+
+
+def create_earth() -> str:
+    return "Earth element created"
+
+
+def create_air() -> str:
+    return "Air element created"

06/alchemy/grimoire/__init__.py

@@ -0,0 +1,6 @@
+from .spellbook import record_spell
+from .validator import validate_ingredients
+
+__version__ = "1.0.0"
+__author__ = "Master Pythonicus"
+__all__ = ["record_spell", "validate_ingredients"]

06/alchemy/grimoire/spellbook.py

@@ -0,0 +1,8 @@
+def record_spell(spell_name: str, ingredients: str) -> str:
+    from .validator import validate_ingredients
+
+    validation_res = validate_ingredients(ingredients)
+    if validation_res == f"{ingredients} - VALID":
+        return f"Spell recorded: {spell_name} ({validation_res})"
+    else:
+        return f"Spell rejected: {spell_name} ({validation_res})"

06/alchemy/grimoire/validator.py

@@ -0,0 +1,8 @@
+def validate_ingredients(ingredients: str) -> str:
+    try:
+        for ingredient in ingredients.split(" "):
+            if ingredient not in ["fire", "water", "earth", "air"]:
+                raise ValueError
+        return f"{ingredients} - VALID"
+    except ValueError:
+        return f"{ingredients} - INVALID"

06/alchemy/potions.py

@@ -0,0 +1,27 @@
+def healing_potion() -> str:
+    from .elements import create_water, create_fire
+
+    return f"Healing potion brewed with {create_fire()} and {create_water()}"
+
+
+def strength_potion() -> str:
+    from .elements import create_earth, create_fire
+
+    return f"Strength potion brewed with {create_earth()} and {create_fire()}"
+
+
+def invisibility_potion() -> str:
+    from .elements import create_air, create_water
+
+    return (
+        f"Invisibility potion brewed with {create_air()} and {create_water()}"
+    )
+
+
+def wisdom_potion() -> str:
+    from .elements import create_water, create_fire, create_air, create_earth
+
+    res = ""
+    for fn in (create_water(), create_fire(), create_air(), create_earth()):
+        res += fn
+    return f"Wisdom potion brewed with all elements: {res}"

06/alchemy/transmutation/__init__.py

@@ -0,0 +1,11 @@
+from .basic import lead_to_gold, stone_to_gem
+from .advanced import philosophers_stone, elixir_of_life
+
+__version__ = "1.0.0"
+__author__ = "Master Pythonicus"
+__all__ = [
+    "lead_to_gold",
+    "stone_to_gem",
+    "philosophers_stone",
+    "elixir_of_life",
+]

06/alchemy/transmutation/advanced.py

@@ -0,0 +1,11 @@
+from .basic import lead_to_gold
+from ..potions import healing_potion
+
+
+def philosophers_stone() -> str:
+    return f"Philosopher’s stone created using {lead_to_gold()}\
+ and {healing_potion()}"
+
+
+def elixir_of_life() -> str:
+    return f"Elixir of life: eternal youth achieved!"

06/alchemy/transmutation/basic.py

@@ -0,0 +1,9 @@
+from alchemy.elements import create_fire, create_earth
+
+
+def lead_to_gold() -> str:
+    return f"Lead transmuted to gold using {create_fire()}"
+
+
+def stone_to_gem() -> str:
+    return f"Stone transmuted to gem using {create_earth()}"

06/ft_circular_curse.py

@@ -0,0 +1,14 @@
+def ingredients_validation() -> None:
+    import alchemy.grimoire
+
+    print(
+        f'validate_ingredients("fire air"): {validate_ingredients("fire air")}'
+    )
+
+
+def main() -> None:
+    ingredients_validation()
+
+
+if __name__ == "__main__":
+    main()

06/ft_import_transmutation.py

@@ -0,0 +1,54 @@
+def full_module_import() -> None:
+    import alchemy.elements
+
+    print(f"alchemy.elements.create_fire(): {alchemy.elements.create_fire()}")
+
+
+def specific_function_import() -> None:
+    from alchemy.elements import create_water
+
+    print(f"create_water(): {create_water()}")
+
+
+def aliased_import() -> None:
+    from alchemy.potions import healing_potion as heal
+
+    print(f"heal(): {heal()}")
+
+
+def multiple_import() -> None:
+    from alchemy.elements import create_fire, create_earth
+    from alchemy.potions import strength_potion
+
+    print(f"create_earth(): {create_earth()}")
+    print(f"create_fire(): {create_fire()}")
+    print(f"strength_potion(): {strength_potion()}")
+
+
+def main() -> None:
+    print("=== Import Transmutation Mastery ===\n")
+    print("Method 1 - Full module import:")
+    try:
+        full_module_import()
+    except Exception as err:
+        print(err)
+    print("\nMethod 2 - Specific function import:")
+    try:
+        specific_function_import()
+    except Exception as err:
+        print(err)
+    print("\nMethod 3 - Aliased import:")
+    try:
+        aliased_import()
+    except Exception as err:
+        print(err)
+    print("\nMethod 4 - Multiple imports:")
+    try:
+        multiple_import()
+    except Exception as err:
+        print(err)
+    print("\nAll import transmutation methods mastered!")
+
+
+if __name__ == "__main__":
+    main()

06/ft_pathway_debate.py

@@ -0,0 +1,48 @@
+def absolute_import() -> None:
+    from alchemy.transmutation import lead_to_gold, stone_to_gem
+
+    print(f"lead_to_gold(): {lead_to_gold()}")
+    print(f"stone_to_gem(): {stone_to_gem()}")
+
+
+def relative_import() -> None:
+    from alchemy.transmutation.advanced import (
+        philosophers_stone,
+        elixir_of_life,
+    )
+
+    print(f"philosophers_stone(): {philosophers_stone()}")
+    print(f"elixir_of_life(): {elixir_of_life()}")
+
+
+def package_import() -> None:
+    import alchemy.transmutation
+
+    print(f"alchemy.transmutation.lead_to_gold():\
+ {alchemy.transmutation.lead_to_gold()}")
+    print(f"alchemy.transmutation.philosophers_stone():\
+ {alchemy.transmutation.philosophers_stone()}")
+
+
+def main() -> None:
+    print("=== Pathway Debate Mastery ===\n")
+    print("Testing Absolute Imports (from basic.py):")
+    try:
+        absolute_import()
+    except Exception as err:
+        print(err)
+    print("\nTesting Relative Imports (from advanced.py):")
+    try:
+        relative_import()
+    except Exception as err:
+        print(err)
+    print("\nTesting Package Access:")
+    try:
+        package_import()
+    except Exception as err:
+        print(err)
+    print("\nBoth pathways work! Absolute: clear, Relative: concise")
+
+
+if __name__ == "__main__":
+    main()

06/ft_sacred_scroll.py

@@ -0,0 +1,32 @@
+def main() -> None:
+    import alchemy
+
+    print("=== Sacred Scroll Mastery ===\n")
+    print("Testing direct module access:")
+    print(f"alchemy.elements.create_fire(): {alchemy.elements.create_fire()}")
+    print(
+        f"alchemy.elements.create_water(): {alchemy.elements.create_water()}"
+    )
+    print(
+        f"alchemy.elements.create_earth(): {alchemy.elements.create_earth()}"
+    )
+    print(f"alchemy.elements.create_air(): {alchemy.elements.create_air()}")
+    print("\nTesting package-level access (controlled by __init__.py):")
+    print(f"alchemy.create_fire(): {alchemy.create_fire()}")
+    print(f"alchemy.create_water(): {alchemy.create_water()}")
+
+    print("alchemy.create_earth(): ", end="")
+    try:
+        print(f"{alchemy.create_earth()}")
+    except AttributeError:
+        print("AttributeError - not exposed")
+
+    print("alchemy.create_air(): ", end="")
+    try:
+        print(f"{alchemy.create_air()}")
+    except AttributeError:
+        print("AttributeError - not exposed")
+
+
+if __name__ == "__main__":
+    main()