메뉴
Expert skill for membrane filtration and separation system design including process selection, flux calculations, fouling analysis, and concentrate management.
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/a5c-ai/babysitter --skill membrane-system-design명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
Reference for querying the Atlas knowledge graph through its MCP tools — the SECONDARY enrichment/comparison layer that adds best-practice context to systems you have ALREADY scanned from your real sources (`az`, repos, dirs). Use when you need to look up nodes, edges, kinds, clusters, stats, or wiki pages in Atlas to compare against your real inventory. (atlas graph, query atlas, atlas mcp, search the graph, graph neighbors, atlas record, atlas kinds, enrichment layer)
Atlas turns your STATED NEED into a real systems atlas by SCANNING your actual sources (Azure via `az`, git repos, local dirs) and process/data mining them, THEN enriching against the Atlas knowledge graph. Use this skill when asked to inventory/map your real systems, scan your cloud + repos + directories, mine the real processes or data they contain, or collect their real constraints/gotchas. (atlas, scan my systems, inventory our azure account, map my repos, real systems atlas, process mining, data mining, collect nuances, system discovery)
This skill should be used when the user asks to "find skills in the wild", "assimilate popular workflows", "discover SKILL.md files in repos", "research external skills", "find workflow patterns", "survey the skill landscape", "what skills exist out there", or wants to investigate public repositories for extractable processes, babysitter plugins, and reusable procedural insights. Searches GitHub for SKILL.md files, classifies repos by archetype, and maintains structured research under docs/reference-repos/.
SKILL.md 표시 중
SOC 직업 분류 기준
| name | membrane-system-design |
| description | Expert skill for membrane filtration and separation system design including process selection, flux calculations, fouling analysis, and concentrate management. |
| allowed-tools | Read, Grep, Write, Bash, Edit, Glob |
| category | Water and Wastewater Treatment |
| backlog-id | SK-003 |
| metadata | {"author":"babysitter-sdk","version":"1.0.0"} |
| graph | {"domains":["domain:environmental-engineering"],"skillAreas":["skill-area:data-analysis","skill-area:statistical-analysis","skill-area:geospatial-data-analysis"],"workflows":["workflow:experiment-design"],"roles":["role:research-engineer"]} |
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']}")