Expert skill for membrane filtration and separation system design including process selection, flux calculations, fouling analysis, and concentrate management.
Designs membrane filtration systems with process selection, flux calculations, fouling analysis, and concentrate management.
npx claudepluginhub a5c-ai/babysitterThis skill is limited to using the following tools:
Membrane filtration and separation system design for water and wastewater treatment applications.
This skill provides comprehensive capabilities for designing membrane treatment systems, including process selection, flux and recovery calculations, fouling analysis, pretreatment requirements, and concentrate management planning.
pip install numpy scipy pandas matplotlib
# For optimization
pip install scipy pymoo
# For visualization
pip install plotly seaborn
import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional
@dataclass
class FeedWaterQuality:
"""Feed water quality parameters"""
tds_mg_l: float
temperature_c: float
ph: float
tss_mg_l: float
toc_mg_l: float
hardness_mg_l: float = 0 # as CaCO3
silica_mg_l: float = 0
sdi: float = 0 # Silt Density Index
@dataclass
class MembraneElement:
"""Membrane element specifications"""
manufacturer: str
model: str
area_m2: float
permeability_lmh_bar: float # L/m2/hr/bar
salt_rejection: float # decimal
max_recovery: float # decimal
min_concentrate_flow_m3_hr: float
class ROSystemDesign:
"""Reverse Osmosis system design"""
def __init__(self, feed: FeedWaterQuality, element: MembraneElement):
self.feed = feed
self.element = element
def osmotic_pressure(self, tds_mg_l: float) -> float:
"""Calculate osmotic pressure in bar"""
# Simplified van't Hoff equation
# pi = i * C * R * T
# For typical water: pi (bar) ≈ 0.0385 * TDS (g/L)
return 0.0385 * (tds_mg_l / 1000) * (273.15 + self.feed.temperature_c) / 298.15
def temperature_correction_factor(self) -> float:
"""Temperature correction factor for flux"""
# TCF = exp(2640 * (1/298 - 1/T))
T_kelvin = 273.15 + self.feed.temperature_c
return np.exp(2640 * (1/298.15 - 1/T_kelvin))
def calculate_flux(self, feed_pressure_bar: float, recovery: float) -> float:
"""Calculate permeate flux in LMH"""
# Average osmotic pressure (considering concentration polarization)
avg_concentration_factor = 1 / (1 - recovery/2)
avg_tds = self.feed.tds_mg_l * avg_concentration_factor
pi_avg = self.osmotic_pressure(avg_tds)
# Net driving pressure
ndp = feed_pressure_bar - pi_avg - 1 # 1 bar backpressure
# Flux with temperature correction
tcf = self.temperature_correction_factor()
flux = self.element.permeability_lmh_bar * ndp * tcf
return flux
def design_system(self, feed_flow_m3_hr: float, target_recovery: float,
feed_pressure_bar: float) -> Dict:
"""Design RO system for given requirements"""
# Calculate flux
flux = self.calculate_flux(feed_pressure_bar, target_recovery)
# Permeate flow
permeate_flow = feed_flow_m3_hr * target_recovery
# Required membrane area
required_area = (permeate_flow * 1000) / flux # m2
# Number of elements
num_elements = np.ceil(required_area / self.element.area_m2)
# Elements per vessel (typical 6-8)
elements_per_vessel = 6
num_vessels = np.ceil(num_elements / elements_per_vessel)
actual_elements = num_vessels * elements_per_vessel
# Concentrate flow and quality
concentrate_flow = feed_flow_m3_hr * (1 - target_recovery)
concentrate_tds = self.feed.tds_mg_l / (1 - target_recovery)
# Permeate quality
avg_passage = 1 - self.element.salt_rejection
permeate_tds = self.feed.tds_mg_l * avg_passage * (1 + target_recovery)
# Specific energy consumption estimate
pump_efficiency = 0.80
sec_kwh_m3 = (feed_pressure_bar * 100) / (36 * pump_efficiency * target_recovery)
return {
'design_flux_lmh': flux,
'required_area_m2': required_area,
'num_vessels': int(num_vessels),
'elements_per_vessel': elements_per_vessel,
'total_elements': int(actual_elements),
'permeate_flow_m3_hr': permeate_flow,
'concentrate_flow_m3_hr': concentrate_flow,
'permeate_tds_mg_l': permeate_tds,
'concentrate_tds_mg_l': concentrate_tds,
'specific_energy_kwh_m3': sec_kwh_m3
}
# Example usage
feed = FeedWaterQuality(
tds_mg_l=2000,
temperature_c=25,
ph=7.5,
tss_mg_l=5,
toc_mg_l=3,
sdi=3
)
element = MembraneElement(
manufacturer='Example',
model='BW30-400',
area_m2=37.2,
permeability_lmh_bar=3.5,
salt_rejection=0.995,
max_recovery=0.15,
min_concentrate_flow_m3_hr=3.6
)
ro_system = ROSystemDesign(feed, element)
design = ro_system.design_system(
feed_flow_m3_hr=100,
target_recovery=0.75,
feed_pressure_bar=15
)
print(f"Design flux: {design['design_flux_lmh']:.1f} LMH")
print(f"Number of vessels: {design['num_vessels']}")
print(f"Total elements: {design['total_elements']}")
print(f"Permeate TDS: {design['permeate_tds_mg_l']:.0f} mg/L")
print(f"Specific energy: {design['specific_energy_kwh_m3']:.2f} kWh/m3")
class ScalingAnalysis:
"""Membrane scaling potential analysis"""
# Solubility product constants at 25C
Ksp = {
'CaCO3': 3.3e-9,
'CaSO4': 4.9e-5,
'BaSO4': 1.1e-10,
'SrSO4': 3.4e-7,
'SiO2': 120 # mg/L saturation
}
def __init__(self, water_quality: Dict):
self.wq = water_quality
def langelier_saturation_index(self, temperature_c: float,
tds_mg_l: float) -> float:
"""Calculate Langelier Saturation Index for CaCO3"""
pH = self.wq.get('pH', 7.5)
Ca = self.wq.get('Ca_mg_l', 100)
alkalinity = self.wq.get('alkalinity_mg_l', 100)
# pHs calculation (simplified)
pCa = -np.log10(Ca / 40080) # Convert to mol/L
pAlk = -np.log10(alkalinity / 50040)
# Temperature and TDS corrections
A = (np.log10(tds_mg_l) - 1) / 10
B = -13.12 * np.log10(temperature_c + 273) + 34.55
C = np.log10(Ca / 40.08) - 0.4
D = np.log10(alkalinity / 50.04)
pHs = (9.3 + A + B) - C - D
lsi = pH - pHs
return lsi
def stiff_davis_index(self, ionic_strength: float) -> float:
"""Calculate Stiff-Davis Stability Index for high TDS waters"""
pH = self.wq.get('pH', 7.5)
Ca = self.wq.get('Ca_mg_l', 100)
alkalinity = self.wq.get('alkalinity_mg_l', 100)
# Activity coefficient correction
K = 2.22e-14 # CaCO3 equilibrium constant
pCa = -np.log10(Ca / 40080 * 0.4) # With activity correction
pAlk = -np.log10(alkalinity / 50040 * 0.4)
pK = -np.log10(K)
pHs = pK + pCa + pAlk
sdi = pH - pHs
return sdi
def calcium_sulfate_saturation(self, recovery: float) -> float:
"""Calculate CaSO4 saturation ratio at given recovery"""
Ca = self.wq.get('Ca_mg_l', 100)
SO4 = self.wq.get('SO4_mg_l', 200)
# Concentration factor
cf = 1 / (1 - recovery)
# Ion product in concentrate
Ca_conc = (Ca / 40080) * cf # mol/L
SO4_conc = (SO4 / 96060) * cf # mol/L
ip = Ca_conc * SO4_conc
# Saturation ratio
sr = ip / self.Ksp['CaSO4']
return sr
def silica_saturation(self, recovery: float, temperature_c: float) -> float:
"""Calculate silica saturation ratio"""
SiO2 = self.wq.get('SiO2_mg_l', 20)
# Concentration factor
cf = 1 / (1 - recovery)
SiO2_conc = SiO2 * cf
# Temperature-dependent saturation (simplified)
saturation_limit = self.Ksp['SiO2'] + (temperature_c - 25) * 2
return SiO2_conc / saturation_limit
def analyze_scaling_potential(self, recovery: float,
temperature_c: float = 25) -> Dict:
"""Complete scaling potential analysis"""
tds_concentrate = self.wq.get('tds_mg_l', 1000) / (1 - recovery)
results = {
'recovery': recovery,
'concentration_factor': 1 / (1 - recovery),
'concentrate_tds_mg_l': tds_concentrate,
'lsi': self.langelier_saturation_index(temperature_c, tds_concentrate),
'caso4_saturation': self.calcium_sulfate_saturation(recovery),
'silica_saturation': self.silica_saturation(recovery, temperature_c)
}
# Risk assessment
results['caco3_risk'] = 'HIGH' if results['lsi'] > 0.5 else \
'MODERATE' if results['lsi'] > 0 else 'LOW'
results['caso4_risk'] = 'HIGH' if results['caso4_saturation'] > 0.8 else \
'MODERATE' if results['caso4_saturation'] > 0.5 else 'LOW'
results['silica_risk'] = 'HIGH' if results['silica_saturation'] > 0.8 else \
'MODERATE' if results['silica_saturation'] > 0.5 else 'LOW'
return results
# Example usage
water_quality = {
'pH': 7.8,
'tds_mg_l': 2000,
'Ca_mg_l': 150,
'SO4_mg_l': 300,
'alkalinity_mg_l': 180,
'SiO2_mg_l': 25
}
scaling = ScalingAnalysis(water_quality)
results = scaling.analyze_scaling_potential(recovery=0.75, temperature_c=25)
print(f"Concentration factor: {results['concentration_factor']:.2f}x")
print(f"LSI: {results['lsi']:.2f} - CaCO3 risk: {results['caco3_risk']}")
print(f"CaSO4 saturation: {results['caso4_saturation']:.2f} - Risk: {results['caso4_risk']}")
print(f"Silica saturation: {results['silica_saturation']:.2f} - Risk: {results['silica_risk']}")
class CIPProtocol:
"""Clean-in-Place protocol development"""
def __init__(self, membrane_type: str = 'polyamide'):
self.membrane_type = membrane_type
# Chemical compatibility
self.ph_limits = {
'polyamide': (2, 11),
'cellulose_acetate': (4, 7),
'polysulfone': (1, 13)
}
self.temperature_limit = {
'polyamide': 45,
'cellulose_acetate': 35,
'polysulfone': 50
}
def recommend_cleaning_chemicals(self, fouling_type: str) -> List[Dict]:
"""Recommend cleaning chemicals based on fouling type"""
recommendations = {
'biofouling': [
{'chemical': 'NaOH', 'concentration': '0.1%', 'ph': 12, 'temperature_c': 35},
{'chemical': 'Biocide', 'concentration': 'Per manufacturer', 'ph': 7, 'temperature_c': 25}
],
'organic': [
{'chemical': 'NaOH', 'concentration': '0.1%', 'ph': 12, 'temperature_c': 35},
{'chemical': 'Surfactant', 'concentration': '0.1%', 'ph': 10, 'temperature_c': 30}
],
'colloidal': [
{'chemical': 'NaOH', 'concentration': '0.1%', 'ph': 11, 'temperature_c': 30},
{'chemical': 'EDTA', 'concentration': '1%', 'ph': 10, 'temperature_c': 30}
],
'scale_calcium': [
{'chemical': 'HCl', 'concentration': '0.2%', 'ph': 2, 'temperature_c': 25},
{'chemical': 'Citric acid', 'concentration': '2%', 'ph': 3, 'temperature_c': 30}
],
'scale_silica': [
{'chemical': 'NaOH', 'concentration': '0.1%', 'ph': 11, 'temperature_c': 35}
]
}
return recommendations.get(fouling_type, [])
def generate_cip_procedure(self, fouling_types: List[str],
system_volume_m3: float) -> Dict:
"""Generate complete CIP procedure"""
ph_min, ph_max = self.ph_limits.get(self.membrane_type, (2, 12))
temp_max = self.temperature_limit.get(self.membrane_type, 40)
procedure = {
'membrane_type': self.membrane_type,
'ph_operating_range': f'{ph_min} - {ph_max}',
'max_temperature_c': temp_max,
'system_volume_m3': system_volume_m3,
'steps': []
}
# Initial flush
procedure['steps'].append({
'step': 1,
'action': 'Low pressure flush',
'duration_min': 10,
'flow_rate': 'High',
'description': 'Flush with permeate water to remove loose deposits'
})
# Cleaning steps based on fouling types
step_num = 2
for fouling in fouling_types:
chemicals = self.recommend_cleaning_chemicals(fouling)
for chem in chemicals:
if chem['temperature_c'] <= temp_max and ph_min <= chem['ph'] <= ph_max:
procedure['steps'].append({
'step': step_num,
'action': f'Chemical clean - {chem["chemical"]}',
'concentration': chem['concentration'],
'ph': chem['ph'],
'temperature_c': chem['temperature_c'],
'duration_min': 30,
'recirculation_time_min': 60,
'soak_time_min': 30,
'description': f'Target: {fouling} fouling removal'
})
step_num += 1
# Final flush
procedure['steps'].append({
'step': step_num,
'action': 'Final flush',
'duration_min': 20,
'description': 'Flush until pH and conductivity match feed water'
})
return procedure
# Example usage
cip = CIPProtocol(membrane_type='polyamide')
procedure = cip.generate_cip_procedure(
fouling_types=['biofouling', 'scale_calcium'],
system_volume_m3=5
)
print("CIP Procedure:")
for step in procedure['steps']:
print(f" Step {step['step']}: {step['action']}")
Activates when the user asks about AI prompts, needs prompt templates, wants to search for prompts, or mentions prompts.chat. Use for discovering, retrieving, and improving prompts.
Search, retrieve, and install Agent Skills from the prompts.chat registry using MCP tools. Use when the user asks to find skills, browse skill catalogs, install a skill for Claude, or extend Claude's capabilities with reusable AI agent components.
This skill should be used when the user wants to "create a skill", "add a skill to plugin", "write a new skill", "improve skill description", "organize skill content", or needs guidance on skill structure, progressive disclosure, or skill development best practices for Claude Code plugins.