Update app.py

app.py

@@ -1,134 +1,318 @@
-import time
-from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
+"""
+Interactive Reciprocal Tariffs Calculator
+Original Source: Yi-Cheng Wu (https://github.com/jrycw/tech/blob/master/posts/clone-reciprocal-tariffs-table/20250403.qmd)
+"""
 
-import numpy as np
+# =============================================================================
+# Imports
+# =============================================================================
+import random
+
+import polars as pl
+from great_tables import GT, google_font, html, loc, style, vals
 import panel as pn
 import param
-from asyncio import wrap_future
-
-class ProgressExtMod(pn.viewable.Viewer):
-    """A custom component for easy progress reporting"""
-
-    completed = param.Integer(default=0)
-    bar_color = param.String(default="info")
-    num_tasks = param.Integer(default=100, bounds=(1, None))
-
-    # @param.depends('completed', 'num_tasks')
-    @property
-    def value(self) -> int:
-        """Returns the progress value
-
-        Returns:
-            int: The progress value
-        """
-        return int(100 * (self.completed / self.num_tasks))
-
-    def reset(self):
-        """Resets the value and message"""
-        # Please note the order matters as the Widgets updates two times. One for each change
-        self.completed = 0
-
-    def __panel__(self):
-        return self.view
-
-    @param.depends("completed", "bar_color")
-    def view(self):
-        """View the widget
-        Returns:
-            pn.viewable.Viewable: Add this to your app to see the progress reported
-        """
-        if self.value:
-            return pn.widgets.Progress(
-                active=True, value=self.value, align="center", sizing_mode="stretch_width"
-            )
-        return None
-
-    @contextmanager
-    def increment(self):
-        """Increments the value
-        
-        Can be used as context manager or decorator
-        
-        Yields:
-            None: Nothing is yielded
-        """
-        self.completed += 1
-        yield
-        if self.completed == self.num_tasks:
-            self.reset()
-
-executor = ThreadPoolExecutor(max_workers=2)  # pylint: disable=consider-using-with
-progress = ProgressExtMod()
-
-
-class AsyncComponent(pn.viewable.Viewer):
-    """A component that demonstrates how to run a Blocking Background task asynchronously
-    in Panel"""
-
-    select = param.Selector(objects=range(10))
-    slider = param.Number(2, bounds=(0, 10))
+from panel.custom import JSComponent
+
+# Initialize Panel extension with bootstrap design
+pn.extension(design="bootstrap")
+
+
+# =============================================================================
+# Constants & Global Variables
+# =============================================================================
+# Color palette
+DARK_NAVY_BLUE = "#0B162A"  # background
+LIGHT_BLUE = "#B5D3E7"     # even row background
+WHITE = "#FFFFFF"          # odd row background
+YELLOW = "#F6D588"         # reciprocal tariffs background
+GOLD = "#FFF8DE"           # logo background
+
+# Logo image (first element of the returned tuple)
+logo = vals.fmt_image("logo.png", height=150)[0]
+
+# Base tariffs imposed data
+TARIFS_IMPOSED = {
+    "country": [
+        "China", "European Union", "Vietnam", "Taiwan", "Japan", "India",
+        "South Korea", "Thailand", "Switzerland", "Indonesia", "Malaysia",
+        "Cambodia", "United Kingdom", "South Africa", "Brazil", "Bangladesh",
+        "Singapore", "Israel", "Philippines", "Chile", "Australia", "Pakistan",
+        "Turkey", "Sri Lanka", "Colombia",
+    ],
+    "tariffs_charged": [
+        "67%", "39%", "90%", "64%", "46%", "52%", "50%", "72%", "61%", "64%",
+        "47%", "97%", "10%", "60%", "10%", "74%", "10%", "33%", "34%", "10%",
+        "10%", "58%", "10%", "88%", "10%",
+    ],
+    "reciprocal_tariffs": [
+        "34%", "20%", "46%", "32%", "24%", "26%", "25%", "36%", "31%", "32%",
+        "24%", "49%", "10%", "30%", "10%", "37%", "10%", "17%", "17%", "10%",
+        "10%", "29%", "10%", "44%", "10%",
+    ],
+}
+
+
+# =============================================================================
+# Helper Functions for Tariffs Calculations
+# =============================================================================
+def scale_tariffs(tarifs=TARIFS_IMPOSED, factor: float = 0.5) -> dict:
+    """
+    Create a new tariffs dictionary where the 'reciprocal_tariffs' are
+    scaled based on the factor applied to the 'tariffs_charged'.
+    
+    The 'tariffs_charged' column remains unchanged.
+    """
+    new_reciprocal = []
+    for tariff in TARIFS_IMPOSED["tariffs_charged"]:
+        # Convert percentage string to integer, scale it, and round to nearest integer.
+        num = int(tariff.strip('%'))
+        scaled_value = round(num * factor)
+        new_reciprocal.append(f"{scaled_value}%")
+    
+    return {
+        "country": TARIFS_IMPOSED["country"],
+        "tariffs_charged": TARIFS_IMPOSED["tariffs_charged"],
+        "reciprocal_tariffs": new_reciprocal,
+    }
+
+
+def randomize_tariffs(scale: float = 0.5) -> dict:
+    """
+    Create a new tariffs dictionary with randomized 'tariffs_charged'
+    values (between 10% and 100%) and corresponding 'reciprocal_tariffs'
+    values computed as the scaled (rounded) half.
+    """
+    new_charged = []
+    new_reciprocal = []
+    
+    for _ in TARIFS_IMPOSED["country"]:
+        rand_value = random.randint(10, 100)
+        new_charged.append(f"{rand_value}%")
+        reciprocal_value = round(rand_value * scale)
+        new_reciprocal.append(f"{reciprocal_value}%")
     
-    run_blocking_task = param.Event(label="RUN")
-    result = param.Number(0)
-    view = param.Parameter()
-
-    def __init__(self, **params):
-        super().__init__(**params)
-
-        self._layout = pn.Column(
-            pn.pane.Markdown("## Blocking Task Running in Background"),
-            pn.Param(
-                self,
-                parameters=["run_blocking_task", "result"],
-                widgets={"result": {"disabled": True}, "run_blocking_task": {"button_type": "primary"}},
-                show_name=False,
+    return {
+        "country": TARIFS_IMPOSED["country"],
+        "tariffs_charged": new_charged,
+        "reciprocal_tariffs": new_reciprocal,
+    }
+
+
+# =============================================================================
+# HTML & Style Helpers
+# =============================================================================
+def change_border_radius(content: str, border_radius: int, background_color1: str, background_color2: str) -> str:
+    """
+    Wrap the content with nested divs that have the specified border radius and background colors.
+    """
+    return f"""
+    <div style="background: {background_color1}; border-radius: {border_radius}px; padding: 2px">
+      <div style="background: {background_color2}; border-radius: {border_radius}px; padding: 2px">
+        {content}
+      </div>
+    </div>
+    """
+
+
+def change_border_radius_expr(
+    cols: pl.Expr,
+    return_dtype: pl.DataType,
+    border_radius: int,
+    background_color1: str,
+    background_color2: str,
+) -> pl.Expr:
+    """
+    Apply the change_border_radius function to each element in a Polars column.
+    """
+    return cols.map_elements(
+        lambda x: change_border_radius(x, border_radius, background_color1, background_color2),
+        return_dtype=return_dtype,
+    )
+
+
+# =============================================================================
+# Table Creation Function
+# =============================================================================
+def create_tarifs_table(tarifs: dict = TARIFS_IMPOSED) -> GT:
+    """
+    Create and style the tariffs table using Polars and Great Tables (GT).
+    The table is styled with alternating row colors and custom fonts.
+    """
+    # Build the Polars DataFrame with an index (used to alternate row styles)
+    df = (
+        pl.DataFrame(tarifs)
+        .with_row_index("mod")
+        .with_columns(pl.col("mod").mod(2))
+        .with_columns(*[pl.lit("").alias(str(i)) for i in range(4)])
+        .with_columns(
+            # Apply border radius styling for "country" and "tariffs_charged" based on the mod value
+            pl.when(pl.col("mod").eq(0))
+            .then(
+                change_border_radius_expr(
+                    pl.col("country", "tariffs_charged"),
+                    pl.String,
+                    5,
+                    DARK_NAVY_BLUE,
+                    LIGHT_BLUE,
+                )
+            )
+            .otherwise(
+                change_border_radius_expr(
+                    pl.col("country", "tariffs_charged"),
+                    pl.String,
+                    5,
+                    DARK_NAVY_BLUE,
+                    WHITE,
+                )
             ),
-            progress,
-            pn.pane.Markdown("## Other, Non-Blocked Tasks"),
-            pn.Param(
-                self,
-                parameters=["select", "slider"],
-                widgets={"text": {"disabled": True}},
-                show_name=False,
+            # Style the "reciprocal_tariffs" column
+            change_border_radius_expr(
+                pl.col("reciprocal_tariffs"), pl.String, 5, DARK_NAVY_BLUE, YELLOW
             ),
-            self.text
         )
+        .select(["0", "country", "1", "tariffs_charged", "2", "reciprocal_tariffs", "3"])
+        # Add an extra empty row at the end of the table
+        .pipe(lambda df_: pl.concat([df_, pl.DataFrame({col: "" for col in df_.columns})], how="vertical"))
+    )
 
-    def __panel__(self):
-        return self._layout
-
-    @param.depends("slider", "select")
-    def text(self):
-        if self.select:
-            select = self.select
-        else:
-            select = 0
-        return f"{select} + {self.slider} = {select + self.slider}"
-
-    @pn.depends("run_blocking_task", watch=True)
-    async def _run_blocking_tasks(self, num_tasks=10):
-        """Runs background tasks num_tasks times"""
-        num_tasks = 20
-        progress.num_tasks = num_tasks
-        for _ in range(num_tasks):
-            future = executor.submit(self._run_blocking_task)
-            result = await wrap_future(future)
-            self._update(result)
-
-    @progress.increment()
-    def _update(self, number):
-        self.result += number
-
-    @staticmethod
-    def _run_blocking_task():
-        time.sleep(np.random.randint(1, 2))
-        return 5
-
-if pn.state.served:
-    pn.extension()
+    # Create a GT table and apply styling options
+    table = (
+        GT(df)
+        .cols_align("center", columns=["tariffs_charged", "reciprocal_tariffs"])
+        .cols_label({
+            "country": html(
+                f"""<div>
+                    {logo}&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Reciprocal Tariffs&nbsp;&nbsp;&nbsp;<br>
+                    Country
+                    </div>"""
+            ),
+            "tariffs_charged": html(
+                """Tariffs Charged<br>to the U.S.A.<br>Including Currency Manipulation and Trade Barriers"""
+            ),
+            "reciprocal_tariffs": html("U.S.A. Discounted<br>Reciprocal Tariffs"),
+            "0": "", "1": "", "2": "", "3": "",
+        })
+        .cols_width({
+            "country": "50%",
+            "0": "3%", "1": "7%", "2": "7%", "3": "3%",
+            "tariffs_charged": "18%",
+            "reciprocal_tariffs": "18%",
+        })
+        .tab_style(style=style.fill(color=DARK_NAVY_BLUE), locations=[loc.column_labels(), loc.body()])
+        .tab_style(style=style.borders(sides="all", color=DARK_NAVY_BLUE), locations=loc.body())
+        .tab_style(
+            style=style.text(font=google_font(name="Trajan Pro"), weight="bold", size="xx-large"),
+            locations=loc.body(),
+        )
+        .tab_style(
+            style=style.text(font=google_font(name="Georgia"), weight="bold", size="large"),
+            locations=loc.column_labels(),
+        )
+        .tab_style(style=style.text(color=WHITE), locations=loc.column_labels())
+        .tab_style(style=style.css("text-align: center;"), locations=loc.column_labels())
+        .tab_options(column_labels_border_bottom_style="hidden")
+        .tab_style(style=style.fill(color=DARK_NAVY_BLUE), locations=loc.body(rows=[-1]))
+    )
     
-    component = AsyncComponent()
-    pn.template.FastListTemplate(
-        site="Awesome Panel", site_url="https://awesome-panel.org", title="Async Tasks", main=[component], main_layout=None, main_max_width="400px"
-    ).servable()
+    return table
+
+
+# =============================================================================
+# Panel Widgets and Callbacks
+# =============================================================================
+# Reactive variable for tariffs data
+tarifs_rx = pn.rx(TARIFS_IMPOSED, sizing_mode="stretch_width")
+
+# Scale slider to control tariff scaling factor
+scale_slider = pn.widgets.FloatSlider(name="DISCOUNT FACTOR", value=0.5, step=0.1, start=0.1, end=2.0, sizing_mode="stretch_width")
+
+
+@pn.depends(scale_slider, watch=True)
+def update_scaled_tariffs(scale_value: float):
+    """
+    Update the tariffs data based on the scaling factor from the slider.
+    """
+    tarifs_rx.rx.value = scale_tariffs(tarifs_rx.rx.value, scale_value)
+
+
+# Button to randomize tariff values
+randomize_button = pn.widgets.Button(name="RANDOM TARIFS", width=200, button_type="primary")
+
+
+@pn.depends(randomize_button, watch=True)
+def update_random_tariffs(_):
+    """
+    Update the tariffs data with randomized values when the randomize button is clicked.
+    """
+    tarifs_rx.rx.value = randomize_tariffs(scale_slider.value)
+
+
+# Button to reset tariffs to the original imposed values
+reset_button = pn.widgets.Button(name="PSEUDO TARIFS", width=200, button_type="primary")
+
+
+@pn.depends(reset_button, watch=True)
+def reset_tariffs(_):
+    """
+    Reset the tariffs data to the original imposed tariffs.
+    """
+    tarifs_rx.rx.value = TARIFS_IMPOSED
+
+
+# =============================================================================
+# Custom JavaScript Component: Confetti Button
+# =============================================================================
+class ConfettiButton(JSComponent):
+    """
+    A custom Panel component that renders a button.
+    When clicked, it triggers a confetti animation using an external JS module.
+    """
+    scale = param.Number(0.5, allow_refs=True)
+
+    _stylesheets = ["""
+    button {
+        width: 100%;
+        background: #F6D588;
+        color: #0B162A;
+        border: none;
+        padding: 10px;
+        border-radius: 4px;
+    }
+    button:hover {
+        background: #F6D599;
+    }
+    """]
+    _esm = """
+    import confetti from "https://esm.sh/canvas-confetti@1.6.0";
+    
+    export function render({ model }) {
+      let btn = document.createElement("button");
+      btn.innerHTML = "IMPOSE NOW. ❤️ TARIFS.";
+      btn.addEventListener("click", () => {
+        const options = { scalar: 1.0 + 2.0 * model.scale, particleCount: 200 * model.scale, spread: 360 };
+        confetti(options);
+      });
+      return btn;
+    }
+    """
+
+
+# Create an instance of the confetti button
+confetti_button = ConfettiButton(scale=scale_slider, sizing_mode="stretch_width", margin=10)
+
+
+# =============================================================================
+# Layout and Serve
+# =============================================================================
+layout = pn.Column(
+    "# Reciprocal Tarifs Calculator",
+    pn.Row(reset_button, randomize_button, scale_slider),
+    confetti_button,
+    pn.pane.HTML(tarifs_rx.rx.pipe(create_tarifs_table)),
+    width=1000,
+    styles={"margin-right": "auto", "margin-left": "auto"},
+    stylesheets=[ "div: {background-color: #F6D588}" ],
+)
+
+layout.servable()