CommonDesignPatterns.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "231453f2",
   "metadata": {},
   "source": [
    "# Common Design Patterns\n",
    "\n",
    "- learn some common design patterns and how they're implemented in Python\n",
    "- topics:\n",
    "    - the Decorator pattern\n",
    "    - the Observer pattern\n",
    "    - the Strategy pattern\n",
    "    - the Command pattern\n",
    "    - the State pattern\n",
    "    - the Singleton pattern\n",
    "    \n",
    "## The Decorator pattern\n",
    "\n",
    "- the Decorator pattern allows us to *wrap* an object that provides core functionalities with other objects\n",
    "- two primary uses:\n",
    " - enhancing the response of a component as it sends data to a second component\n",
    " - supporting multiple optional behaviors\n",
    "     - suitable alternative to multiple inheritance\n",
    "\n",
    "![Decorator Pattern UML](resources/decorator_pattern.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "56f2bf32",
   "metadata": {},
   "source": [
    "## Decorators in python\n",
    "\n",
    "- python provides wraps() function to wrap any function and decorate it\n",
    "- functools provides wraps and many built-in decorators that can be very useful\n",
    "    - e.g., least recently used (LRU)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "eb2c9839",
   "metadata": {},
   "outputs": [],
   "source": [
    "from functools import wraps\n",
    "from typing import Callable, Any\n",
    "\n",
    "def log_args(function: Callable[..., Any]) -> Callable[..., Any]:\n",
    "    @wraps(function)\n",
    "    def wrapped_function(*args: Any, **kwargs: Any) -> Any:\n",
    "        print(f\"Calling {function.__name__}(*{args}, **{kwargs})\")\n",
    "        result = function(*args, **kwargs)\n",
    "        return result\n",
    "    \n",
    "    return wrapped_function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d2f619d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test1(a: int, b: int, c:int) -> float:\n",
    "    return sum(range(a, b + 1)) / c\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "091508af",
   "metadata": {},
   "outputs": [],
   "source": [
    "test1 = log_args(test1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "03a305d6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Calling test1(*(1, 9, 2), **{})\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "22.5"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test1(1, 9, 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "dfe131e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# better option\n",
    "@log_args\n",
    "def test2(a: int, b: int, c:int) -> float:\n",
    "    return sum(range(a, b + 1)) / c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "5d2cf3a1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Calling test2(*(1, 9, 2), **{})\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "22.5"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test2(1, 9, 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "52b6f924",
   "metadata": {},
   "outputs": [],
   "source": [
    "from math import factorial\n",
    "\n",
    "def binom(n: int, k: int) -> int:\n",
    "    return factorial(n) // (factorial(k) * factorial(n-k))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "a2055780",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'6-card deals: 20,358,520'"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f\"6-card deals: {binom(52, 6):,d}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "0915bccf",
   "metadata": {},
   "outputs": [],
   "source": [
    "from functools import lru_cache\n",
    "\n",
    "@lru_cache(64)\n",
    "def binom(n: int, k: int) -> int:\n",
    "    return factorial(n) // (factorial(k) * factorial(n-k))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "7a113558",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'6-card deals: 20,358,520'"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f\"6-card deals: {binom(52, 6):,d}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "5d5badb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# class decorators\n",
    "# run the named_logger.py file to see the log being generated\n",
    "# demo.log is not generated from the Jupyter Notebook\n",
    "import logging\n",
    "import time\n",
    "\n",
    "class NamedLogger:\n",
    "    # configure logger\n",
    "    logging.basicConfig(filename=\"demo.log\",\n",
    "                format='%(asctime)s %(message)s',\n",
    "                filemode='w')\n",
    "        \n",
    "    def __init__(self, logger_name: str) -> None:\n",
    "        self.logger = logging.getLogger(logger_name)\n",
    "        self.logger.setLevel(logging.DEBUG)\n",
    "        \n",
    "    def __call__(\n",
    "           self,\n",
    "           function: Callable[..., Any]\n",
    "    ) -> Callable[..., Any]:\n",
    "        @wraps(function)\n",
    "        def wrapped_function(*args: Any, **kwargs: Any) -> Any:\n",
    "            start = time.perf_counter()\n",
    "            try:\n",
    "                result = function(*args, **kwargs)\n",
    "                μs = (time.perf_counter() - start) * 1_000_000\n",
    "                self.logger.info(\n",
    "                    f\"{function.__name__}, { μs:.1f}μs\")\n",
    "                return result\n",
    "            except Exception as ex:\n",
    "                μs = (time.perf_counter() - start) * 1_000_000\n",
    "                self.logger.error(\n",
    "                    f\"{ex}, {function.__name__}, { μs:.1f}μs\")\n",
    "                raise\n",
    "        return wrapped_function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "1a4223d8",
   "metadata": {},
   "outputs": [],
   "source": [
    "@NamedLogger(\"log4\")\n",
    "def test4(median: float, sample: float) -> float:\n",
    "    return sample - median"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "2c7eb728",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test4(4, 10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "84330272",
   "metadata": {},
   "source": [
    "## The Observer pattern\n",
    "\n",
    "- useful for state monitoring and event handling situations\n",
    "- pattern allows a given object to be monitored by an unknown and dynamic group of *observer* objects\n",
    "- the core object being observed needs to implement an interface that makes it *observable*\n",
    "- whenever a value on the core object changes, it lets all the observer objects know that a change has occured\n",
    "    - calling a method announing of the changes\n",
    "- widely used in GUIs\n",
    "\n",
    "![Observer pattern](resources/observer_pattern.png)\n",
    "\n",
    "- at a high level, the pattern is also called `Publisher-Subscriber` model\n",
    "    - Puplisher/Subject is oberserved by many objects called Subscriber \n",
    "    - any any changes in the state of the Subject/Publisher are notified to the subscribers\n",
    "- obserer pattern example using dice game: Zonk or Zilch or Ten Thousand\n",
    "- see `src/design_patterns/inventory.py` that implments observer patter for the dice game\n",
    "\n",
    "### Pros\n",
    "\n",
    "- **Open/Closed Principle**\n",
    "    - open for extension but closed for modification\n",
    "    - introducing new subscriber classes is much easier without making changes in existing publisher or observers' code\n",
    "- **Establishes Relationships**\n",
    "    - easily establishes and describes the relationships among objects during the runtime between the objects\n",
    "\n",
    "\n",
    "\n",
    "### Applications\n",
    "\n",
    "- **Multi-dependency**:\n",
    "    - when multiple objects are dependent on the state of one object\n",
    "- **Getting Notifications**\n",
    "    e.g., ocial media, RSS feeds, email subscription\n",
    "- **Reflections of Object**\n",
    "     - objects are losely coupled, the change of a state in one object must be reflected in another object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "id": "b0845c87",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Publisher:\n",
    "    def __init__(self):\n",
    "        self._observers = []\n",
    " \n",
    "    def notify(self, modifier = None):\n",
    "        \"\"\"notifiy the observers and not the modifier\"\"\"\n",
    "        for observer in self._observers:\n",
    "            if modifier != observer:\n",
    "                observer.update(self)\n",
    " \n",
    "    def attach(self, observer):\n",
    "        if observer not in self._observers:\n",
    "            self._observers.append(observer)\n",
    " \n",
    "    def detach(self, observer):\n",
    "        try:\n",
    "            self._observers.remove(observer)\n",
    "        except ValueError:\n",
    "            pass\n",
    "        \n",
    "class Data(Publisher):\n",
    "    \"\"\"monitor the object\"\"\"\n",
    "    def __init__(self, name = ''):\n",
    "        super().__init__()\n",
    "        self.name = name\n",
    "        self._data = 0\n",
    " \n",
    "    @property\n",
    "    def data(self):\n",
    "        return self._data\n",
    " \n",
    "    @data.setter\n",
    "    def data(self, value):\n",
    "        self._data = value\n",
    "        self.notify()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "id": "60c907d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create observers/subscribers\n",
    "\n",
    "class HexViewer:\n",
    "    \"\"\"updates the Hexviewer\"\"\"\n",
    "    def update(self, subject):\n",
    "        print('HexViewer: Subject {} has data 0x{:x}'.format(subject.name, subject.data))\n",
    " \n",
    "\n",
    "class OctalViewer:\n",
    "    \"\"\"updates the Octal viewer\"\"\"\n",
    " \n",
    "    def update(self, subject):\n",
    "        print(f'OctalViewer: Subject {subject.name} has data {oct(subject.data)}')\n",
    " \n",
    "\n",
    "class DecimalViewer:\n",
    "    \"\"\"updates the Decimal viewer\"\"\"\n",
    " \n",
    "    def update(self, subject):\n",
    "        print('DecimalViewer: Subject % s has data % d' % (subject.name, subject.data))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "id": "970c60ec",
   "metadata": {},
   "outputs": [],
   "source": [
    "pub1 = Data('Publisher 1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "id": "cf9b10df",
   "metadata": {},
   "outputs": [],
   "source": [
    "subs1 = DecimalViewer()\n",
    "subs2 = HexViewer()\n",
    "subs3 = OctalViewer()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "id": "600cabe8",
   "metadata": {},
   "outputs": [],
   "source": [
    "pub1.attach(subs1)\n",
    "pub1.attach(subs2)\n",
    "pub1.attach(subs3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "id": "9dac70cf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DecimalViewer: Subject Publisher 1 has data  10\n",
      "HexViewer: Subject Publisher 1 has data 0xa\n",
      "OctalViewer: Subject Publisher 1 has data 0o12\n"
     ]
    }
   ],
   "source": [
    "pub1.data = 10"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fce7fae4",
   "metadata": {},
   "source": [
    "## The Strategy pattern\n",
    "\n",
    "- see this resource: https://refactoring.guru/design-patterns/strategy\n",
    "- common demonstration of abstraction in OOP\n",
    "- pattern implements different solutions to a single problem, each in a different object\n",
    "- the core class can then choose the most appropriate implementation dynamically at runtime\n",
    "    - usually, different algorithms have different tradeoffs\n",
    "    - e.g., faster, more RAM, distributed systems, etc.\n",
    "    \n",
    "![Strategy pattern](resources/strategy_pattern.png)\n",
    "- see `src/design_patterns/image_filler.py` for example\n",
    "- may need to install `pillow` third-party library to manipulate images\n",
    "\n",
    "```python\n",
    "python image_filer.py\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "id": "646eaa68",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting pillow\n",
      "  Downloading Pillow-9.4.0-2-cp310-cp310-macosx_10_10_x86_64.whl (3.3 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m3.3/3.3 MB\u001b[0m \u001b[31m5.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m0m\n",
      "\u001b[?25hInstalling collected packages: pillow\n",
      "Successfully installed pillow-9.4.0\n"
     ]
    }
   ],
   "source": [
    "! python -m pip install pillow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "id": "0f8a6536",
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from abc import ABC, abstractmethod\n",
    "from typing import List\n",
    "\n",
    "\n",
    "class Context():\n",
    "    \"\"\"\n",
    "    The Context defines the interface of interest to clients.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, strategy: Strategy) -> None:\n",
    "        \"\"\"\n",
    "        Usually, the Context accepts a strategy through the constructor, but\n",
    "        also provides a setter to change it at runtime.\n",
    "        \"\"\"\n",
    "\n",
    "        self._strategy = strategy\n",
    "\n",
    "    @property\n",
    "    def strategy(self) -> Strategy:\n",
    "        \"\"\"\n",
    "        The Context maintains a reference to one of the Strategy objects. The\n",
    "        Context does not know the concrete class of a strategy. It should work\n",
    "        with all strategies via the Strategy interface.\n",
    "        \"\"\"\n",
    "\n",
    "        return self._strategy\n",
    "\n",
    "    @strategy.setter\n",
    "    def strategy(self, strategy: Strategy) -> None:\n",
    "        \"\"\"\n",
    "        Usually, the Context allows replacing a Strategy object at runtime.\n",
    "        \"\"\"\n",
    "\n",
    "        self._strategy = strategy\n",
    "\n",
    "    def do_some_business_logic(self) -> None:\n",
    "        \"\"\"\n",
    "        The Context delegates some work to the Strategy object instead of\n",
    "        implementing multiple versions of the algorithm on its own.\n",
    "        \"\"\"\n",
    "\n",
    "        # ...\n",
    "\n",
    "        print(\"Context: Sorting data using the strategy (not sure how it'll do it)\")\n",
    "        result = self._strategy.do_algorithm([\"a\", \"b\", \"c\", \"d\", \"e\"])\n",
    "        print(\",\".join(result))\n",
    "\n",
    "        # ...\n",
    "\n",
    "\n",
    "class Strategy(ABC):\n",
    "    \"\"\"\n",
    "    The Strategy interface declares operations common to all supported versions\n",
    "    of some algorithm.\n",
    "\n",
    "    The Context uses this interface to call the algorithm defined by Concrete\n",
    "    Strategies.\n",
    "    \"\"\"\n",
    "\n",
    "    @abstractmethod\n",
    "    def do_algorithm(self, data: List):\n",
    "        pass\n",
    "\n",
    "\n",
    "\"\"\"\n",
    "Concrete Strategies implement the algorithm while following the base Strategy\n",
    "interface. The interface makes them interchangeable in the Context.\n",
    "\"\"\"\n",
    "\n",
    "\n",
    "class ConcreteStrategyA(Strategy):\n",
    "    def do_algorithm(self, data: List) -> List:\n",
    "        return sorted(data)\n",
    "\n",
    "\n",
    "class ConcreteStrategyB(Strategy):\n",
    "    def do_algorithm(self, data: List) -> List:\n",
    "        return reversed(sorted(data))\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "id": "51e94a76",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Client: Strategy is set to normal sorting.\n",
      "Context: Sorting data using the strategy (not sure how it'll do it)\n",
      "a,b,c,d,e\n",
      "\n",
      "Client: Strategy is set to reverse sorting.\n",
      "Context: Sorting data using the strategy (not sure how it'll do it)\n",
      "e,d,c,b,a\n"
     ]
    }
   ],
   "source": [
    "context = Context(ConcreteStrategyA())\n",
    "print(\"Client: Strategy is set to normal sorting.\")\n",
    "context.do_some_business_logic()\n",
    "print()\n",
    "\n",
    "print(\"Client: Strategy is set to reverse sorting.\")\n",
    "context.strategy = ConcreteStrategyB()\n",
    "context.do_some_business_logic()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aec2c5fb",
   "metadata": {},
   "source": [
    "## The Command pattern\n",
    "\n",
    "- the UML diagram of command pattern looks similar to strategy pattern\n",
    "- this pattern generally involves a hierarhy of classes that each do something\n",
    "- a Core class can create a command (or a sequence of commands) to carry out actions\n",
    "- the pattern can be viewed as a collaboration between a \"passive\" observer and the more \"active\" Commander\n",
    "- A commander will be active making state changes in other objects\n",
    "- An observer is notified that somehing changed\n",
    "    - behaves somewhat like observer pattern\n",
    "    \n",
    "![Command pattern](resources/command_pattern.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "id": "e7dccd47",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Conceptual example from: https://refactoring.guru/design-patterns/command/python/example\n",
    "from __future__ import annotations\n",
    "from abc import ABC, abstractmethod\n",
    "\n",
    "\n",
    "class Command(ABC):\n",
    "    \"\"\"\n",
    "    The Command interface declares a method for executing a command.\n",
    "    \"\"\"\n",
    "\n",
    "    @abstractmethod\n",
    "    def execute(self) -> None:\n",
    "        pass\n",
    "    \n",
    "    \n",
    "class SimpleCommand(Command):\n",
    "    \"\"\"\n",
    "    Some commands can implement simple operations on their own.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, payload: str) -> None:\n",
    "        self._payload = payload\n",
    "\n",
    "    def execute(self) -> None:\n",
    "        print(f\"SimpleCommand: See, I can do simple things like printing\"\n",
    "              f\"({self._payload})\")\n",
    "        \n",
    "\n",
    "class ComplexCommand(Command):\n",
    "    \"\"\"\n",
    "    However, some commands can delegate more complex operations to other\n",
    "    objects, called \"receivers.\"\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, receiver: Receiver, a: str, b: str) -> None:\n",
    "        \"\"\"\n",
    "        Complex commands can accept one or several receiver objects along with\n",
    "        any context data via the constructor.\n",
    "        \"\"\"\n",
    "\n",
    "        self._receiver = receiver\n",
    "        self._a = a\n",
    "        self._b = b\n",
    "\n",
    "    def execute(self) -> None:\n",
    "        \"\"\"\n",
    "        Commands can delegate to any methods of a receiver.\n",
    "        \"\"\"\n",
    "\n",
    "        print(\"ComplexCommand: Complex stuff should be done by a receiver object\", end=\"\")\n",
    "        self._receiver.do_something(self._a)\n",
    "        self._receiver.do_something_else(self._b)\n",
    "        \n",
    "\n",
    "class Receiver:\n",
    "    \"\"\"\n",
    "    The Receiver classes contain some important business logic. They know how to\n",
    "    perform all kinds of operations, associated with carrying out a request. In\n",
    "    fact, any class may serve as a Receiver.\n",
    "    \"\"\"\n",
    "\n",
    "    def do_something(self, a: str) -> None:\n",
    "        print(f\"\\nReceiver: Working on ({a}.)\", end=\"\")\n",
    "\n",
    "    def do_something_else(self, b: str) -> None:\n",
    "        print(f\"\\nReceiver: Also working on ({b}.)\", end=\"\")\n",
    "        \n",
    "    \n",
    "class Invoker:\n",
    "    \"\"\"\n",
    "    The Invoker is associated with one or several commands. It sends a request\n",
    "    to the command.\n",
    "    \"\"\"\n",
    "\n",
    "    _on_start = None\n",
    "    _on_finish = None\n",
    "\n",
    "    \"\"\"\n",
    "    Initialize commands.\n",
    "    \"\"\"\n",
    "\n",
    "    def set_on_start(self, command: Command):\n",
    "        self._on_start = command\n",
    "\n",
    "    def set_on_finish(self, command: Command):\n",
    "        self._on_finish = command\n",
    "\n",
    "    def do_something_important(self) -> None:\n",
    "        \"\"\"\n",
    "        The Invoker does not depend on concrete command or receiver classes. The\n",
    "        Invoker passes a request to a receiver indirectly, by executing a\n",
    "        command.\n",
    "        \"\"\"\n",
    "\n",
    "        print(\"Invoker: Does anybody want something done before I begin?\")\n",
    "        if isinstance(self._on_start, Command):\n",
    "            self._on_start.execute()\n",
    "\n",
    "        print(\"Invoker: ...doing something really important...\")\n",
    "\n",
    "        print(\"Invoker: Does anybody want something done after I finish?\")\n",
    "        if isinstance(self._on_finish, Command):\n",
    "            self._on_finish.execute()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "id": "eb74083e",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "The client code can parameterize an invoker with any commands.\n",
    "\"\"\"\n",
    "\n",
    "invoker = Invoker()\n",
    "invoker.set_on_start(SimpleCommand(\"Say Hi!\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "id": "84f930e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "receiver = Receiver()\n",
    "invoker.set_on_finish(ComplexCommand(\n",
    "    receiver, \"Send email\", \"Save report\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "id": "e07cf19c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Invoker: Does anybody want something done before I begin?\n",
      "SimpleCommand: See, I can do simple things like printing(Say Hi!)\n",
      "Invoker: ...doing something really important...\n",
      "Invoker: Does anybody want something done after I finish?\n",
      "ComplexCommand: Complex stuff should be done by a receiver object\n",
      "Receiver: Working on (Send email.)\n",
      "Receiver: Also working on (Save report.)"
     ]
    }
   ],
   "source": [
    "invoker.do_something_important()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7d65d04a",
   "metadata": {},
   "source": [
    "## The State pattern\n",
    "\n",
    "- lets an object alter its behavior when its internal state changes\n",
    "- based on the concept of **Finite State Machine**\n",
    "- see this artcle: https://refactoring.guru/design-patterns/state\n",
    "- the goal is to represent state transition systems:\n",
    "    - systems where an object's behavior is constrained by the state it's in\n",
    "    - there are narrowly defined transition to other states\n",
    "- there's a manager/context class that provides an interface for switching states\n",
    "    - this class contains a pointer to the current state\n",
    "- each state know what other states will be allowed to transition to\n",
    "\n",
    "![State pattern](resources/state_pattern.png)\n",
    "- applications:\n",
    "    - parsing text such as JSON, XML, HTML, YAML, etc.\n",
    "    - parsing and inerpreting/compiling programming languages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "id": "c7e13bcd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "from abc import ABC, abstractmethod\n",
    "\n",
    "\n",
    "class Context:\n",
    "    \"\"\"\n",
    "    The Context defines the interface of interest to clients. \n",
    "    It also maintains a reference to an instance of a State \n",
    "    subclass, which represents the current state of the Context.\n",
    "    \"\"\"\n",
    "    _state = None\n",
    "    \n",
    "    \"\"\"\n",
    "    A reference to the current state of the Context.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, state: State) -> None:\n",
    "        self.transition_to(state)\n",
    "\n",
    "    def transition_to(self, state: State):\n",
    "        \"\"\"\n",
    "        The Context allows changing the State object at runtime.\n",
    "        \"\"\"\n",
    "\n",
    "        print(f\"Context: Transition to {type(state).__name__}\")\n",
    "        self._state = state\n",
    "        self._state.context = self\n",
    "\n",
    "    \"\"\"\n",
    "    The Context delegates part of its behavior \n",
    "    to the current State object.\n",
    "    \"\"\"\n",
    "\n",
    "    def request1(self):\n",
    "        self._state.handle1()\n",
    "\n",
    "    def request2(self):\n",
    "        self._state.handle2()\n",
    "\n",
    "\n",
    "class State(ABC):\n",
    "    \"\"\"\n",
    "    The base State class declares methods that all Concrete State should\n",
    "    implement and also provides a backreference to the Context object,\n",
    "    associated with the State. This backreference can be used by States to\n",
    "    transition the Context to another State.\n",
    "    \"\"\"\n",
    "\n",
    "    @property\n",
    "    def context(self) -> Context:\n",
    "        return self._context\n",
    "\n",
    "    @context.setter\n",
    "    def context(self, context: Context) -> None:\n",
    "        self._context = context\n",
    "\n",
    "    @abstractmethod\n",
    "    def handle1(self) -> None:\n",
    "        pass\n",
    "\n",
    "    @abstractmethod\n",
    "    def handle2(self) -> None:\n",
    "        pass\n",
    "\n",
    "\n",
    "\"\"\"\n",
    "Concrete States implement various behaviors, associated with a state of the\n",
    "Context.\n",
    "\"\"\"\n",
    "\n",
    "\n",
    "class ConcreteStateA(State):\n",
    "    def handle1(self) -> None:\n",
    "        print(\"ConcreteStateA handles request1.\")\n",
    "        print(\"ConcreteStateA wants to change the state of the context.\")\n",
    "        self.context.transition_to(ConcreteStateB())\n",
    "\n",
    "    def handle2(self) -> None:\n",
    "        print(\"ConcreteStateA handles request2.\")\n",
    "\n",
    "\n",
    "class ConcreteStateB(State):\n",
    "    def handle1(self) -> None:\n",
    "        print(\"ConcreteStateB handles request1.\")\n",
    "\n",
    "    def handle2(self) -> None:\n",
    "        print(\"ConcreteStateB handles request2.\")\n",
    "        print(\"ConcreteStateB wants to change the state of the context.\")\n",
    "        self.context.transition_to(ConcreteStateA())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "id": "bc0db759",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Context: Transition to ConcreteStateA\n",
      "ConcreteStateA handles request1.\n",
      "ConcreteStateA wants to change the state of the context.\n",
      "Context: Transition to ConcreteStateB\n",
      "ConcreteStateB handles request2.\n",
      "ConcreteStateB wants to change the state of the context.\n",
      "Context: Transition to ConcreteStateA\n"
     ]
    }
   ],
   "source": [
    "context = Context(ConcreteStateA())\n",
    "context.request1()\n",
    "context.request2()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6044061a",
   "metadata": {},
   "source": [
    "### IoT Example\n",
    "\n",
    "- parging data from GPS statement\n",
    "- GPS uses NMEA 0183 language from the National Marine Electronics Association\n",
    "- a typical message looks like:\n",
    "```\n",
    " $GPGLL,3723.2475,N,12158.3416,W,161229.487,A,A*41\n",
    " ```\n",
    "\n",
    "| value | meaning |\n",
    "| :--- | :--- |\n",
    "| \\$ | Starts the sentence |\n",
    "| GPGLL | The \"talker,\" GP, and the type of message, GLL |\n",
    "| 3723.2475 | Latitude, 37°23.2475 |\n",
    "| N | North of the equator |\n",
    "| 12158.3416 | Longitude, 121°58.3416 |\n",
    "| W | West of the 0° meridian |\n",
    "| 161229.487 | The timestamp in UTC: 16:12:29.487 |\n",
    "| A | Status, A=valid, V=not valid |\n",
    "| A | Mode, A=Autonomous, D=DGPS, E=DR |\n",
    "| * | Ends the sentence, starts the checksum |\n",
    "| 41 | Hexadecimal checksum of the text (excluding the \\$ and \\* characters) |\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "id": "2c786577",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\"\"\"\r\n",
      "Python 3 Object-Oriented Programming\r\n",
      "\r\n",
      "Chapter 11. Common Design Patterns\r\n",
      "\"\"\"\r\n",
      "from __future__ import annotations\r\n",
      "from typing import Optional, Iterable, Iterator, cast\r\n",
      "\r\n",
      "\r\n",
      "class NMEA_State:\r\n",
      "    def __init__(self, message: \"Message\") -> None:\r\n",
      "        self.message = message\r\n",
      "\r\n",
      "    def feed_byte(self, input: int) -> \"NMEA_State\":\r\n",
      "        return self\r\n",
      "\r\n",
      "    def valid(self) -> bool:\r\n",
      "        return False\r\n",
      "\r\n",
      "    def __repr__(self) -> str:\r\n",
      "        return f\"{self.__class__.__name__}({self.message})\"\r\n",
      "\r\n",
      "\r\n",
      "class Waiting(NMEA_State):\r\n",
      "    def feed_byte(self, input: int) -> NMEA_State:\r\n",
      "        if input == ord(b\"$\"):\r\n",
      "            return Header(self.message)\r\n",
      "        return self\r\n",
      "\r\n",
      "\r\n",
      "class Header(NMEA_State):\r\n",
      "    def __init__(self, message: \"Message\") -> None:\r\n",
      "        self.message = message\r\n",
      "        self.message.reset()\r\n",
      "\r\n",
      "    def feed_byte(self, input: int) -> NMEA_State:\r\n",
      "        if input == ord(b\"$\"):\r\n",
      "            # Reset any accumulated bytes\r\n",
      "            return Header(self.message)\r\n",
      "        size = self.message.body_append(input)\r\n",
      "        if size == 5:\r\n",
      "            return Body(self.message)\r\n",
      "        return self\r\n",
      "\r\n",
      "\r\n",
      "class Body(NMEA_State):\r\n",
      "    def feed_byte(self, input: int) -> NMEA_State:\r\n",
      "        if input == ord(b\"$\"):\r\n",
      "            return Header(self.message)\r\n",
      "        if input == ord(b\"*\"):\r\n",
      "            return Checksum(self.message)\r\n",
      "        self.message.body_append(input)\r\n",
      "        return self\r\n",
      "\r\n",
      "\r\n",
      "class Checksum(NMEA_State):\r\n",
      "    def feed_byte(self, input: int) -> NMEA_State:\r\n",
      "        if input == ord(b\"$\"):\r\n",
      "            return Header(self.message)\r\n",
      "        if input in {ord(b\"\\n\"), ord(b\"\\r\")}:\r\n",
      "            # Incomplete checksum... Will be invalid.\r\n",
      "            return End(self.message)\r\n",
      "        size = self.message.checksum_append(input)\r\n",
      "        if size == 2:\r\n",
      "            return End(self.message)\r\n",
      "        return self\r\n",
      "\r\n",
      "\r\n",
      "class End(NMEA_State):\r\n",
      "    def feed_byte(self, input: int) -> NMEA_State:\r\n",
      "        if input == ord(b\"$\"):\r\n",
      "            return Header(self.message)\r\n",
      "        elif input not in {ord(b\"\\n\"), ord(b\"\\r\")}:\r\n",
      "            return Waiting(self.message)\r\n",
      "        return self\r\n",
      "\r\n",
      "    def valid(self) -> bool:\r\n",
      "        return self.message.valid\r\n",
      "\r\n",
      "\r\n",
      "class Message:\r\n",
      "    def __init__(self) -> None:\r\n",
      "        self.body = bytearray(80)\r\n",
      "        self.checksum_source = bytearray(2)\r\n",
      "        self.body_len = 0\r\n",
      "        self.checksum_len = 0\r\n",
      "        self.checksum_computed = 0\r\n",
      "\r\n",
      "    def reset(self) -> None:\r\n",
      "        self.body_len = 0\r\n",
      "        self.checksum_len = 0\r\n",
      "        self.checksum_computed = 0\r\n",
      "\r\n",
      "    def body_append(self, input: int) -> int:\r\n",
      "        self.body[self.body_len] = input\r\n",
      "        self.body_len += 1\r\n",
      "        self.checksum_computed ^= input\r\n",
      "        return self.body_len\r\n",
      "\r\n",
      "    def checksum_append(self, input: int) -> int:\r\n",
      "        self.checksum_source[self.checksum_len] = input\r\n",
      "        self.checksum_len += 1\r\n",
      "        return self.checksum_len\r\n",
      "\r\n",
      "    @property\r\n",
      "    def valid(self) -> bool:\r\n",
      "        return (\r\n",
      "            self.checksum_len == 2\r\n",
      "            and int(self.checksum_source, 16) == self.checksum_computed\r\n",
      "        )\r\n",
      "\r\n",
      "    def header(self) -> bytes:\r\n",
      "        return bytes(self.body[:5])\r\n",
      "\r\n",
      "    def fields(self) -> list[bytes]:\r\n",
      "        return bytes(self.body[: self.body_len]).split(b\",\")\r\n",
      "\r\n",
      "    def __repr__(self) -> str:\r\n",
      "        body = self.body[: self.body_len]\r\n",
      "        checksum = self.checksum_source[: self.checksum_len]\r\n",
      "        return f\"Message({body}, {checksum}, computed={self.checksum_computed:02x})\"\r\n",
      "\r\n",
      "    def message(self) -> bytes:\r\n",
      "        return (\r\n",
      "            b\"$\"\r\n",
      "            + bytes(self.body[: self.body_len])\r\n",
      "            + b\"*\"\r\n",
      "            + bytes(self.checksum_source[: self.checksum_len])\r\n",
      "        )\r\n",
      "\r\n",
      "\r\n",
      "class Reader:\r\n",
      "    def __init__(self) -> None:\r\n",
      "        self.buffer = Message()\r\n",
      "        self.state: NMEA_State = Waiting(self.buffer)\r\n",
      "\r\n",
      "    def read(self, source: Iterable[bytes]) -> Iterator[Message]:\r\n",
      "        for byte in source:\r\n",
      "            self.state = self.state.feed_byte(cast(int, byte))\r\n",
      "            if self.buffer.valid:\r",
      "\r\n",
      "                yield self.buffer\r\n",
      "                self.buffer = Message()\r\n",
      "                self.state = Waiting(self.buffer)\r\n",
      "\r\n",
      "\r\n",
      "test_reader = \"\"\"\r\n",
      ">>> message = b'''\r\n",
      "... $GPGGA,161229.487,3723.2475,N,12158.3416,W,1,07,1.0,9.0,M,,,,0000*18\r\n",
      "... $GPGLL,3723.2475,N,12158.3416,W,161229.487,A,A*41\r\n",
      "... '''\r\n",
      ">>> rdr = Reader()\r\n",
      ">>> result = list(rdr.read(message))\r\n",
      ">>> result\r\n",
      "[Message(bytearray(b'GPGGA,161229.487,3723.2475,N,12158.3416,W,1,07,1.0,9.0,M,,,,0000'), bytearray(b'18'), computed=18), Message(bytearray(b'GPGLL,3723.2475,N,12158.3416,W,161229.487,A,A'), bytearray(b'41'), computed=41)]\r\n",
      ">>> result[0].message()\r\n",
      "b'$GPGGA,161229.487,3723.2475,N,12158.3416,W,1,07,1.0,9.0,M,,,,0000*18'\r\n",
      ">>> result[1].message()\r\n",
      "b'$GPGLL,3723.2475,N,12158.3416,W,161229.487,A,A*41'\r\n",
      ">>> result[0].fields()\r\n",
      "[b'GPGGA', b'161229.487', b'3723.2475', b'N', b'12158.3416', b'W', b'1', b'07', b'1.0', b'9.0', b'M', b'', b'', b'', b'0000']\r\n",
      ">>> result[1].fields()\r\n",
      "[b'GPGLL', b'3723.2475', b'N', b'12158.3416', b'W', b'161229.487', b'A', b'A']\r\n",
      "\r\n",
      "\"\"\"\r\n",
      "\r\n",
      "\r\n",
      "__test__ = {name: case for name, case in globals().items() if name.startswith(\"test_\")}\r\n"
     ]
    }
   ],
   "source": [
    "# let's see the provided demo in nmesa_state.py module\n",
    "! cat src/design_patterns/nmea_states.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "id": "dfb16c34",
   "metadata": {},
   "outputs": [],
   "source": [
    "from src.design_patterns.nmea_states import Reader"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "id": "216b77e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "message = b'''\n",
    "$GPGGA,161229.487,3723.2475,N,12158.3416,W,1,07,1.0,9.0,M,,,,0000*18\n",
    "$GPGLL,3723.2475,N,12158.3416,W,161229.487,A,A*41\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "id": "42785b0c",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdr = Reader()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "id": "c433e6ca",
   "metadata": {},
   "outputs": [],
   "source": [
    "result = list(rdr.read(message))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "id": "06a73576",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Message(bytearray(b'GPGGA,161229.487,3723.2475,N,12158.3416,W,1,07,1.0,9.0,M,,,,0000'), bytearray(b'18'), computed=18),\n",
       " Message(bytearray(b'GPGLL,3723.2475,N,12158.3416,W,161229.487,A,A'), bytearray(b'41'), computed=41)]"
      ]
     },
     "execution_count": 109,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88c5fa34",
   "metadata": {},
   "source": [
    "## The Single pattern\n",
    "\n",
    "- may also be called anti-pattern \n",
    "- more common in more strick OOP languages such as Java than in scripting languages such as Python\n",
    "- the basic idea is to allow excatly one instance of certain objects in the program\n",
    "\n",
    "![Single pattern](resources/singleton_pattern.png)\n",
    "\n",
    "- singletons are normally enforced by making the constructor private (so no one can create additional instances of it)\n",
    "- then providing a static method to retrieve the single instance. \n",
    "    - this method creates a new instance the first time it is called, and then returns that same instance for all subsequent calls\n",
    "    \n",
    "- since python doesn't have private access specifier, we've to be a little creative\n",
    "- can use `__new__()` method and class variable to ensure that only one instance is created"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "57c525e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "class OneOnly:\n",
    "    _singleton = None\n",
    "    def __new__(cls, *args, **kwargs):\n",
    "        if not cls._singleton:\n",
    "            cls._singleton = super().__new__(cls, *args, **kwargs)\n",
    "        return cls._singleton\n",
    "    \n",
    "    def business_operation(self):\n",
    "        print('perform some business opearation')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "8b4c8be8",
   "metadata": {},
   "outputs": [],
   "source": [
    "s1 = OneOnly()\n",
    "s2 = OneOnly()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "151266d1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "s1 == s2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "fce54ad5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "id(s1) == id(s2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "9a1107c4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "perform some business opearation\n"
     ]
    }
   ],
   "source": [
    "s1.business_operation()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "a3eacfbb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<__main__.OneOnly at 0x7fb4b9363640>"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "s1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "611a4d3f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<__main__.OneOnly at 0x7fb4b9363640>"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "s2"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "395e9fc5",
   "metadata": {},
   "source": [
    "### Python singleton\n",
    "\n",
    "- Python provides two built-in Singleton patterns we can leverage -- - rather than invent something hard to read, there are two choices:\n",
    "\n",
    "1. Python *module*:\n",
    "    - One `import` will create a module\n",
    "    - all other attemtpts to import the module return the one-and-only singleton instance of the module\n",
    "\n",
    "2. Python *class* definition:\n",
    "    - a python class can only be created once in a given namespace\n",
    "    - consider using a class with class-level attributes as a singleton object\n",
    "    - use `@staticmethod` decorator to not have to use instance variable `self`\n",
    "    \n",
    "- see https://github.com/rambasnet/Kattis-Demos-Testing/tree/main/egypt/python3/OOP/main_singleton.py solution that uses static methods and class level variables to use single pattern"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e12acc78",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "- the Decorator pattern is often implemented using Python's more generic decorator syntax\n",
    "    - @wraps\n",
    "    - useful pattern to extend functionality of a function or class without modifiying the code being decorated\n",
    "    \n",
    "- the Observer pattern is useful way to decouple events from actions taken on those events\n",
    "- the Strategy pattern allows different algogithms to be chosen to accomplish the same task\n",
    "- the Command pattern helps us design active classes that share a common interface but carry out distinct actions\n",
    "- the Single pattern popular in some statically typed language is almost always an anti-pattern in Python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ccc4ef74",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}