Upload 2 files

src/app.py

@@ -1,12 +1,12 @@
 """
 Streamlit App for PVD Consolidation Analysis
-Shows settlement vs time curve
+Shows settlement vs time curve with vacuum and staged loading support
 """
 
 import streamlit as st
 import numpy as np
 import matplotlib.pyplot as plt
-from pvd_consolidation import PVDConsolidation, SoilLayer, PVDProperties
+from pvd_consolidation import PVDConsolidation, SoilLayer, PVDProperties, LoadingStage
 
 # Page configuration
 st.set_page_config(
@@ -101,13 +101,75 @@ if well_resistance == "Custom qw":
 else:
     qw = 1e12  # Very large value for negligible resistance
 
-pvd = PVDProperties(dw=dw, ds=ds, De=De, L_drain=L_drain, qw=qw)
+# Vacuum depth loss option
+vacuum_depth_loss = st.sidebar.checkbox(
+    "Vacuum loss with depth",
+    value=True,
+    help="If checked: vacuum decreases with depth (u(z) = u_surface × exp(-z/L)). If unchecked: uniform vacuum at all depths.",
+)
+
+pvd = PVDProperties(
+    dw=dw, ds=ds, De=De, L_drain=L_drain, qw=qw, vacuum_depth_loss=vacuum_depth_loss
+)
+
+# Loading Configuration
+st.sidebar.subheader("Loading Configuration")
+
+loading_type = st.sidebar.radio(
+    "Loading Type",
+    ["Single Stage", "Multi-Stage"],
+    help="Single stage: One loading condition. Multi-stage: Multiple loading stages over time.",
+)
+
+if loading_type == "Single Stage":
+    st.sidebar.markdown("**Single Stage Loading**")
+    surcharge = st.sidebar.number_input(
+        "Surcharge (kPa)", min_value=0.0, value=100.0, step=10.0
+    )
+    vacuum = st.sidebar.number_input(
+        "Vacuum (kPa)",
+        min_value=0.0,
+        value=0.0,
+        step=10.0,
+        help="Vacuum pressure (positive value). 80 kPa vacuum ≈ 80 kPa surcharge",
+    )
+    loading_stages = None
+
+else:  # Multi-Stage
+    st.sidebar.markdown("**Multi-Stage Loading**")
+    n_stages = st.sidebar.number_input(
+        "Number of Stages", min_value=1, max_value=10, value=2
+    )
+
+    loading_stages = []
+    for i in range(n_stages):
+        st.sidebar.markdown(f"**Stage {i + 1}**")
+        col1, col2, col3 = st.sidebar.columns(3)
+
+        with col1:
+            start_time = st.number_input(
+                f"Time (y)",
+                min_value=0.0,
+                value=float(i * 0.5),
+                step=0.1,
+                key=f"time_{i}",
+            )
+        with col2:
+            stage_surcharge = st.number_input(
+                f"Surcharge", min_value=0.0, value=50.0, step=10.0, key=f"sur_{i}"
+            )
+        with col3:
+            stage_vacuum = st.number_input(
+                f"Vacuum", min_value=0.0, value=0.0, step=10.0, key=f"vac_{i}"
+            )
+
+        loading_stages.append(LoadingStage(start_time, stage_surcharge, stage_vacuum))
+
+    surcharge = 0.0  # Not used in multi-stage
+    vacuum = 0.0
 
 # Analysis Parameters
 st.sidebar.subheader("Analysis Parameters")
-surcharge = st.sidebar.number_input(
-    "Surcharge (kPa)", min_value=1.0, value=100.0, step=10.0
-)
 t_max = st.sidebar.number_input("Max Time (years)", min_value=0.1, value=2.0, step=0.1)
 dt = st.sidebar.number_input(
     "Time Step (years)", min_value=0.001, value=0.01, step=0.001, format="%.3f"
@@ -117,8 +179,15 @@ dt = st.sidebar.number_input(
 if st.sidebar.button("🚀 Run Analysis", type="primary"):
     with st.spinner("Calculating consolidation..."):
         try:
-            # Create analysis
-            analysis = PVDConsolidation(layers, pvd, surcharge, dt)
+            # Create analysis with vacuum and staged loading support
+            if loading_stages:
+                analysis = PVDConsolidation(
+                    layers, pvd, loading_stages=loading_stages, dt=dt
+                )
+            else:
+                analysis = PVDConsolidation(
+                    layers, pvd, surcharge=surcharge, vacuum=vacuum, dt=dt
+                )
 
             # Calculate settlement vs time
             time, settlement = analysis.settlement_vs_time(t_max=t_max, n_points=200)

src/pvd_consolidation.py

@@ -12,6 +12,22 @@ from dataclasses import dataclass
 from typing import List, Tuple, Dict, Any
 
 
+@dataclass
+class LoadingStage:
+    """Loading stage definition for staged loading"""
+
+    start_time: float  # Time when this stage starts (years)
+    surcharge: float  # Surcharge load (kPa)
+    vacuum: float  # Vacuum pressure (kPa, positive value, will be applied as negative)
+
+    def __post_init__(self):
+        """Validate inputs"""
+        if self.vacuum < 0:
+            raise ValueError(
+                "Vacuum should be specified as positive value (e.g., 80 for -80 kPa)"
+            )
+
+
 @dataclass
 class SoilLayer:
     """Soil layer properties"""
@@ -36,6 +52,13 @@ class PVDProperties:
     De: float  # Equivalent diameter of unit cell (m)
     L_drain: float  # Total drain spacing for two-way drainage (m)
     qw: float  # Well discharge capacity (m³/year)
+    r_influence: float = None  # Vacuum influence radius (m), default = De/4
+    vacuum_depth_loss: bool = True  # Use vacuum loss with depth (default: True)
+
+    def __post_init__(self):
+        """Set default vacuum influence radius"""
+        if self.r_influence is None:
+            self.r_influence = self.De / 4
 
 
 class PVDConsolidation:
@@ -47,7 +70,9 @@ class PVDConsolidation:
         self,
         soil_layers: List[SoilLayer],
         pvd: PVDProperties,
-        surcharge: float,
+        surcharge: float = 0.0,
+        vacuum: float = 0.0,
+        loading_stages: List[LoadingStage] = None,
         dt: float = 0.01,
     ):
         """
@@ -60,15 +85,33 @@ class PVDConsolidation:
         pvd : PVDProperties
             PVD installation properties
         surcharge : float
-            Applied surcharge load (kPa)
+            Applied surcharge load (kPa) - for single-stage loading
+        vacuum : float
+            Applied vacuum pressure (kPa, positive value) - for single-stage loading
+        loading_stages : List[LoadingStage]
+            List of loading stages for multi-stage loading (overrides surcharge/vacuum)
         dt : float
             Time step for finite difference (years)
         """
         self.layers = soil_layers
         self.pvd = pvd
-        self.surcharge = surcharge
         self.dt = dt
 
+        # Setup loading
+        if loading_stages is not None:
+            self.loading_stages = sorted(loading_stages, key=lambda x: x.start_time)
+            self.use_staged_loading = True
+        else:
+            # Single stage loading
+            self.loading_stages = [
+                LoadingStage(start_time=0.0, surcharge=surcharge, vacuum=vacuum)
+            ]
+            self.use_staged_loading = False
+
+        # For compatibility
+        self.surcharge = surcharge
+        self.vacuum = vacuum
+
         # Calculate total thickness
         self.total_thickness = sum(layer.thickness for layer in soil_layers)
 
@@ -119,6 +162,127 @@ class PVDConsolidation:
         self.n_nodes = len(self.z_coords)
         self.dz = np.diff(self.z_coords)
 
+    def get_current_loading(self, t: float) -> Tuple[float, float]:
+        """
+        Get current surcharge and vacuum at time t
+
+        Parameters:
+        -----------
+        t : float
+            Current time (years)
+
+        Returns:
+        --------
+        surcharge, vacuum : float, float
+            Current surcharge and vacuum values (kPa)
+        """
+        # Find the active loading stage
+        active_stage = self.loading_stages[0]
+        for stage in self.loading_stages:
+            if t >= stage.start_time:
+                active_stage = stage
+            else:
+                break
+
+        return active_stage.surcharge, active_stage.vacuum
+
+    def calculate_vacuum_at_radius(self, r: float, vacuum_magnitude: float) -> float:
+        """
+        Calculate vacuum pressure at radial distance r from drain
+
+        u(r) = u_drain × exp(-r/r_influence)
+
+        Parameters:
+        -----------
+        r : float
+            Radial distance from drain (m)
+        vacuum_magnitude : float
+            Vacuum at drain (kPa, positive value)
+
+        Returns:
+        --------
+        vacuum_at_r : float
+            Vacuum pressure at distance r (kPa, positive value)
+        """
+        if vacuum_magnitude == 0:
+            return 0.0
+
+        # Vacuum decreases exponentially with distance
+        vacuum_at_r = vacuum_magnitude * np.exp(-r / self.pvd.r_influence)
+
+        return vacuum_at_r
+
+    def calculate_layer_average_vacuum(
+        self, layer_idx: int, vacuum_surface: float
+    ) -> float:
+        """
+        Calculate average vacuum pressure for a specific soil layer
+
+        If vacuum_depth_loss = True:
+            Vacuum is full at surface and decreases with depth
+            u(z) = u_surface × exp(-z/L_drain)
+
+        If vacuum_depth_loss = False:
+            Uniform vacuum throughout (full vacuum at all depths within L_drain)
+
+        Parameters:
+        -----------
+        layer_idx : int
+            Layer index
+        vacuum_surface : float
+            Vacuum at surface (kPa, positive value)
+
+        Returns:
+        --------
+        avg_vacuum : float
+            Average vacuum for this layer (kPa)
+        """
+        if vacuum_surface == 0:
+            return 0.0
+
+        layer = self.layers[layer_idx]
+
+        # Find layer depth range
+        depth_top = sum(self.layers[i].thickness for i in range(layer_idx))
+        depth_bottom = depth_top + layer.thickness
+
+        # Check if layer is within drain length
+        if depth_top >= self.pvd.L_drain:
+            # Layer is completely below drain length - no vacuum effect
+            return 0.0
+
+        # Option 1: Uniform vacuum (no depth loss)
+        if not self.pvd.vacuum_depth_loss:
+            # Full vacuum throughout the layer (within drain length)
+            return vacuum_surface
+
+        # Option 2: Vacuum loss with depth
+        # Adjust bottom depth if it exceeds drain length
+        if depth_bottom > self.pvd.L_drain:
+            depth_bottom = self.pvd.L_drain
+            effective_thickness = depth_bottom - depth_top
+        else:
+            effective_thickness = layer.thickness
+
+        # Integrate vacuum over layer thickness
+        # u(z) = u_surface × exp(-z/L_drain)
+        # Average = (1/h) ∫[z_top to z_bottom] u_surface × exp(-z/L_drain) dz
+
+        L = self.pvd.L_drain
+
+        # Analytical integration:
+        # ∫ u_surface × exp(-z/L) dz = -u_surface × L × exp(-z/L) + C
+
+        integral_bottom = -vacuum_surface * L * np.exp(-depth_bottom / L)
+        integral_top = -vacuum_surface * L * np.exp(-depth_top / L)
+
+        integral = integral_bottom - integral_top
+
+        # Average over effective layer thickness
+        avg_vacuum = integral / effective_thickness
+
+        return avg_vacuum
+
     def calculate_pvd_factors_layer(self, layer_idx: int) -> Tuple[float, float, float]:
         """
         Calculate PVD influence factors for a specific layer
@@ -265,6 +429,9 @@ class PVDConsolidation:
         """
         Calculate degree of consolidation in vertical direction
 
+        For layers below PVD: drainage path is from layer midpoint to bottom of PVD
+        For layers within/above PVD: normal two-way drainage
+
         Parameters:
         -----------
         t : float
@@ -278,18 +445,52 @@ class PVDConsolidation:
         Uv = np.zeros(self.n_nodes)
 
         for i in range(self.n_nodes):
-            z = self.z_coords[i]
+            depth = self.z_coords[i]
             Cv = self.Cv_profile[i]
-            H = self.total_thickness
+            layer_idx = self.layer_indices[i]
+
+            # Determine drainage path length H
+            if depth > self.pvd.L_drain:
+                # Below PVD: drainage from this point UP to bottom of PVD
+                # For layer below PVD, use distance from layer midpoint to PVD bottom
+
+                # Find layer boundaries
+                cumulative_depth = 0
+                for idx in range(layer_idx):
+                    cumulative_depth += self.layers[idx].thickness
+                layer_top = cumulative_depth
+                layer_bottom = cumulative_depth + self.layers[layer_idx].thickness
+                layer_midpoint = (layer_top + layer_bottom) / 2
+
+                # Drainage path: from layer midpoint to PVD bottom
+                H = abs(layer_midpoint - self.pvd.L_drain)
+
+                # One-way drainage (upward only to PVD)
+                drainage_type = "one-way"
+            else:
+                # Within or above PVD: normal two-way drainage
+                H = self.total_thickness
+                drainage_type = "two-way"
 
             # Time factor
-            Tv = Cv * t / H**2
+            if H > 0:
+                Tv = Cv * t / H**2
+            else:
+                Tv = 1e10  # Very large, instant drainage
 
             # Calculate Uv using Terzaghi's solution
-            if Tv < 0.217:
-                Uv[i] = np.sqrt(4 * Tv / np.pi)
+            if drainage_type == "one-way":
+                # One-way drainage (for layers below PVD)
+                if Tv < 0.05:
+                    Uv[i] = np.sqrt(4 * Tv / np.pi)
+                else:
+                    Uv[i] = 1 - (8 / np.pi**2) * np.exp(-(np.pi**2) * Tv / 4)
             else:
-                Uv[i] = 1 - (8 / np.pi**2) * np.exp(-(np.pi**2) * Tv / 4)
+                # Two-way drainage (for layers within PVD zone)
+                if Tv < 0.217:
+                    Uv[i] = np.sqrt(4 * Tv / np.pi)
+                else:
+                    Uv[i] = 1 - (8 / np.pi**2) * np.exp(-(np.pi**2) * Tv / 4)
 
             Uv[i] = min(Uv[i], 1.0)
 
@@ -332,6 +533,7 @@ class PVDConsolidation:
     def calculate_settlement(self, t: float) -> Tuple[float, np.ndarray]:
         """
         Calculate total settlement at time t
+        For staged loading: calculates settlement contribution from each stage
 
         Parameters:
         -----------
@@ -345,48 +547,107 @@ class PVDConsolidation:
         layer_settlements : ndarray
             Settlement for each layer (m)
         """
-        U = self.calculate_total_U(t)
-
         layer_settlements = np.zeros(len(self.layers))
 
-        for i, layer in enumerate(self.layers):
-            # Find nodes in this layer
-            layer_mask = self.layer_indices == i
-            U_layer = np.mean(U[layer_mask])
-
-            # Calculate final settlement for this layer
-            sigma_final = layer.sigma_ini + self.surcharge
-
-            # Settlement components
-            if sigma_final <= layer.sigma_p:
-                # Recompression only
-                Sc_final = (
-                    layer.RR * np.log10(sigma_final / layer.sigma_ini) * layer.thickness
-                )
-            else:
-                if layer.sigma_ini >= layer.sigma_p:
-                    # Virgin compression only
-                    Sc_final = (
-                        layer.CR
-                        * np.log10(sigma_final / layer.sigma_ini)
-                        * layer.thickness
+        # For each loading stage, calculate its contribution to settlement
+        for stage_idx, stage in enumerate(self.loading_stages):
+            if t < stage.start_time:
+                # This stage hasn't started yet
+                continue
+
+            # Time since this stage started
+            t_stage = t - stage.start_time
+
+            # Calculate consolidation degree for this stage's time duration
+            U_stage = self.calculate_total_U(t_stage)
+
+            # Calculate settlement for each layer from this load increment
+            for i, layer in enumerate(self.layers):
+                # Find nodes in this layer
+                layer_mask = self.layer_indices == i
+                U_layer = np.mean(U_stage[layer_mask])
+
+                # Get layer-specific vacuum (decreases with depth)
+                # Previous stage vacuum for this layer
+                if stage_idx == 0:
+                    avg_vacuum_prev_layer = 0.0
+                else:
+                    prev_stage = self.loading_stages[stage_idx - 1]
+                    avg_vacuum_prev_layer = self.calculate_layer_average_vacuum(
+                        i, prev_stage.vacuum
                     )
+
+                # Current stage vacuum for this layer
+                avg_vacuum_current_layer = self.calculate_layer_average_vacuum(
+                    i, stage.vacuum
+                )
+
+                # Previous stage load for this layer
+                if stage_idx == 0:
+                    prev_surcharge = 0.0
+                    sigma_prev_layer = 0.0
+                else:
+                    prev_surcharge = self.loading_stages[stage_idx - 1].surcharge
+                    sigma_prev_layer = prev_surcharge + avg_vacuum_prev_layer
+
+                # Current stage load for this layer
+                sigma_current_layer = stage.surcharge + avg_vacuum_current_layer
+
+                # Load increment for this layer
+                delta_sigma_layer = sigma_current_layer - sigma_prev_layer
+
+                if abs(delta_sigma_layer) < 1e-6:
+                    # No significant load change in this layer
+                    continue
+
+                # Stress before this stage
+                sigma_before = layer.sigma_ini + sigma_prev_layer
+
+                # Stress after this stage (if fully consolidated)
+                sigma_after = layer.sigma_ini + sigma_current_layer
+
+                # Calculate ultimate settlement for this load increment
+                Sc_increment = 0.0
+
+                if delta_sigma_layer > 0:
+                    # Loading increment
+                    if sigma_after <= layer.sigma_p:
+                        # All in recompression range
+                        Sc_increment = (
+                            layer.RR
+                            * np.log10(sigma_after / sigma_before)
+                            * layer.thickness
+                        )
+                    elif sigma_before >= layer.sigma_p:
+                        # All in virgin compression range
+                        Sc_increment = (
+                            layer.CR
+                            * np.log10(sigma_after / sigma_before)
+                            * layer.thickness
+                        )
+                    else:
+                        # Crosses from recompression to virgin compression
+                        Sc_recomp = (
+                            layer.RR
+                            * np.log10(layer.sigma_p / sigma_before)
+                            * layer.thickness
+                        )
+                        Sc_virgin = (
+                            layer.CR
+                            * np.log10(sigma_after / layer.sigma_p)
+                            * layer.thickness
+                        )
+                        Sc_increment = Sc_recomp + Sc_virgin
                 else:
-                    # Both recompression and virgin compression
-                    Sc_recomp = (
+                    # Unloading (e.g., vacuum removal) - use recompression/swelling
+                    Sc_increment = (
                         layer.RR
-                        * np.log10(layer.sigma_p / layer.sigma_ini)
-                        * layer.thickness
-                    )
-                    Sc_virgin = (
-                        layer.CR
-                        * np.log10(sigma_final / layer.sigma_p)
+                        * np.log10(sigma_after / sigma_before)
                         * layer.thickness
                     )
-                    Sc_final = Sc_recomp + Sc_virgin
 
-            # Settlement at time t
-            layer_settlements[i] = U_layer * Sc_final
+                # Add this stage's contribution (with consolidation degree)
+                layer_settlements[i] += U_layer * Sc_increment
 
         total_settlement = np.sum(layer_settlements)