Add Bridge Pattern

Author: ELC

README.md

@@ -4,3 +4,4 @@
 ## Patterns
 
 - [Strategy](./patterns/behavioural/strategy/README.md)
+- [Bridge](./patterns/structural/bridge/README.md)

patterns/structural/__init__.py

@@ -0,0 +1,10 @@
+import uuid
+from io import StringIO
+
+
+BufferData = StringIO
+BufferOutput = str
+
+
+def generate_id() -> str:
+    return str(uuid.uuid4())

patterns/structural/bridge/__init__.py

@@ -0,0 +1,26 @@
+from .services import (
+    YouTubeStreamingService,
+    YouTubeStreamingServiceWithWebcam,
+    TwitchStreamingService,
+    TwitchStreamingServiceWithDSLRCamera,
+    TwitchStreamingServiceWithWebcam,
+)
+
+
+def main() -> None:
+    youtube_service = YouTubeStreamingService()
+    youtube_service.add_device(YouTubeStreamingServiceWithWebcam(quality="720p"))
+    youtube_service.fill_buffer()
+    youtube_output = youtube_service.collect_and_close_stream()
+    print(youtube_output)
+
+    twitch_service = TwitchStreamingService()
+    twitch_service.add_device(TwitchStreamingServiceWithDSLRCamera())
+    twitch_service.add_device(TwitchStreamingServiceWithWebcam(quality="720p"))
+    twitch_service.fill_buffer()
+    twitch_output = twitch_service.collect_and_close_stream()
+    print(twitch_output)
+
+
+if __name__ == "__main__":
+    main()

patterns/structural/bridge/problem/__init__.py

@@ -0,0 +1,11 @@
+from .streaming_service import StreamingService, Quality
+from .twitch_stream import (
+    TwitchStreamingService,
+    TwitchStreamingServiceWithWebcam,
+    TwitchStreamingServiceWithDSLRCamera,
+)
+from .youtube_stream import (
+    YouTubeStreamingService,
+    YouTubeStreamingServiceWithWebcam,
+    YouTubeStreamingServiceWithDSLRCamera,
+)

patterns/structural/bridge/problem/main.py

@@ -0,0 +1,42 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import List, Literal
+from io import StringIO
+
+from .. import BufferData, BufferOutput, generate_id
+
+
+Quality = Literal["360p", "480p", "720p", "1080p", "2160p"]
+
+
+@dataclass
+class StreamingService(ABC):
+    devices: List["StreamingService"] = field(default_factory=list)
+    output: BufferData = field(default_factory=StringIO)
+    reference: str = field(init=False)
+
+    def __post_init__(self):
+        self.reference = generate_id()
+
+    def add_device(self, device: "StreamingService") -> None:
+        if not isinstance(device, type(self)):
+            raise ValueError(
+                f"Only subclasses can be used - {type(device)} is not subclass of {type(self)}"
+            )
+
+        self.devices.append(device)
+
+    def retrieve_buffer_data(self) -> List[BufferData]:
+        return [device.get_buffer_data() for device in self.devices]
+
+    @abstractmethod
+    def fill_buffer(self) -> None:
+        ...
+
+    @abstractmethod
+    def collect_and_close_stream(self) -> BufferOutput:
+        ...
+
+    @abstractmethod
+    def get_buffer_data(self) -> BufferData:
+        ...

patterns/structural/bridge/problem/services/__init__.py

@@ -0,0 +1,37 @@
+from io import StringIO
+from dataclasses import dataclass
+
+from . import StreamingService, Quality
+from .. import BufferOutput, BufferData
+
+
+class TwitchStreamingService(StreamingService):
+    def fill_buffer(self) -> None:
+        buffer_data = self.retrieve_buffer_data()
+        for buffer in buffer_data:
+            buffer_content = buffer.getvalue()
+            self.output.write(
+                f"Received buffer data: {buffer_content}. Sending to Twitch stream: {self.reference}.\n"
+            )
+
+    def collect_and_close_stream(self) -> BufferOutput:
+        self.output.write(f"Closing Twitch stream with reference {self.reference}.\n")
+        collected_buffer = self.output.getvalue()
+        self.output.close()
+        return collected_buffer
+
+    def get_buffer_data(self) -> BufferData:
+        raise NotImplementedError()
+
+
+class TwitchStreamingServiceWithDSLRCamera(TwitchStreamingService):
+    def get_buffer_data(self) -> BufferData:
+        return StringIO("###DSLRDATA###")
+
+
+@dataclass
+class TwitchStreamingServiceWithWebcam(TwitchStreamingService):
+    quality: Quality = "1080p"
+
+    def get_buffer_data(self) -> BufferData:
+        return StringIO(f"###WEBCAMDATA at {self.quality}###")

patterns/structural/bridge/problem/services/streaming_service.py

@@ -0,0 +1,37 @@
+from io import StringIO
+from dataclasses import dataclass
+
+from . import StreamingService, Quality
+from .. import BufferOutput, BufferData
+
+
+class YouTubeStreamingService(StreamingService):
+    def fill_buffer(self) -> None:
+        buffer_data = self.retrieve_buffer_data()
+        for buffer in buffer_data:
+            buffer_content = buffer.getvalue()
+            self.output.write(
+                f"Received buffer data: {buffer_content}. Sending to YouTube stream: {self.reference}.\n"
+            )
+
+    def collect_and_close_stream(self) -> BufferOutput:
+        self.output.write(f"Closing YouTube stream with reference {self.reference}.\n")
+        collected_buffer = self.output.getvalue()
+        self.output.close()
+        return collected_buffer
+
+    def get_buffer_data(self) -> BufferData:
+        raise NotImplementedError()
+
+
+class YouTubeStreamingServiceWithDSLRCamera(YouTubeStreamingService):
+    def get_buffer_data(self) -> BufferData:
+        return StringIO("###DSLRDATA###")
+
+
+@dataclass
+class YouTubeStreamingServiceWithWebcam(YouTubeStreamingService):
+    quality: Quality = "1080p"
+
+    def get_buffer_data(self) -> BufferData:
+        return StringIO(f"###WEBCAMDATA at {self.quality}###")

patterns/structural/bridge/problem/services/twitch_stream.py

@@ -0,0 +1,10 @@
+import uuid
+from io import StringIO
+
+
+BufferData = StringIO
+BufferOutput = str
+
+
+def generate_id() -> str:
+    return str(uuid.uuid4())