From 80c76b3c6c52f11cd2861987e67fb65cd7a07915 Mon Sep 17 00:00:00 2001 From: Yalin Date: Sun, 1 Oct 2023 21:57:18 -0400 Subject: [PATCH] update module names to be consistent --- qsdsan/sanunits/_anaerobic_reactor.py | 1061 ++++++++++++++++++ qsdsan/sanunits/_membrane_bioreactor.py | 1342 +++++++++++++++++++++++ qsdsan/sanunits/_tank.py | 162 +++ qsdsan/sanunits/_toilet.py | 1190 ++++++++++++++++++++ qsdsan/sanunits/_treatment_bed.py | 412 +++++++ 5 files changed, 4167 insertions(+) create mode 100644 qsdsan/sanunits/_anaerobic_reactor.py create mode 100644 qsdsan/sanunits/_membrane_bioreactor.py create mode 100644 qsdsan/sanunits/_tank.py create mode 100644 qsdsan/sanunits/_toilet.py create mode 100644 qsdsan/sanunits/_treatment_bed.py diff --git a/qsdsan/sanunits/_anaerobic_reactor.py b/qsdsan/sanunits/_anaerobic_reactor.py new file mode 100644 index 00000000..69d086a3 --- /dev/null +++ b/qsdsan/sanunits/_anaerobic_reactor.py @@ -0,0 +1,1061 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +''' +QSDsan: Quantitative Sustainable Design for sanitation and resource recovery systems + +This module is developed by: + Yalin Li + + Joy Zhang + +This module is under the University of Illinois/NCSA Open Source License. +Please refer to https://github.com/QSD-Group/QSDsan/blob/main/LICENSE.txt +for license details. +''' + +import numpy as np +from math import ceil, pi +from biosteam import Stream +from .. import SanUnit, Construction, WasteStream +from ..processes import Decay +from ..sanunits import HXutility, WWTpump, CSTR +from ..utils import ospath, load_data, data_path, auom, \ + calculate_excavation_volume, ExogenousDynamicVariable as EDV +__all__ = ( + 'AnaerobicBaffledReactor', + 'AnaerobicCSTR', + 'AnaerobicDigestion', + 'SludgeDigester', + ) + + +# %% + +abr_path = ospath.join(data_path, 'sanunit_data/_anaerobic_baffled_reactor.tsv') + +class AnaerobicBaffledReactor(SanUnit, Decay): + ''' + Anaerobic baffled reactor with the production of biogas based on + `Trimmer et al. `_ + + To enable life cycle assessment, the following impact items should be pre-constructed: + `Concrete`, `Gravel`, `Excavation`. + + Parameters + ---------- + ins : Iterable(stream) + Waste for treatment. + outs : Iterable(stream) + Treated waste, biogas, fugitive CH4, and fugitive N2O. + + Examples + -------- + `bwaise systems `_ + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.processes.Decay ` + ''' + _N_ins = 1 + _N_outs = 4 + _run = Decay._first_order_run + _units = { + 'Residence time': 'd', + 'Reactor length': 'm', + 'Reactor width': 'm', + 'Reactor height': 'm', + 'Single reactor volume': 'm3' + } + gravel_density = 1600 + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + include_construction=True, + degraded_components=('OtherSS',), + if_capture_biogas=True, + if_N2O_emission=False, + **kwargs): + Decay.__init__(self, ID, ins, outs, thermo, init_with, F_BM_default=1, + include_construction=include_construction, + degraded_components=degraded_components, + if_capture_biogas=if_capture_biogas, + if_N2O_emission=if_N2O_emission, + ) + + data = load_data(path=abr_path) + for para in data.index: + value = float(data.loc[para]['expected']) + setattr(self, '_'+para, value) + del data + + for attr, value in kwargs.items(): + setattr(self, attr, value) + + + def _init_lca(self): + self.construction = [ + Construction('concrete', linked_unit=self, item='Concrete', quantity_unit='m3'), + Construction('gravel', linked_unit=self, item='Gravel', quantity_unit='kg'), + Construction('excavation', linked_unit=self, item='Excavation', quantity_unit='m3'), + ] + + + def _design(self): + design = self.design_results + design['Residence time'] = self.tau + design['Reactor number'] = N = self.N_reactor + design['Baffle number'] = N_b = self.N_baffle + design['Reactor length'] = L = self.reactor_L + design['Reactor width'] = W = self.reactor_W + design['Reactor height'] = H = self.reactor_H + design['Single reactor volume'] = V = L*W*H + + if self.include_construction: + constr = self.construction + concrete = N*self.concrete_thickness*(2*L*W+2*L*H+(2+N_b)*W*H)*self.add_concrete + constr[0].quantity = concrete + constr[1].quantity = N*V/(N_b+1) * self.gravel_density + constr[2].quantity = N * V # excavation + + self.add_construction() + + + @property + def tau(self): + '''[float] Residence time, [d].''' + return self._tau + @tau.setter + def tau(self, i): + self._tau = i + + @property + def COD_removal(self): + '''[float] Fraction of COD removed during treatment.''' + return self._COD_removal + @COD_removal.setter + def COD_removal(self, i): + self._COD_removal = i + + @property + def N_removal(self): + '''[float] Fraction of N removed during treatment.''' + return self._N_removal + @N_removal.setter + def N_removal(self, i): + self._N_removal = i + + @property + def N_reactor(self): + '''[int] Number of reactors, float will be converted to the smallest integer.''' + return self._N_reactor + @N_reactor.setter + def N_reactor(self, i): + self._N_reactor = ceil(i) + + @property + def reactor_L(self): + '''[float] Reactor length, [m].''' + return self._reactor_L + @reactor_L.setter + def reactor_L(self, i): + self._reactor_L = i + + @property + def reactor_W(self): + '''[float] Reactor width, [m].''' + return self._reactor_W + @reactor_W.setter + def reactor_W(self, i): + self._reactor_W = i + + @property + def reactor_H(self): + '''[float] Reactor height, [m].''' + return self._reactor_H + @reactor_H.setter + def reactor_H(self, i): + self._reactor_H = i + + @property + def N_baffle(self): + '''[int] Number of reactors, float will be converted to the smallest integer.''' + return self._N_baffle + @N_baffle.setter + def N_baffle(self, i): + self._N_baffle = ceil(i) + + @property + def add_concrete(self): + ''' + [float] Additional concrete as a fraction of the reactor concrete usage + to account for receiving basin and biogas tank. + ''' + return self._add_concrete + @add_concrete.setter + def add_concrete(self, i): + self._add_concrete = i + + @property + def concrete_thickness(self): + '''[float] Thickness of the concrete wall.''' + return self._concrete_thickness + @concrete_thickness.setter + def concrete_thickness(self, i): + self._concrete_thickness = i + + +# %% + +class AnaerobicCSTR(CSTR): + + ''' + An anaerobic continuous stirred tank reactor with biogas in headspace. [1]_, [2]_ + + Parameters + ---------- + ins : :class:`WasteStream` + Influent to the reactor. + outs : Iterable + Biogas and treated effluent(s). + V_liq : float, optional + Liquid-phase volume [m^3]. The default is 3400. + V_gas : float, optional + Headspace volume [m^3]. The default is 300. + model : :class:`Processes`, optional + The kinetic model, typically ADM1-like. The default is None. + T : float, optional + Operation temperature [K]. The default is 308.15. + headspace_P : float, optional + Headspace pressure, if fixed [bar]. The default is 1.013. + external_P : float, optional + External pressure, typically atmospheric pressure [bar]. The default is 1.013. + pipe_resistance : float, optional + Biogas extraction pipe resistance [m3/d/bar]. The default is 5.0e4. + fixed_headspace_P : bool, optional + Whether to assume fixed headspace pressure. The default is False. + retain_cmps : Iterable[str], optional + IDs of the components that are assumed to be retained in the reactor, ideally. + The default is (). + fraction_retain : float, optional + The assumed fraction of ideal retention of select components. The default is 0.95. + + References + ---------- + .. [1] Batstone, D. J.; Keller, J.; Angelidaki, I.; Kalyuzhnyi, S. V; + Pavlostathis, S. G.; Rozzi, A.; Sanders, W. T. M.; Siegrist, H.; + Vavilin, V. A. The IWA Anaerobic Digestion Model No 1 (ADM1). + Water Sci. Technol. 2002, 45 (10), 65–73. + .. [2] Rosen, C.; Jeppsson, U. Aspects on ADM1 Implementation within + the BSM2 Framework; Lund, 2006. + ''' + + _N_ins = 1 + _N_outs = 2 + _ins_size_is_fixed = False + _outs_size_is_fixed = False + _R = 8.3145e-2 # Universal gas constant, [bar/M/K] + + def __init__(self, ID='', ins=None, outs=(), thermo=None, + init_with='WasteStream', V_liq=3400, V_gas=300, model=None, + T=308.15, headspace_P=1.013, external_P=1.013, + pipe_resistance=5.0e4, fixed_headspace_P=False, + retain_cmps=(), fraction_retain=0.95, + isdynamic=True, exogenous_vars=(), **kwargs): + if len(exogenous_vars) == 0: + exogenous_vars = (EDV('T', function=lambda t: T), ) + super().__init__(ID=ID, ins=ins, outs=outs, thermo=thermo, + init_with=init_with, V_max=V_liq, aeration=None, + DO_ID=None, suspended_growth_model=None, + isdynamic=isdynamic, exogenous_vars=exogenous_vars, **kwargs) + self.V_gas = V_gas + self.T = T + # self._S_gas = None + self._q_gas = 0 + self._n_gas = None + self._gas_cmp_idx = None + self._state_keys = None + self._S_vapor = None + self.model = model + self._biogas = WasteStream(phase='g') + self.headspace_P = headspace_P + self.external_P = external_P + self.pipe_resistance = pipe_resistance + self.fixed_headspace_P = fixed_headspace_P + self._f_retain = np.array([fraction_retain if cmp.ID in retain_cmps \ + else 0 for cmp in self.components]) + self._mixed = WasteStream() + self._tempstate = [] + + def ideal_gas_law(self, p=None, S=None): + '''Calculates partial pressure [bar] given concentration [M] at + operation temperature or vice versa according to the ideal gas law .''' + # p in bar, S in M + if p: return p/self._R/self.T + elif S: return S*self._R*self.T + + def p_vapor(self, convert_to_bar=True): + '''Calculates the saturated vapor pressure at operation temperature.''' + p = self.components.H2O.Psat(self.T) + if convert_to_bar: + return p*auom('Pa').conversion_factor('bar') + else: return p + + @property + def DO_ID(self): + '''Not applicable.''' + return None + @DO_ID.setter + def DO_ID(self, doid): + '''Does nothing.''' + pass + + @property + def aeration(self): + '''Not applicable''' + return None + @aeration.setter + def aeration(self, ae): + '''Does nothing.''' + pass + + V_liq = property(CSTR.V_max.fget) + @V_liq.setter + def V_liq(self, V): + '''[float] The liquid-phase volume, in m^3.''' + CSTR.V_max.fset(self, V) + + model = property(CSTR.suspended_growth_model.fget) + @model.setter + def model(self, model): + '''[:class:`CompiledProcesses` or NoneType] Anaerobic digestion model.''' + CSTR.suspended_growth_model.fset(self, model) + if model is not None: + #!!! how to make unit conversion generalizable to all models? + self._S_vapor = self.ideal_gas_law(p=self.p_vapor()) + self._n_gas = len(model._biogas_IDs) + self._state_keys = list(self.components.IDs) \ + + [ID+'_gas' for ID in self.model._biogas_IDs] \ + + ['Q'] + self._gas_cmp_idx = self.components.indices(self.model._biogas_IDs) + self._state_header = self._state_keys + + + split = property(CSTR.split.fget) + @split.setter + def split(self, split): + if split is None: self._split = split + else: + if len(split) != len(self._outs)-1: + raise ValueError('split and outs must have the same size') + self._split = np.array(split)/sum(split) + + @property + def headspace_P(self): + '''Headspace total pressure [bar].''' + return self._P_gas + @headspace_P.setter + def headspace_P(self, P): + self._P_gas = P + + @property + def external_P(self): + '''External (atmospheric) pressure [bar].''' + return self._P_atm + @external_P.setter + def external_P(self, P): + self._P_atm = P + + @property + def pipe_resistance(self): + '''Gas pipe resistance coefficient [m3/d/bar].''' + return self._k_p + @pipe_resistance.setter + def pipe_resistance(self, k): + self._k_p = k + + @property + def fixed_headspace_P(self): + '''Headspace total pressure [bar].''' + return self._fixed_P_gas + @fixed_headspace_P.setter + def fixed_headspace_P(self, b): + self._fixed_P_gas = bool(b) + + def set_retention_efficacy(self, i): + if i < 0 or i > 1: + raise ValueError('retention efficacy must be within [0,1]') + self._f_retain = (self._f_retain > 0) * i + + @property + def state(self): + '''The state of the anaerobic CSTR, including component concentrations [kg/m3], + biogas concentrations in the headspace [M biogas], and liquid flow rate [m^3/d].''' + if self._state is None: return None + else: + return dict(zip(self._state_keys, self._state)) + + @state.setter + def state(self, arr): + arr = np.asarray(arr) + n_state = len(self._state_keys) + if arr.shape != (n_state, ): + raise ValueError(f'state must be a 1D array of length {n_state}') + self._state = arr + + def _run(self): + '''Only to converge volumetric flows.''' + mixed = self._mixed # avoid creating multiple new streams + mixed.mix_from(self.ins) + # mixed.mix_from(self.ins, energy_balance=False) + if self.split is None: + gas, liquid = self.outs + liquid.copy_like(mixed) + liquid.T = self.T + # self._rQ = liquid.F_vol / mixed.F_vol + else: + gas = self.outs[0] + liquids = self._outs[1:] + Q = mixed.F_vol # m3/hr + for liquid, spl in zip(liquids, self.split): + liquid.copy_like(mixed) + liquid.set_total_flow(Q*spl, 'm3/hr') + liquid.T = self.T + # self._rQ = sum([ws.F_vol for ws in liquids]) / mixed.F_vol + gas.copy_like(self._biogas) + gas.T = self.T + if self._fixed_P_gas: + gas.P = self.headspace_P * auom('bar').conversion_factor('Pa') + + def _init_state(self): + mixed = self._mixed + Q = mixed.get_total_flow('m3/d') + #!!! how to make unit conversion generalizable to all models? + if self._concs is not None: Cs = self._concs * 1e-3 # mg/L to kg/m3 + else: Cs = mixed.conc * 1e-3 # mg/L to kg/m3 + self._state = np.append(Cs, [0]*self._n_gas + [Q]).astype('float64') + self._dstate = self._state * 0. + + def _update_state(self): + y = self._state + f_rtn = self._f_retain + i_mass = self.components.i_mass + chem_MW = self.components.chem_MW + n_cmps = len(self.components) + Cs = y[:n_cmps]*(1-f_rtn)*1e3 # kg/m3 to mg/L + if self.split is None: + gas, liquid = self._outs + if liquid.state is None: + liquid.state = np.append(Cs, y[-1]) + else: + liquid.state[:n_cmps] = Cs + liquid.state[-1] = y[-1] + else: + gas = self._outs[0] + liquids = self._outs[1:] + for liquid, spl in zip(liquids, self.split): + if liquid.state is None: + liquid.state = np.append(Cs, y[-1]*spl) + else: + liquid.state[:n_cmps] = Cs + liquid.state[-1] = y[-1]*spl + if gas.state is None: + gas.state = np.zeros(n_cmps+1) + gas.state[self._gas_cmp_idx] = y[n_cmps:(n_cmps + self._n_gas)] + gas.state[self.components.index('H2O')] = self._S_vapor + gas.state[-1] = self._q_gas + gas.state[:n_cmps] = gas.state[:n_cmps] * chem_MW / i_mass * 1e3 # i.e., M biogas to mg (measured_unit) / L + + def _update_dstate(self): + self._tempstate = self.model.rate_function._params['root'].data.copy() + dy = self._dstate + f_rtn = self._f_retain + n_cmps = len(self.components) + dCs = dy[:n_cmps]*(1-f_rtn)*1e3 + if self.split is None: + gas, liquid = self._outs + if liquid.dstate is None: + liquid.dstate = np.append(dCs, dy[-1]) + else: + liquid.dstate[:n_cmps] = dCs + liquid.dstate[-1] = dy[-1] + else: + gas = self._outs[0] + liquids = self._outs[1:] + for liquid, spl in zip(liquids, self.split): + if liquid.dstate is None: + liquid.dstate = np.append(dCs, dy[-1]*spl) + else: + liquid.dstate[:n_cmps] = dCs + liquid.dstate[-1] = dy[-1]*spl + if gas.dstate is None: + # contains no info on dstate + gas.dstate = np.zeros(n_cmps+1) + + + def f_q_gas_fixed_P_headspace(self, rhoTs, S_gas, T): + cmps = self.components + gas_mass2mol_conversion = (cmps.i_mass / cmps.chem_MW)[self._gas_cmp_idx] + self._q_gas = self._R*T/(self._P_gas-self.p_vapor(convert_to_bar=True))\ + *self.V_liq*sum(rhoTs*gas_mass2mol_conversion) + return self._q_gas + + def f_q_gas_var_P_headspace(self, rhoTs, S_gas, T): + p_gas = S_gas * self._R * T + self._P_gas = P = sum(p_gas) + self.p_vapor(convert_to_bar=True) + self._q_gas = max(0, self._k_p * (P - self._P_atm)) + return self._q_gas + + @property + def ODE(self): + if self._ODE is None: + self._compile_ODE() + return self._ODE + + def _compile_ODE(self): + if self._model is None: + CSTR._compile_ODE(self) + else: + cmps = self.components + f_rtn = self._f_retain + _dstate = self._dstate + _update_dstate = self._update_dstate + _f_rhos = self.model.rate_function + _f_param = self.model.params_eval + _M_stoichio = self.model.stoichio_eval + n_cmps = len(cmps) + n_gas = self._n_gas + V_liq = self.V_liq + V_gas = self.V_gas + gas_mass2mol_conversion = (cmps.i_mass / cmps.chem_MW)[self._gas_cmp_idx] + hasexo = bool(len(self._exovars)) + f_exovars = self.eval_exo_dynamic_vars + # _rQ = self._rQ + if self._fixed_P_gas: + f_qgas = self.f_q_gas_fixed_P_headspace + else: + f_qgas = self.f_q_gas_var_P_headspace + def dy_dt(t, QC_ins, QC, dQC_ins): + S_liq = QC[:n_cmps] + S_gas = QC[n_cmps: (n_cmps+n_gas)] + #!!! Volume change due to temperature difference accounted for + # in _run and _init_state + # Q = QC[-1] + # S_in = QC_ins[0,:-1] * 1e-3 # mg/L to kg/m3 + # Q_in = QC_ins[0,-1] + Q_ins = QC_ins[:, -1] + S_ins = QC_ins[:, :-1] * 1e-3 # mg/L to kg/m3 + Q = sum(Q_ins) + if hasexo: + exo_vars = f_exovars(t) + QC = np.append(QC, exo_vars) + T = exo_vars[0] + else: T = self.T + _f_param(QC) + M_stoichio = _M_stoichio() + rhos =_f_rhos(QC) + _dstate[:n_cmps] = (Q_ins @ S_ins - Q*S_liq*(1-f_rtn))/V_liq \ + + np.dot(M_stoichio.T, rhos) + q_gas = f_qgas(rhos[-3:], S_gas, T) + _dstate[n_cmps: (n_cmps+n_gas)] = - q_gas*S_gas/V_gas \ + + rhos[-3:] * V_liq/V_gas * gas_mass2mol_conversion + _dstate[-1] = dQC_ins[0,-1] #* _rQ + _update_dstate() + self._ODE = dy_dt + + def get_retained_mass(self, biomass_IDs): + cmps = self.components + mass = cmps.i_mass * self._state[:len(cmps)] * 1e3 # kg/m3 to mg/L + return self._V_max * mass[cmps.indices(biomass_IDs)].sum() + + def _design(self): + inf = self.ins[0] + T_in = inf.T + T_target = self.T + if T_target > T_in: + unit_duty = inf.F_mass * inf.Cp * (T_target - T_in) #kJ/hr + self.add_heat_utility(unit_duty, T_in, T_out=T_target, + heat_transfer_efficiency=0.8) + + +# %% + +ad_path = ospath.join(data_path, 'sanunit_data/_anaerobic_digestion.tsv') + +class AnaerobicDigestion(SanUnit, Decay): + ''' + Anaerobic digestion of wastes with the production of biogas based on + `Trimmer et al. `_ + + To enable life cycle assessment, the following impact items should be pre-constructed: + `Concrete`, `Excavation`. + + Cost is calculated by the unit cost of the impact items and their quantities. + + Parameters + ---------- + ins : Iterable + Waste for treatment. + outs : Iterable + Treated waste, captured biogas, fugitive CH4, and fugitive N2O. + flow_rate : float + Total flow rate through the reactor (for sizing purpose), [m3/d]. + If not provided, will use F_vol_in. + + Examples + -------- + `bwaise systems `_ + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.processes.Decay ` + ''' + _N_ins = 1 + _N_outs = 4 + _run = Decay._first_order_run + _units = { + 'Volumetric flow rate': 'm3/hr', + 'Residence time': 'd', + 'Single reactor volume': 'm3', + 'Reactor diameter': 'm', + 'Reactor height': 'm' + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + include_construction=True, + flow_rate=None, degraded_components=('OtherSS',), + if_capture_biogas=True, if_N2O_emission=False, + **kwargs): + Decay.__init__(self, ID, ins, outs, thermo, init_with, F_BM_default=1, + include_construction=include_construction, + degraded_components=degraded_components, + if_capture_biogas=if_capture_biogas, + if_N2O_emission=if_N2O_emission,) + self._flow_rate = flow_rate + + data = load_data(path=ad_path) + for para in data.index: + value = float(data.loc[para]['expected']) + setattr(self, '_'+para, value) + del data + + for attr, value in kwargs.items(): + setattr(self, attr, value) + + def _init_lca(self): + self.construction = [ + Construction('concrete', linked_unit=self, item='Concrete', quantity_unit='m3'), + Construction('excavation', linked_unit=self, item='Excavation', quantity_unit='m3'), + ] + + + def _design(self): + design = self.design_results + design['Volumetric flow rate'] = Q = self.flow_rate + design['Residence time'] = tau = self.tau + design['Reactor number'] = N = self.N_reactor + V_tot = Q * tau*24 + + # One extra as a backup + design['Single reactor volume'] = V_single = V_tot/(1-self.headspace_frac)/(N-1) + + # Rx modeled as a cylinder + design['Reactor diameter'] = D = (4*V_single*self.aspect_ratio/pi)**(1/3) + design['Reactor height'] = H = self.aspect_ratio * D + + if self.include_construction: + constr = self.construction + concrete = N*self.concrete_thickness*(2*pi/4*(D**2)+pi*D*H) + constr[0].quantity = concrete + constr[1].quantity = V_tot # excavation + + self.add_construction() + + + @property + def flow_rate(self): + ''' + [float] Total flow rate through the reactor (for sizing purpose), [m3/d]. + If not provided, will calculate based on F_vol_in. + ''' + return self._flow_rate if self._flow_rate else self.F_vol_in*24 + @flow_rate.setter + def flow_rate(self, i): + self._flow_rate = i + + @property + def tau(self): + '''[float] Residence time, [d].''' + return self._tau + @tau.setter + def tau(self, i): + self._tau = i + + @property + def COD_removal(self): + '''[float] Fraction of COD removed during treatment.''' + return self._COD_removal + @COD_removal.setter + def COD_removal(self, i): + self._COD_removal = i + + @property + def N_reactor(self): + '''[int] Number of reactors, float will be converted to the smallest integer.''' + return self._N_reactor + @N_reactor.setter + def N_reactor(self, i): + self._N_reactor = ceil(i) + + @property + def aspect_ratio(self): + '''[float] Diameter-to-height ratio of the reactor.''' + return self._aspect_ratio + @aspect_ratio.setter + def aspect_ratio(self, i): + self._aspect_ratio = i + + @property + def headspace_frac(self): + '''[float] Fraction of the reactor volume for headspace gas.''' + return self._headspace_frac + @headspace_frac.setter + def headspace_frac(self, i): + self._headspace_frac = i + + @property + def concrete_thickness(self): + '''[float] Thickness of the concrete wall.''' + return self._concrete_thickness + @concrete_thickness.setter + def concrete_thickness(self, i): + self._concrete_thickness = i + + +# %% + +F_BM_pump = 1.18*(1+0.007/100) # 0.007 is for miscellaneous costs +default_F_BM = { + 'Pump': F_BM_pump, + 'Pump building': F_BM_pump, + } +default_equipment_lifetime = { + 'Pump': 15, + 'Pump pipe stainless steel': 15, + 'Pump stainless steel': 15, + } + +class SludgeDigester(SanUnit): + ''' + A conventional digester for anaerobic digestion of sludge as in + `Shoener et al. `_. + + Note that the `CompiledComponents` object set in system simulation must + have defined `active_biomass`. + + Parameters + ---------- + ins : Iterable(stream) + Sludge for digestion. + outs : Iterable(stream) + Digested sludge, generated biogas. + HRT : float + Hydraulic retention time, [d]. + SRT : float + Solids retention time, [d]. + T : float + Temperature within the digester, [K]. + Y : float + Biomass yield, [mg VSS/mg BOD]. + b : float + Endogenous decay coefficient, [1/d]. + organics_conversion : float + Conversion of the organics (i.e., COD) of the sludge in fraction (i.e., 0.7 for 70%). + COD_factor : float + Biomass-to-COD conversion factor, [g COD/g VSS]. + methane_yield : float + Methane yield from the digested organics, [m3/kg]. + methane_fraction : float + Fraction of methane in the biogas, the rest is assumed to be CO2. + depth : float + Side depth of the digester, [m]. + heat_transfer_coeff : dict + Heat transfer coefficients for heat loss calculation, [W/m2/°C], + keys should contain "wall", "floor", and "ceiling". + wall_concrete_unit_cost : float + Unit cost of the wall concrete, [UDS/ft3]. + slab_concrete_unit_cost : float + Unit cost of the slab concrete, [UDS/ft3]. + excavation_unit_cost : float + Unit cost of the excavation activity, [UDS/ft3]. + + References + ---------- + [1] Shoener, B. D.; Zhong, C.; Greiner, A. D.; Khunjar, W. O.; Hong, P.-Y.; Guest, J. S. + Design of Anaerobic Membrane Bioreactors for the Valorization + of Dilute Organic Carbon Waste Streams. + Energy Environ. Sci. 2016, 9 (3), 1102–1112. + https://doi.org/10.1039/C5EE03715H. + + ''' + _N_outs = 2 + + # All in K + _T_air = 17 + 273.15 + _T_earth = 10 + 273.15 + + # All in ft + _freeboard = 3 + _t_wall = 6/12 + _t_slab = 8/12 + + # Pump building, all in ft + _L_PB = 50 + _W_PB = 30 + _D_PB = 10 + + # Excavation + _excav_slope = 1.5 # horizontal/vertical + _constr_access = 3 # ft + + auxiliary_unit_names = ('heat_exchanger',) + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + include_construction=False, F_BM_default=1, + HRT=20, SRT=20, T=35+273.15, Y=0.08, b=0.03, + organics_conversion=0.7, COD_factor=1.42, + methane_yield=0.4, methane_fraction=0.65, + depth=10, + heat_transfer_coeff=dict(wall=0.7, floor=1.7, ceiling=0.95), + wall_concrete_unit_cost=24, # from $650/yd3 + slab_concrete_unit_cost=13, # from $350/yd3 + excavation_unit_cost=0.3, # from $8/yd3 + F_BM=default_F_BM, lifetime=default_equipment_lifetime, + **kwargs): + SanUnit.__init__(self, ID, ins, outs, thermo, init_with, F_BM_default=1, + include_construction=include_construction) + self.HRT = HRT + self.SRT = SRT + self.T = T + self.Y = Y + self.b = b + self.organics_conversion = organics_conversion + self.COD_factor = COD_factor + self.methane_yield = methane_yield + self.methane_fraction = methane_fraction + self.depth = depth + self.heat_transfer_coeff = heat_transfer_coeff + ID = self.ID + hx_in = Stream(f'{ID}_hx_in') + hx_out = Stream(f'{ID}_hx_out') + self.heat_exchanger = HXutility(ID=f'{ID}_hx', ins=hx_in, outs=hx_out) + self.wall_concrete_unit_cost = wall_concrete_unit_cost + self.slab_concrete_unit_cost = slab_concrete_unit_cost + self.excavation_unit_cost = excavation_unit_cost + self.F_BM.update(F_BM) + self._default_equipment_lifetime.update(lifetime) + self.sludge_pump = WWTpump( + ID=f'{ID}_sludge', ins=self.ins[0].proxy(), pump_type='', + Q_mgd=None, add_inputs=(1,), capacity_factor=1., + include_pump_cost=True, include_building_cost=False, + include_OM_cost=False) + + + def _run(self): + sludge, = self.ins + digested, biogas = self.outs + digested.T = biogas.T = self.T + biogas.phase = 'g' + + # Biogas production estimation based on Example 13-5 of Metcalf & Eddy, 5th edn. + Y, b, SRT = self.Y, self.b, self.SRT + organics_conversion, COD_factor = self.organics_conversion, self.COD_factor + methane_yield, methane_fraction = self.methane_yield, self.methane_fraction + tot = sludge.imass['active_biomass'] + tot = tot if isinstance(tot, (int, float)) else tot.sum() + biomass_COD = tot*1e3*24*1.42 # [g/d], 1.42 converts VSS to COD + + digested.mass = sludge.mass + digested.imass['active_biomass'] = 0 # biomass-derived COD calculated separately + substrate_COD = digested.COD*24*digested.F_vol # [g/d] + + tot_COD = biomass_COD + substrate_COD # [g/d] + + digestion_yield = Y*tot_COD*organics_conversion/(1+b*SRT) # [g/d] + methane_vol = methane_yield*tot_COD*organics_conversion - COD_factor*digestion_yield + + # Update stream flows + digested.imass['substrates'] *= (1-organics_conversion) + digested.imass['active_biomass'] = \ + sludge.imass['active_biomass']*(1-organics_conversion) + + biogas.empty() + biogas.ivol['CH4'] = methane_vol + biogas.ivol['CO2'] = methane_vol/methane_fraction*(1-methane_fraction) + + + _units = { + 'HRT': 'd', + 'SRT': 'd', + 'Volume': 'm3', + 'Surface area': 'm2', + 'Diameter': 'm', + 'Wall concrete': 'ft3', + 'Slab concrete': 'ft3', + 'Excavation': 'ft3', + 'Pump pipe stainless steel': 'kg', + 'Pump stainless steel': 'kg', + } + def _design(self): + design = self.design_results + sludge, = self.ins + Q = sludge.F_vol * 24 # from m3/hr to m3/d + + # Dimensions + design['SRT'] = self.SRT + HRT = design['HRT'] = self.HRT + V = design['Volume'] = Q * HRT # m3 + depth = design['depth'] = self.depth # m + A = design['Surface area'] = V / depth # m2 + dia = design['Diameter']= (A*4/pi) ** 0.5 # m + + # Calculate needed heating + T = self.T + hx = self.heat_exchanger + hx_ins0, hx_outs0 = hx.ins[0], hx.outs[0] + hx_ins0.copy_flow(sludge) + hx_outs0.copy_flow(sludge) + hx_ins0.T = sludge.T + hx_outs0.T = T + hx_ins0.P = hx_outs0.P = sludge.P + + # Heat loss + coeff = self.heat_transfer_coeff + A_wall = pi * dia * depth + wall_loss = coeff['wall'] * A_wall * (T-self.T_air) # [W] + floor_loss = coeff['floor'] * A * (T-self.T_earth) # [W] + ceiling_loss = coeff['ceiling'] * A * (T-self.T_air) # [W] + duty = (wall_loss+floor_loss+ceiling_loss)*60*60/1e3 # kJ/hr + hx.H = hx_ins0.H + duty # stream heating and heat loss + hx.simulate_as_auxiliary_exchanger(ins=hx.ins, outs=hx.outs) + + # Concrete usage + ft_2_m = auom('ft').conversion_factor('m') + design['Wall concrete'] = self.t_wall * pi*(dia*ft_2_m)*(depth*ft_2_m+self.freeboard) + design['Slab concrete'] = 2 * self.t_slab * A*(ft_2_m**2) # floor and ceiling + + # Excavation + design['Excavation'] = calculate_excavation_volume( + self.L_PB, self.W_PB, self.D_PB, self.excav_slope, self.constr_access) + + # Pump + sludge_pump = self.sludge_pump + sludge_pump.simulate() + design.update(sludge_pump.design_results) + + def _cost(self): + D, C = self.design_results, self.baseline_purchase_costs + # F_BM, lifetime = self.F_BM, self._default_equipment_lifetime + C['Wall concrete'] = D['Wall concrete'] * self.wall_concrete_unit_cost + C['Slab concrete'] = D['Slab concrete'] * self.slab_concrete_unit_cost + C['Excavation'] = D['Excavation'] * self.excavation_unit_cost + sludge_pump = self.sludge_pump + C.update(sludge_pump.baseline_purchase_costs) + self.power_utility.rate = sludge_pump.power_utility.rate + + + @property + def T_air(self): + '''[float] Temperature of the air, [K].''' + return self._T_air + @T_air.setter + def T_air(self, i): + self._T_air = i + + @property + def T_earth(self): + '''[float] Temperature of the air, [K].''' + return self._T_earth + @T_earth.setter + def T_earth(self, i): + self._T_earth = i + + @property + def freeboard(self): + '''[float] Freeboard added to the depth of the reactor tank, [ft].''' + return self._freeboard + @freeboard.setter + def freeboard(self, i): + self._freeboard = i + + @property + def t_wall(self): + '''[float] Concrete wall thickness, [ft].''' + return self._t_wall + @t_wall.setter + def t_wall(self, i): + self._t_wall = i + + @property + def t_slab(self): + ''' + [float] Concrete slab thickness, [ft], + default to be 2 in thicker than the wall thickness. + ''' + return self._t_slab or self.t_wall+2/12 + @t_slab.setter + def t_slab(self, i): + self._t_slab = i + + @property + def L_PB(self): + '''[float] Length of the pump building, [ft].''' + return self._L_PB + @L_PB.setter + def L_PB(self, i): + self._L_PB = i + + @property + def W_PB(self): + '''[float] Width of the pump building, [ft].''' + return self._W_PB + @W_PB.setter + def W_PB(self, i): + self._W_PB = i + + @property + def D_PB(self): + '''[float] Depth of the pump building, [ft].''' + return self._D_PB + @D_PB.setter + def D_PB(self, i): + self._D_PB = i + + @property + def excav_slope(self): + '''[float] Slope for excavation (horizontal/vertical).''' + return self._excav_slope + @excav_slope.setter + def excav_slope(self, i): + self._excav_slope = i + + @property + def constr_access(self): + '''[float] Extra room for construction access, [ft].''' + return self._constr_access + @constr_access.setter + def constr_access(self, i): + self._constr_access = i \ No newline at end of file diff --git a/qsdsan/sanunits/_membrane_bioreactor.py b/qsdsan/sanunits/_membrane_bioreactor.py new file mode 100644 index 00000000..99603c57 --- /dev/null +++ b/qsdsan/sanunits/_membrane_bioreactor.py @@ -0,0 +1,1342 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +''' +QSDsan: Quantitative Sustainable Design for sanitation and resource recovery systems + +This module is developed by: + Yalin Li + +This module is under the University of Illinois/NCSA Open Source License. +Please refer to https://github.com/QSD-Group/QSDsan/blob/main/LICENSE.txt +for license details. +''' + +from math import ceil, floor +from biosteam import Stream +from biosteam.exceptions import DesignError +from . import HXutility, WWTpump, InternalCirculationRx +from .. import SanStream, SanUnit +from ..equipments import Blower +from ..utils import ( + auom, + compute_stream_COD, + format_str, + default_component_dict, + calculate_excavation_volume, + ) + +__all__ = ('AnMBR',) + +degassing = SanStream.degassing + +_ft_to_m = auom('ft').conversion_factor('m') +_ft2_to_m2 = auom('ft2').conversion_factor('m2') +_ft3_to_m3 = auom('ft3').conversion_factor('m3') +_ft3_to_gal = auom('ft3').conversion_factor('gallon') +_m3_to_gal = auom('m3').conversion_factor('gal') +_cmh_to_mgd = _m3_to_gal * 24 / 1e6 # cubic meter per hour to million gallon per day +_lb_to_kg = auom('lb').conversion_factor('kg') + +F_BM_pump = 1.18*(1+0.007/100) # 0.007 is for miscellaneous costs +default_F_BM = { + 'Membrane': 1+0.15, # assume 15% for replacement labor + 'Pumps': F_BM_pump, + 'Pump building': F_BM_pump, + } +default_equipment_lifetime = { + 'Membrane': 10, + 'Pumps': 15, + 'Pump pipe stainless steel': 15, + 'Pump stainless steel': 15, + 'Pump chemical storage HDPE': 30, + } + + +# %% + +class AnMBR(SanUnit): + ''' + Anaerobic membrane bioreactor (AnMBR) for wastewater treatment as in + Shoener et al. [1]_ Some assumptions adopted from Humbird et al. [2]_ + + In addition to the anaerobic treatment, an optional second stage can be added, + which can be aerobic filter or granular activated carbon (GAC). + + Parameters + ---------- + ins : Inlets(obj) + Influent, recycle (optional), naocl, citric acid, bisulfite, air (optional). + outs : Outlets(obj) + Biogas, effluent, waste sludge, air (optional). + reactor_type : str + Can either be "CSTR" for continuous stirred tank reactor + or "AF" for anaerobic filter. + N_train : int + Number of treatment train, should be at least two in case one failing. + membrane_configuration : str + Can either be "cross-flow" or "submerged". + membrane_type : str + Can be "hollow fiber" ("submerged" configuration only), + "flat sheet" (either "cross-flow" or "submerged" configuration), + or "multi-tube" ("cross-flow" configuration only). + membrane_material : str + Can be any of the plastics ("PES", "PVDF", "PET", "PTFE") + for any of the membrane types ("hollow fiber", "flat sheet", "multi-tube"), + or "sintered steel" for "flat sheet", + or "ceramic" for "multi-tube". + membrane_unit_cost : float + Cost of membrane, [$/ft2] + include_aerobic_filter : bool + Whether to include an aerobic filtration process in this AnMBR, + can only be True in "AF" (not "CSTR") reactor. + add_GAC : bool + If to add granular activated carbon to enhance biomass retention, + can only be True for the "submerged" configuration. + include_degassing_membrane : bool + If to include a degassing membrane to enhance methane + (generated through the digestion reaction) recovery. + Y_biogas : float + Biogas yield, [kg biogas/kg consumed COD]. + Y_biomass : float + Biomass yield, [kg biomass/kg consumed COD]. + biodegradability : float or dict + Biodegradability of components, + when shown as a float, all biodegradable components are assumed to have + the same degradability. + solids : Iterable(str) + IDs of the solid components. + If not provided, will be set to the default `solids` attribute of the components. + split : dict + Component-wise split to the treated water. + E.g., {'NaCl':1, 'WWTsludge':0} indicates all of the NaCl goes to + the treated water and all of the WWTsludge goes to the wasted sludge. + Default splits (based on the membrane bioreactor in [2]_) will be used + if not provided. + Note that the split for `Water` will be ignored as it will be adjusted + to satisfy the `solids_conc` setting. + biomass_ID: str + ID of the Component that represents the biomass. + solids_conc : float + Concentration of the biomass in the waste sludge, [g/L]. + T : float + Temperature of the reactor. + Will not control temperature if provided as None. + kwargs : dict + Other keyword arguments (e.g., J_max, SGD). + + References + ---------- + .. [1] Shoener et al., Design of Anaerobic Membrane Bioreactors for the + Valorization of Dilute Organic Carbon Waste Streams. + Energy Environ. Sci. 2016, 9 (3), 1102–1112. + https://doi.org/10.1039/C5EE03715H. + .. [2] Humbird et al., Process Design and Economics for Biochemical Conversion of + Lignocellulosic Biomass to Ethanol: Dilute-Acid Pretreatment and Enzymatic + Hydrolysis of Corn Stover; Technical Report NREL/TP-5100-47764; + National Renewable Energy Lab (NREL), 2011. + https://www.nrel.gov/docs/fy11osti/47764.pdf + + .. note:: + + TODO + - Add algorithms for other configurations (AF, submerged, sparging, GAC, flat sheet, hollow fiber) + - Maybe add AeMBR as well (and make an MBR superclass) + + - AeMBR can use higher flux and allows for lower transmembrane pressure + + See Also + -------- + `MATLAB codes `_ used in ref 1, + especially the system layout `diagrams `_. + ''' + _N_ins = 6 # influent, recycle (optional), naocl, citric acid, bisulfite, air (optional) + _N_outs = 4 # biogas, effluent, waste sludge, air (optional) + + # Equipment-related parameters + _cas_per_tank_spare = 2 + + _mod_surface_area = { + 'hollow fiber': 370, + 'flat sheet': 1.45/_ft2_to_m2, + 'multi-tube': 32/_ft2_to_m2 + } + + _mod_per_cas = None + _mod_per_cas_range = { + 'hollow fiber': (30, 48), # min, max + 'flat sheet': (150, 200), + 'multi-tube': (44, 48) + } + + _cas_per_tank = None + _cas_per_tank_range = (16, 22) + + _N_blower = 0 + + _W_tank = 21 # ft + _D_tank = 12 # ft + _freeboard = 2 # ft + _W_dist = 4.5 # ft + _W_eff = 4.5 # ft + + _L_well = 8 # ft + _W_well = 8 # ft + _D_well = 12 # ft + + _t_wall = None + _t_slab = None + + _excav_slope = 1.5 + _constr_access = 3 # ft + + # Operation-related parameters + _HRT = 10 # hr + _J_max = 12 + _TMP_dct = { + 'cross-flow': 2.5, + 'submerged': 2.5, + } + _TMP_aerobic = None + _recir_ratio = 2.25 # from the 0.5-4 uniform range in ref [1] + _v_cross_flow = 1.2 # from the 0.4-2 uniform range in ref [1] + _v_GAC = 8 + _SGD = 0.625 # from the 0.05-1.2 uniform range in ref [1] + _AFF = 3.33 + + # Heating + T_air = 17 + 273.15 + T_earth = 10 + 273.15 + # Heat transfer coefficients, all in W/m2/°C + heat_transfer_coeff=dict(wall=0.7, floor=1.7, ceiling=0.95), + + # Costs + excav_unit_cost = (8+0.3) / 27 # $/ft3, 27 is to convert from $/yd3 + wall_concrete_unit_cost = 650 / 27 # $/ft3 + slab_concrete_unit_cost = 350 / 27 # $/ft3 + GAC_price = 13.78 # $/kg + + _refresh_rxns = InternalCirculationRx._refresh_rxns + + # Other equipment + pumps = ('perm', 'retent', 'recir', 'sludge', 'naocl', 'citric', 'bisulfite', + 'AF', 'AeF') + auxiliary_unit_names = ('heat_exchanger',) + + _units = { + 'Total volume': 'ft3', + 'Wall concrete': 'ft3', + 'Slab concrete': 'ft3', + 'Excavation': 'ft3', + 'Membrane': 'm3', + 'Pump pipe stainless steel': 'kg', + 'Pump stainless steel': 'kg', + 'Pump chemical storage HDPE': 'm3', + 'Total air flow': 'CFM', + 'Blower capacity': 'CFM', + 'Packing LDPE': 'm3', + 'Packing HDPE': 'm3', + 'GAC': 'kg', + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, + init_with='WasteStream', *, + reactor_type='CSTR', + N_train=2, + membrane_configuration='cross-flow', + membrane_type='multi-tube', + membrane_material='ceramic', + membrane_unit_cost=8, + include_aerobic_filter=False, + add_GAC=False, + include_degassing_membrane=True, + Y_biogas = 0.86, + Y_biomass=0.05, # from the 0.02-0.08 uniform range in ref [1] + biodegradability=1.0, + solids=(), split={}, + biomass_ID='WWTsludge', solids_conc=10.5, + T=35+273.15, + F_BM=default_F_BM, lifetime=default_equipment_lifetime, + F_BM_default=1, **kwargs): + SanUnit.__init__(self, ID, ins, outs, thermo, init_with=init_with, F_BM_default=1) + self._inf = inf = self.ins[0].copy() # this stream will be preserved (i.e., no reaction) + # For pump design + self._retent = inf.copy(f'{ID}_rentent') + self._recir = inf.copy(f'{ID}_recir') + self.reactor_type = reactor_type + self.N_train = N_train + self.include_aerobic_filter = include_aerobic_filter + self.membrane_configuration = membrane_configuration + self.membrane_type = membrane_type + self.membrane_material = membrane_material + self.membrane_unit_cost = membrane_unit_cost + self.add_GAC = add_GAC + self.include_degassing_membrane = include_degassing_membrane + self.biodegradability = biodegradability + self.Y_biogas = Y_biogas + self.Y_biomass = Y_biomass + cmps = self.components + self.split = split if split else default_component_dict( + cmps=cmps, gas=0.15, solubles=0.125, solids=0) # ref[2] + self.solids = solids or cmps.solids + self._xcmp = getattr(self.components, biomass_ID) + self.solids_conc = solids_conc + self.T = T + self.F_BM.update(F_BM) + self._default_equipment_lifetime.update(lifetime) + + # Initialize the attributes + self.AF = self.AeF = None + hx_in = Stream(f'{ID}_hx_in') + hx_out = Stream(f'{ID}_hx_out') + self.heat_exchanger = HXutility(ID=f'{ID}_hx', ins=hx_in, outs=hx_out) + self._refresh_rxns() + + for k, v in kwargs.items(): setattr(self, k, v) + + blower = self.blower = Blower(ID+'_blower', linked_unit=self) + self.equipments = (blower,) + self._check_design() + + + def _check_design(self): + reactor_type = self.reactor_type + m_config = self.membrane_configuration + m_type = self.membrane_type + m_material = self.membrane_material + + if reactor_type == 'CSTR': + if self.include_aerobic_filter: + raise DesignError('Aerobic filtration cannot be used in CSTR.') + + if m_config == 'submerged': + if not m_type in ('hollow fiber', 'flat sheet'): + raise DesignError('Only "hollow fiber" or "flat sheet" is allowed ' + 'for "submerged" membrane, not {m_type}.') + else: # cross-flow + if not m_type in ('flat sheet', 'multi-tube'): + raise DesignError('Only "flat sheet" or "multi-tube" is allowed ' + 'for "cross-flow" membrane, not {m_type}.') + if self.add_GAC: + raise DesignError('No GAC should be added ' + '(i.e., `add_GAC` can only be False' + 'for "cross-flow" membrane.') + + plastics = ('PES', 'PVDF', 'PET', 'PTFE') + if m_type == 'hollow fiber': + if not m_material in plastics: + raise DesignError(f'Only plastic materials {plastics} ' + 'allowed for "hollow fiber" membrane', + f'not "{m_material}".') + elif m_type == 'flat sheet': + if not m_material in (*plastics, 'sintered steel'): + raise DesignError(f'Only plastic materials {plastics} and "sintered steel"' + 'allowed for "flat sheet" membrane', + f'not "{m_material}".') + else: # multi-tube + if not m_material in (*plastics, 'ceramic'): + raise DesignError(f'Only plastic materials {plastics} and "ceramic"' + 'allowed for "multi-tube" membrane', + f'not "{m_material}".') + + + # ========================================================================= + # _run + # ========================================================================= + def _run(self): + raw, recycled, naocl, citric, bisulfite, air_in = self.ins + biogas, perm, sludge, air_out = self.outs + + # Initialize the streams + biogas.phase = 'g' + biogas.empty() + + inf = self._inf + inf.mix_from((raw, recycled)) + + # Chemicals for cleaning, assume all chemicals will be used up + # 2.2 L/yr/cmd of 12.5 wt% solution (15% vol) + naocl.empty() + naocl.imass['NaOCl', 'Water'] = [0.125, 1-0.125] + naocl.F_vol = (2.2/1e3/365/24) * (inf.F_vol*24) # m3/hr solution + + # 0.6 L/yr/cmd of 100 wt% solution, 13.8 lb/kg + citric.empty() + citric.ivol['CitricAcid'] = (0.6/1e3/365/24) * (inf.F_vol*24) # m3/hr pure + + # 0.35 L/yr/cmd of 38% solution, 3.5 lb/gal + bisulfite.empty() + bisulfite.imass['Bisulfite', 'Water'] = [0.38, 1-0.38] + bisulfite.F_vol = (0.35/1e3/365/24) * (inf.F_vol*24) # m3/hr solution + + # For pump design + self._compute_mod_case_tank_N() + Q_R_mgd, Q_IR_mgd = self._compute_liq_flows() + retent, recir = self._retent, self._recir + retent.F_mass *= Q_R_mgd / self.Q_mgd + recir.F_mass *= Q_IR_mgd / self.Q_mgd + self._retent, self._recir = retent, recir + + # Effluents + self.growth_rxns(inf.mol) + self.biogas_rxns(inf.mol) + inf.split_to(perm, sludge, self._isplit.data) + + solids_conc = self._solids_conc + m_solids = sludge.imass[self.solids].sum() + if m_solids/sludge.F_vol <= solids_conc: + diff = sludge.ivol['Water'] - m_solids/solids_conc + sludge.ivol['Water'] = m_solids/solids_conc + perm.ivol['Water'] += diff + + degassing(perm, biogas) + degassing(sludge, biogas) + + # Gas for sparging, no sparging needed if submerged or using GAC + air_out.link_with(air_in) + air_in.T = 17 + 273.15 + + # self._design_blower() + self.add_equipment_design() + + if self.T is not None: perm.T = sludge.T = biogas.T = air_out.T = self.T + + + # Called by _run + def _compute_mod_case_tank_N(self): + N_mod_min, N_mod_max = self.mod_per_cas_range[self.membrane_type] + N_cas_min, N_cas_max = self.cas_per_tank_range + + mod_per_cas, cas_per_tank = N_mod_min, N_cas_min + + J, J_max, N_train = self.J, self.J_max, self._N_train_min + while J > J_max: + mod_per_cas += 1 + if mod_per_cas == N_mod_max + 1: + if cas_per_tank == N_cas_max + 1: + N_train += 1 + mod_per_cas, cas_per_tank = N_mod_min, N_cas_min + else: + cas_per_tank += 1 + mod_per_cas = N_mod_min + + self._N_train, self._mod_per_cas, self._cas_per_tank = \ + N_train, mod_per_cas, cas_per_tank + + + # Called by _run + def _compute_liq_flows(self): + m_type = self.membrane_type + if m_type == 'multi-tube': + # Cross-flow flow rate per module, + # based on manufacture specifications for compact 33, [m3/hr] + Q_cross_flow = 53.5 * self.v_cross_flow + Q_R_cmh = self.N_mod_tot * Q_cross_flow # total retentate flow rate, [m3/hr] + Q_R_mgd = Q_R_cmh * _cmh_to_mgd # [mgd] + + Q_mgd, recir_ratio = self.Q_mgd, self.recir_ratio + if Q_mgd*recir_ratio >= Q_R_mgd: + Q_IR_mgd = Q_mgd*recir_ratio - Q_R_mgd + else: + Q_IR_mgd = 0 + # # Gives really large recir_ration, + # # probably shouldn't back-calculate this way + # self._recir_ratio = Q_R_mgd / Q_mgd + + if self.add_GAC: + Q_upflow_req = (self.v_GAC/_ft_to_m) * \ + self.L_membrane_tank*self.W_tank*self.N_train * 24 * _ft3_to_gal / 1e6 + Q_IR_add_mgd = max(0, (Q_mgd+Q_IR_mgd)-Q_upflow_req) + Q_IR_mgd += Q_IR_add_mgd + self._recir_ratio = Q_IR_mgd / Q_mgd + + return Q_R_mgd, Q_IR_mgd + + + # ========================================================================= + # _design + # ========================================================================= + def _design(self): + D = self.design_results + D['Treatment train'] = self.N_train + D['Cassette per train'] = self.cas_per_tank + D['Module per cassette'] = self.mod_per_cas + D['Total membrane modules'] = self.N_mod_tot + + # Step A: Reactor and membrane tanks + # Call the corresponding design function + # (_design_CSTR or _design_AF) + func = getattr(self, f'_design_{self.reactor_type}') + + wall, slab, excavation = func() + D['Wall concrete'] = wall + D['Slab concrete'] = slab + D['Excavation'] = excavation + + # Optional addition of packing media (used in filters) + ldpe, hdpe = 0., 0. + for i in (self.AF, self.AeF): + if i is None: + continue + ldpe += i.design_results['Packing LDPE'] + hdpe += i.design_results['Packing HDPE'] + + # Optional addition of GAC + D['GAC'] = self._design_GAC() + + # Step B: Membrane + # Call the corresponding design function + # (_design_hollow_fiber, _design_flat_sheet, or _design_multi_tube) + m_type = format_str(self.membrane_type) + func = getattr(self, f'_design_{m_type}') + D['Membrane'] = func() + + # Step C: Pumps + pipe, pumps, hdpe = self._design_pump() + D['Pump pipe stainless steel'] = pipe + D['Pump stainless steel'] = pumps + D['Pump chemical storage HDPE'] = hdpe + + # Step D: Degassing membrane + D['Degassing membrane'] = self.N_degasser + + # Total volume + D['Total volume'] = self.V_tot + + + ### Step A functions ### + # Called by _design + def _design_CSTR(self): + N_train = self.N_train + W_dist, L_CSTR , W_eff, L_membrane_tank = \ + self.W_dist, self.L_CSTR, self.W_eff, self.L_membrane_tank + W_PB, W_BB, D_tank = self.W_PB, self.W_BB, self.D_tank + t_wall, t_slab = self.t_wall, self.t_slab + SL, CA = self.excav_slope, self.constr_access + + ### Concrete calculation ### + W_N_trains = (self.W_tank+2*t_wall)*N_train - t_wall*(N_train-1) + + D = D_tank + self.freeboard + t = t_wall + t_slab + + get_VWC = lambda L1, N: N * t_wall * L1 * D + get_VSC = lambda L2: t * L2 * W_N_trains + + # Concrete for distribution channel, [ft3] + VWC_dist = get_VWC(L1=(W_N_trains+W_dist), N=2) + VSC_dist = get_VSC(L2=(W_dist+2*t_wall)) + + # Concrete for CSTR tanks, [ft3] + VWC_CSTR = get_VWC(L1=L_CSTR, N=(N_train+1)) + VSC_CSTR = get_VSC(L2=L_CSTR) + + # Concrete for effluent channel, [ft3] + VWC_eff = get_VWC(L1=(W_N_trains+W_eff), N=2) + VSC_eff = get_VSC(L2=(W_eff+2*t_wall)) + + # Concrete for the pump/blower building, [ft3] + VWC_PBB = get_VWC(L1=(W_N_trains+W_PB+W_BB), N=2) + VSC_PBB = get_VSC(L2=(W_PB+t_wall+W_BB)) + + if self.membrane_configuration == 'submerged': + VWC_membrane_tank, VSC_membrane_tank, VWC_well, VSC_well = \ + self._design_membrane_tank(self, D, N_train, W_N_trains, + self.L_membrane_tank) + else: + VWC_membrane_tank, VSC_membrane_tank, VWC_well, VSC_well = 0., 0., 0., 0. + + # Total volume of wall concrete, [ft3] + VWC = VWC_dist + VWC_CSTR + VWC_eff + VWC_PBB + VWC_membrane_tank + VWC_well + + # Total volume of slab concrete [ft3] + VSC = VSC_dist + VSC_CSTR + VSC_eff + VSC_PBB + VSC_membrane_tank + VSC_well + + ### Excavation calculation ### + # Excavation volume for the reactor and membrane tanks, [ft3] + VEX_tanks = calculate_excavation_volume( + L=(W_dist+L_CSTR+W_eff+L_membrane_tank), + W=W_N_trains, D=D_tank, excav_slope=SL, constr_acess=CA) + + # Excavation volume for pump/blower building, [ft3] + VEX_PBB = calculate_excavation_volume( + L=(W_PB+W_BB), + W=W_N_trains, D=D_tank, excav_slope=SL, constr_acess=CA) + + VEX = VEX_tanks + VEX_PBB + + #!!! Need to add wet wells for submerged configurations + + return VWC, VSC, VEX + + + # Called by _design + def _design_AF(self): + '''NOT READY YET.''' + # Use FilterTank + + + # Called by _design_CSTR/_design_AF + def _design_membrane_tank(self, D, N_train, W_N_trains, L_membrane_tank, + t_wall, t_slab): + L_well, W_well, D_well = self.L_well, self.W_well, self.D_well + + # Concrete for membrane tanks, [ft3] + t = t_wall + t_slab + VWC_membrane_tank = (N_train+1) * t_wall * L_membrane_tank * D + VSC_membrane_tank = t * L_membrane_tank * W_N_trains + + # Concrete for wet well (mixed liquor storage), [ft3] + L = L_well + 2*t_wall + W = W_well + 2*t_wall + VWC_well = 2 * t_wall * (L_well+W) * D_well + VSC_well = (t_slab+t_wall) * L * W + + return VWC_membrane_tank, VSC_membrane_tank, VWC_well, VSC_well + + + # Called by _design + def _design_GAC(self): + '''NOT READY YET.''' + if not self.add_GAC: + return 0 + + M_GAC = 1 + return M_GAC + + + ### Step B functions ### + # Called by _design + def _design_hollow_fiber(self): + '''NOT READY YET.''' + + + # Called by _design + def _design_flat_sheet(self): + '''NOT READY YET.''' + + + # Called by _design + def _design_multi_tube(self): + # # 0.01478 is volume of material for each membrane tube, [m3] + # # L_tube OD ID + # V_tube = 3 * math.pi/4 * ((6e-3)**2-(5.2e-3)**2) + # V_SU = 700 * V_tube # V for each small unit [m3] + # M_SU = 1.78*1e3 * V_SU # mass = density*volume, [kg/m3], not used + return self.N_mod_tot*0.01478 + + + ### Step C function ### + # Called by _design + def _design_pump(self): + pumps = self.pumps + ID, ins, outs = self.ID, self.ins, self.outs + rx_type, m_config, pumps = \ + self.reactor_type, self.membrane_configuration, self.pumps + self.AF_pump = self.AF.lift_pump if self.AF else None + self.AeF_pump = self.AeF.lift_pump if self.AeF else None + #!!! Maybe move `ins_dct` to `__init__` so it won't be repeated + ins_dct = { + 'perm': outs[1].proxy(f'{ID}_perm'), + 'retent': self._retent, + 'recir': self._recir, + 'sludge': outs[2].proxy(f'{ID}_sludge'), + 'naocl': ins[2].proxy(f'{ID}_NaOCl'), + 'citric': ins[3].proxy(f'{ID}_citric'), + 'bisulfite': ins[4].proxy(f'{ID}_bisulfite'), + } + type_dct = { + 'perm': f'permeate_{m_config}', + 'retent': f'retentate_{rx_type}', + 'recir': f'recirculation_{rx_type}', + 'sludge': 'sludge', + 'naocl': 'chemical', + 'citric': 'chemical', + 'bisulfite': 'chemical', + } + inputs_dct = { + 'perm': (self.cas_per_tank, self.D_tank, self.TMP_anaerobic, + self.include_aerobic_filter), + 'retent': (self.cas_per_tank,), + 'recir': (1, self.L_CSTR,), + 'sludge': (1,), + 'naocl': (1,), + 'citric': (1,), + 'bisulfite': (1,), + } + + for i in pumps[:-2]: + if hasattr(self, f'{i}_pump'): + p = getattr(self, f'{i}_pump') + setattr(p, 'add_inputs', inputs_dct[i]) + else: + # Add '.' in ID for auxiliary units + ID = f'.{ID}_{i}' + capacity_factor=2. if i=='perm' else self.recir_ratio if i=='recir' else 1. + pump = WWTpump( + ID=ID, ins=ins_dct[i], pump_type=type_dct[i], + Q_mgd=None, add_inputs=inputs_dct[i], + capacity_factor=capacity_factor, + include_pump_cost=True, + include_building_cost=False, + include_OM_cost=False, + ) + setattr(self, f'{i}_pump', pump) + + pipe_ss, pump_ss, hdpe = 0., 0., 0. + for i in (*pumps, 'AF', 'AeF'): + p = getattr(self, f'{i}_pump') + if p == None: continue + p.simulate() + p_design = p.design_results + pipe_ss += p_design['Pump pipe stainless steel'] + pump_ss += p_design['Pump stainless steel'] + hdpe += p_design['Chemical storage HDPE'] + return pipe_ss, pump_ss, hdpe + + + # ========================================================================= + # _cost + # ========================================================================= + def _cost(self): + D, C = self.design_results, self.baseline_purchase_costs + ### Capital ### + # Concrete and excavation + VEX, VWC, VSC = \ + D['Excavation'], D['Wall concrete'], D['Slab concrete'] + C['Reactor excavation'] = VEX * self.excav_unit_cost + C['Wall concrete'] = VWC * self.wall_concrete_unit_cost + C['Slab concrete'] = VSC * self.slab_concrete_unit_cost + + # Membrane + C['Membrane'] = self.membrane_unit_cost * D['Membrane'] / _ft2_to_m2 + + # GAC + C['GAC'] = self.GAC_price * D['GAC'] + + # Packing material + ldpe, hdpe = 0., 0. + for i in (self.AF, self.AeF): + if i is None: + continue + ldpe += i.baseline_purchase_costs['Packing LDPE'] + hdpe += i.baseline_purchase_costs['Packing HDPE'] + + # Pump + pumps, add_OPEX = self.pumps, self.add_OPEX + pump_cost, building_cost, opex_o, opex_m = 0., 0., 0., 0. + for i in pumps: + p = getattr(self, f'{i}_pump') + if p == None: + continue + p_cost, p_add_opex = p.baseline_purchase_costs, p.add_OPEX + pump_cost += p_cost['Pump'] + building_cost += p_cost['Pump building'] + opex_o += p_add_opex['Pump operating'] + opex_m += p_add_opex['Pump maintenance'] + + C['Pumps'] = pump_cost + C['Pump building'] = building_cost + add_OPEX['Pump operating'] = opex_o + add_OPEX['Pump maintenance'] = opex_m + + # Degassing membrane + C['Degassing membrane'] = 10000 * D['Degassing membrane'] + + # Blower + self.add_equipment_cost() + + ### Heat and power ### + # Heat loss + T = self.T + coeff = self.heat_transfer_coeff + + if T is None: loss = 0. + else: + N_train, L_CSTR, W_tank, D_tank = \ + self.N_train, self.L_CSTR, self.W_tank, self.D_tank + A_W = 2 * (L_CSTR+W_tank) * D_tank + A_F = L_CSTR * W_tank + A_W *= N_train * _ft2_to_m2 + A_F *= N_train * _ft2_to_m2 + loss = coeff['wall'] * (T-self.T_air) * A_W # [W] + loss += coeff['floor'] * (T-self.T_earth) # [W] + loss += coeff['ceiling'] * (T-self.T_air) # [W] + loss *= 60*60/1e3 # W (J/s) to kJ/hr + + # Stream heating + hx = self.heat_exchanger + inf = self._inf + hx_ins0, hx_outs0 = hx.ins[0], hx.outs[0] + hx_ins0.copy_flow(inf) + hx_outs0.copy_flow(inf) + hx_ins0.T = inf.T + hx_outs0.T = T + hx.H = hx_outs0.H + loss # stream heating and heat loss + hx.simulate_as_auxiliary_exchanger(ins=hx.ins, outs=hx.outs) + + # Power for pumping and gas + pumping = 0. + for ID in self.pumps: #!!! check if cost/power of AF_pump/AeF_pump included in AF/AeF + p = getattr(self, f'{ID}_pump') + if p is None: continue + pumping += p.power_utility.rate + sparging = 0. #!!! output from submerge design + degassing = 3 * self.N_degasser # assume each uses 3 kW + self.power_utility.rate = self.blower.power_utility.rate + \ + sparging + degassing + pumping + + + ### Reactor configuration ### + @property + def reactor_type(self): + ''' + [str] Can either be "CSTR" for continuous stirred tank reactor + or "AF" for anaerobic filter. + ''' + return self._reactor_type + @reactor_type.setter + def reactor_type(self, i): + if not i.upper() in ('CSTR', 'AF'): + raise ValueError('`reactor_type` can only be "CSTR", or "AF", ' + f'not "{i}".') + self._reactor_type = i.upper() + + @property + def membrane_configuration(self): + '''[str] Can either be "cross-flow" or "submerged".''' + return self._membrane_configuration + @membrane_configuration.setter + def membrane_configuration(self, i): + i = 'cross-flow' if i.lower() in ('cross flow', 'crossflow') else i + if not i.lower() in ('cross-flow', 'submerged'): + raise ValueError('`membrane_configuration` can only be "cross-flow", ' + f'or "submerged", not "{i}".') + self._membrane_configuration = i.lower() + + @property + def membrane_type(self): + ''' + [str] Can be "hollow fiber" ("submerged" configuration only), + "flat sheet" (either "cross-flow" or "submerged" configuration), + or "multi-tube" ("cross-flow" configuration only). + ''' + return self._membrane_type + @membrane_type.setter + def membrane_type(self, i): + i = 'multi-tube' if i.lower() in ('multi tube', 'multitube') else i + if not i.lower() in ('hollow fiber', 'flat sheet', 'multi-tube'): + raise ValueError('`membrane_type` can only be "hollow fiber", ' + f'"flat sheet", or "multi-tube", not "{i}".') + self._membrane_type = i.lower() + + @property + def membrane_material(self): + ''' + [str] Can be any of the plastics ("PES", "PVDF", "PET", "PTFE") + for any of the membrane types ("hollow fiber", "flat sheet", "multi-tube"), + or "sintered steel" for "flat sheet", + or "ceramic" for "multi-tube". + ''' + return self._membrane_material + @membrane_material.setter + def membrane_material(self, i): + plastics = ('PES', 'PVDF', 'PET', 'PTFE') + if i.upper() in plastics: + self._membrane_material = i.upper() + elif i.lower() in ('sintered steel', 'ceramic'): + self._membrane_material = i.lower() + else: + raise ValueError(f'`membrane_material` can only be plastics materials ' + f'{plastics}, "sintered steel", or "ceramic", not {i}.') + + + ### Reactor/membrane tank ### + @property + def AF(self): + '''[:class:`~.FilterTank`] Anaerobic filter tank.''' + if self.reactor_type == 'CSTR': + return None + return self._AF + + @property + def AeF(self): + '''[:class:`~.FilterTank`] Aerobic filter tank.''' + if not self.include_aerobic_filter: + return None + return self._AeF + + @property + def N_train(self): + ''' + [int] Number of treatment train, should be at least two in case one failing. + ''' + return self._N_train + @N_train.setter + def N_train(self, i): + i = ceil(i) + if i < 2: + raise ValueError('`N_train` should be at least 2.') + self._N_train = i + + @property + def cas_per_tank_spare(self): + '''[int] Number of spare cassettes per train.''' + return self._cas_per_tank_spare + @cas_per_tank_spare.setter + def cas_per_tank_spare(self, i): + self._cas_per_tank_spare = ceil(i) + + @property + def mod_per_cas_range(self): + ''' + [tuple] Range (min, max) of the number of membrane modules per cassette + for the current membrane type. + ''' + return self._mod_per_cas_range + @mod_per_cas_range.setter + def mod_per_cas_range(self, i): + self._mod_per_cas_range[self.membrane_type] = \ + tuple(floor(i[0]), floor(i[1])) + + @property + def mod_per_cas(self): + ''' + [float] Number of membrane modules per cassette for the current membrane type. + ''' + return self._mod_per_cas or self._mod_per_cas_range[self.membrane_type][0] + + @property + def cas_per_tank_range(self): + ''' + [tuple] Range (min, max) of the number of membrane cassette per tank + (same for all membrane types). + ''' + return self._cas_per_tank_range + @cas_per_tank_range.setter + def cas_per_tank_range(self, i): + self._cas_per_tank_range = tuple(floor(i[0]), floor(i[1])) + + @property + def cas_per_tank(self): + ''' + [float] Number of membrane cassettes per tank for the current membrane type. + ''' + return self._cas_per_tank or self._cas_per_tank_range[0] + + @property + def N_mod_tot(self): + '''[int] Total number of membrane modules.''' + return self.N_train * self.cas_per_tank * self.mod_per_cas + + @property + def mod_surface_area(self): + ''' + [float] Surface area of the membrane for the current membrane type, [m2/module]. + Note that one module is one sheet for plat sheet and one tube for multi-tube. + ''' + return self._mod_surface_area[self.membrane_type] + @mod_surface_area.setter + def mod_surface_area(self, i): + self._mod_surface_area[self.membrane_type] = i + + @property + def L_CSTR(self): + '''[float] Length of the CSTR tank, [ft].''' + if self.reactor_type == 'AF': + return 0 + return self._inf.F_vol/_ft3_to_m3*self.HRT/(self.N_train*self.W_tank*self.D_tank) + + @property + def L_membrane_tank(self): + '''[float] Length of the membrane tank, [ft].''' + if self.membrane_configuration=='cross-flow': + return 0. + return ceil((self.cas_per_tank+self.cas_per_tank_spare)*3.4) + + @property + def W_tank(self): + '''[float] Width of the reactor/membrane tank (same value), [ft].''' + return self._W_tank + @W_tank.setter + def W_tank(self, i): + self._W_tank = i + + @property + def D_tank(self): + '''[float] Depth of the reactor/membrane tank (same value), [ft].''' + return self._D_tank + @D_tank.setter + def D_tank(self, i): + self._D_tank = i + + @property + def freeboard(self): + '''[float] Freeboard added to the depth of the reactor/membrane tank, [ft].''' + return self._freeboard + @freeboard.setter + def freeboard(self, i): + self._freeboard = i + + @property + def W_dist(self): + '''[float] Width of the distribution channel, [ft].''' + return self._W_dist + @W_dist.setter + def W_dist(self, i): + self._W_dist = i + + @property + def W_eff(self): + '''[float] Width of the effluent channel, [ft].''' + return self._W_eff + @W_eff.setter + def W_eff(self, i): + self._W_eff = i + + @property + def V_tot(self): + '''[float] Total volume of the unit, [ft3].''' + return self.D_tank*self.W_tank*self.L_CSTR*self.N_train + + @property + def OLR(self): + '''[float] Organic loading rate, [kg COD/m3/hr].''' + return compute_stream_COD(self.ins[0], 'kg/m3')*self.ins[0].F_vol/(self.V_tot*_ft3_to_m3) + + + ### Pump/blower ### + @property + def N_blower(self): + ''' + [int] Number of blowers needed for gas sparging + (not needed for some designs). + Note that this is not used in costing + (the cost is estimated based on the total sparging gas need). + ''' + if not self.add_GAC and self.membrane_configuration=='submerged': + return self._N_blower + return 0 + + @property + def N_degasser(self): + ''' + [int] Number of degassing membrane needed for dissolved biogas removal + (optional). + ''' + if self.include_degassing_membrane: + return ceil(self.Q_cmd/24/30) # assume each can hand 30 m3/d of influent + return 0 + + @property + def W_PB(self): + '''[float] Width of the pump building, [ft].''' + if self.membrane_configuration == 'submerged': + N = self.cas_per_tank + else: # cross-flow + N = ceil(self.L_CSTR/((1+8/12)+(3+4/12))) + + if 0 <= N <= 10: + W_PB = 27 + 4/12 + elif 11 <= N <= 16: + W_PB = 29 + 6/12 + elif 17 <= N <= 22: + W_PB = 31 + 8/12 + elif 23 <= N <= 28: + W_PB = 35 + elif N >= 29: + W_PB = 38 + 4/12 + else: + W_PB = 0 + + return W_PB + + @property + def L_BB(self): + '''[float] Length of the blower building, [ft].''' + if self.membrane_configuration == 'submerged': + return (69+6/12) if self.cas_per_tank<=18 else (76+8/12) + return 0 + + @property + def W_BB(self): + '''[float] Width of the blower building, [ft].''' + if self.membrane_configuration == 'submerged': + return (18+8/12) if self.cas_per_tank<=18 else 22 + return 0 + + + ### Wet well (submerged only) ### + @property + def L_well(self): + ''' + [float] Length of the wet well, [ft]. + Only needed for submerged configuration. + ''' + return self._L_well if self.membrane_configuration == 'submerged' else 0 + @L_well.setter + def L_well(self, i): + self._L_well = i + + @property + def W_well(self): + ''' + [float] Width of the wet well, [ft]. + Only needed for submerged configuration. + ''' + return self._W_well if self.membrane_configuration == 'submerged' else 0 + @W_well.setter + def W_well(self, i): + self._W_well = i + + @property + def D_well(self): + ''' + [float] Depth of the wet well, [ft]. + Only needed for submerged configuration. + ''' + return self._D_well if self.membrane_configuration == 'submerged' else 0 + @D_well.setter + def D_well(self, i): + self._D_well = i + + @property + def t_wall(self): + ''' + [float] Concrete wall thickness, [ft]. + default to be minimum of 1 ft with 1 in added for every ft of depth over 12 ft. + ''' + return self._t_wall or (1 + max(self.D_tank-12, 0)/12) + @t_wall.setter + def t_wall(self, i): + self._t_wall = i + + @property + def t_slab(self): + ''' + [float] Concrete slab thickness, [ft], + default to be 2 in thicker than the wall thickness. + ''' + return self._t_slab or self.t_wall+2/12 + @t_slab.setter + def t_slab(self, i): + self._t_slab = i + + + ### Excavation ### + @property + def excav_slope(self): + '''[float] Slope for excavation (horizontal/vertical).''' + return self._excav_slope + @excav_slope.setter + def excav_slope(self, i): + self._excav_slope = i + + @property + def constr_access(self): + '''[float] Extra room for construction access, [ft].''' + return self._constr_access + @constr_access.setter + def constr_access(self, i): + self._constr_access = i + + + ### Operation-related parameters ### + @property + def Q_mgd(self): + ''' + [float] Influent volumetric flow rate in million gallon per day, [mgd]. + ''' + return self.ins[0].F_vol*_m3_to_gal*24/1e6 + + @property + def Q_gpm(self): + '''[float] Influent volumetric flow rate in gallon per minute, [gpm].''' + return self.Q_mgd*1e6/24/60 + + @property + def Q_cmd(self): + ''' + [float] Influent volumetric flow rate in cubic meter per day, [cmd]. + ''' + return self.Q_mgd *1e6/_m3_to_gal # [m3/day] + + @property + def Q_cfs(self): + '''[float] Influent volumetric flow rate in cubic feet per second, [cfs].''' + return self.Q_mgd*1e6/24/60/60/_ft3_to_gal + + @property + def HRT(self): + ''' + [float] Hydraulic retention time, [hr]. + ''' + return self._HRT + @HRT.setter + def HRT(self, i): + self._HRT = i + + @property + def recir_ratio(self): + ''' + [float] Internal recirculation ratio, will be updated in simulation + if the originally set ratio is not adequate for the desired flow + required by GAC (if applicable). + ''' + return self._recir_ratio + @recir_ratio.setter + def recir_ratio(self, i): + self._recir_ratio = i + + @property + def J_max(self): + '''[float] Maximum membrane flux, [L/m2/hr].''' + return self._J_max + @J_max.setter + def J_max(self, i): + self._J_max = i + + @property + def J(self): + '''[float] Membrane flux, [L/m2/hr].''' + # Based on the flux of one train being offline + SA = (self.N_train-1) * self.cas_per_tank * self.mod_per_cas * self.mod_surface_area + return self._inf.F_vol*1e3/SA # 1e3 is conversion from m3 to L + + @property + def TMP_anaerobic(self): + '''[float] Transmembrane pressure in the anaerobic reactor, [psi].''' + return self._TMP_dct[self.membrane_configuration] + @TMP_anaerobic.setter + def TMP_anaerobic(self, i): + self._TMP_dct[self.membrane_configuration] = i + + @property + def TMP_aerobic(self): + ''' + [float] Transmembrane pressure in the aerobic filter, [psi]. + Defaulted to half of the reactor TMP. + ''' + if not self._include_aerobic_filter: + return 0. + else: + return self._TMP_aerobic or self._TMP_dct[self.membrane_configuration]/2 + @TMP_aerobic.setter + def TMP_aerobic(self, i): + self._TMP_aerobic = i + + @property + def SGD(self): + '''[float] Specific gas demand, [m3 gas/m2 membrane area/h].''' + return self._SGD + @SGD.setter + def SGD(self, i): + self._SGD = i + + @property + def AFF(self): + ''' + [float] Air flow fraction, used in air pipe costing. + The default value is calculated as STE/6 + (STE stands for standard oxygen transfer efficiency, and default STE is 20). + If using different STE value, AFF should be 1 if STE/6<1 + and 3.33 if STE/6>1. + ''' + return self._AFF + @AFF.setter + def AFF(self, i): + self._AFF = i + + @property + def v_cross_flow(self): + '''[float] Cross-flow velocity, [m/s].''' + return self._v_cross_flow if self.membrane_configuration=='cross-flow' else 0 + @v_cross_flow.setter + def v_cross_flow(self, i): + self._v_cross_flow = i + + @property + def v_GAC(self): + ''' + [float] Upflow velocity for GAC bed expansion, [m/hr]. + ''' + return self._v_GAC if self.add_GAC==True else 0 + @v_GAC.setter + def v_GAC(self, i): + self._v_GAC = i + + @property + def biodegradability(self): + ''' + [float of dict] Biodegradability of components, + when shown as a float, all biodegradable component are assumed to have + the same degradability. + ''' + return self._biodegradability + @biodegradability.setter + def biodegradability(self, i): + if not isinstance(i, dict): + if not 0<=i<=1: + raise ValueError('`biodegradability` should be within [0, 1], ' + f'the input value {i} is outside the range.') + self._biodegradability = dict.fromkeys(self.chemicals.IDs, i) + else: + for k, v in i.items(): + if not 0<=v<=1: + raise ValueError('`biodegradability` should be within [0, 1], ' + f'the input value for chemical "{k}" is ' + 'outside the range.') + self._biodegradability = dict.fromkeys(self.chemicals.IDs, i).update(i) + self._refresh_rxns() + + @property + def biomass_ID(self): + '''[str] ID of the Component that represents the biomass.''' + return self._xcmp.ID + @biomass_ID.setter + def biomass_ID(self, i): + self._xcmp = getattr(self.components, i) + + @property + def solids_conc(self): + '''Concentration of solids in the waste sludge, [g/L].''' + return self._solids_conc + @solids_conc.setter + def solids_conc(self, i): + self._solids_conc = i + + @property + def i_rm(self): + '''[:class:`np.array`] Removal of each chemical in this reactor.''' + return self._i_rm + + @property + def split(self): + '''Component-wise split to the treated water.''' + return self._split + @split.setter + def split(self, i): + self._split = i + self._isplit = self.chemicals.isplit(i, order=None) + + @property + def biogas_rxns(self): + ''' + [:class:`tmo.ParallelReaction`] Biogas production reactions. + ''' + return self._biogas_rxns + + @property + def growth_rxns(self): + ''' + [:class:`tmo.ParallelReaction`] Biomass growth reactions. + ''' + return self._growth_rxns + + @property + def organic_rm(self): + '''[float] Overall organic (COD) removal rate.''' + Qi, Qe = self._inf.F_vol, self.outs[1].F_vol + Si = compute_stream_COD(self._inf, 'kg/m3') + Se = compute_stream_COD(self.outs[1], 'kg/m3') + return 1 - Qe*Se/(Qi*Si) \ No newline at end of file diff --git a/qsdsan/sanunits/_tank.py b/qsdsan/sanunits/_tank.py new file mode 100644 index 00000000..21237610 --- /dev/null +++ b/qsdsan/sanunits/_tank.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +''' +QSDsan: Quantitative Sustainable Design for sanitation and resource recovery systems + +This module is developed by: + Yalin Li + +Part of this module is based on the biosteam package: +https://github.com/BioSTEAMDevelopmentGroup/biosteam + +This module is under the University of Illinois/NCSA Open Source License. +Please refer to https://github.com/QSD-Group/QSDsan/blob/main/LICENSE.txt +for license details. +''' + +from warnings import warn +from math import pi +from biosteam.units import Tank as BSTTank, MixTank as BSTMixTank, StorageTank as BSTStorageTank +from biosteam.exceptions import bounds_warning, DesignWarning +from biosteam.units.design_tools import flash_vessel_design +from biosteam.units.design_tools.specification_factors import material_densities_lb_per_ft3 +from .. import SanUnit, Construction +from ..utils import auom + + +__all__ = ('Tank', 'MixTank', 'StorageTank', ) + +_lb_to_kg = auom('lb').conversion_factor('kg') +_m_to_ft = auom('m').conversion_factor('ft') +_Pa_to_psi = auom('Pa').conversion_factor('psi') + + +class Tank(SanUnit, BSTTank, isabstract=True): + ''' + Similar to the :class:`biosteam.units.Tank`, + but can be initialized with :class:`qsdsan.SanStream` and :class:`qsdsan.WasteStream`. + + See Also + -------- + :class:`biosteam.units.Tank` + ''' + + def __init__(self, ID='', ins=None, outs=(), thermo=None, *, + vessel_type=None, tau=None, V_wf=None, + vessel_material=None, kW_per_m3=0., + init_with='WasteStream', F_BM_default=None, + include_construction=True,): + + SanUnit.__init__(self, ID, ins, outs, thermo, + init_with=init_with, F_BM_default=F_BM_default, + include_construction=include_construction,) + + self.vessel_type = vessel_type or self._default_vessel_type + self.tau = tau or self._default_tau + self.V_wf = V_wf or self._default_V_wf + self.vessel_material = vessel_material or self._default_vessel_material + self.kW_per_m3 = kW_per_m3 or self._default_kW_per_m3 + + +class MixTank(Tank, BSTMixTank): + ''' + Similar to the :class:`biosteam.units.MixTank`, + but can be initialized with :class:`qsdsan.SanStream` and :class:`qsdsan.WasteStream`. + + .. note:: + + For dynamic simulation, CSTR should be used. + + See Also + -------- + :class:`biosteam.units.MixTank` + ''' + + +class StorageTank(Tank, BSTStorageTank): + ''' + Similar to the :class:`biosteam.units.MixTank`, but can calculate material usage. + + See Also + -------- + :class:`biosteam.units.StorageTank` + ''' + + _units = {'Diameter': 'ft', + 'Length': 'ft', + 'Wall thickness': 'in', + 'Weight': 'lb'} + _bounds = {'Vertical vessel weight': (4200, 1e6), + 'Horizontal vessel weight': (1e3, 9.2e5), + 'Horizontal vessel diameter': (3, 21), + 'Vertical vessel length': (12, 40)} + _vessel_material = 'Stainless steel' + + def __init__(self, ID='', ins=None, outs=(), thermo=None, + vessel_type=None, tau=None, V_wf=None, + vessel_material=None, kW_per_m3=0., + init_with='WasteStream', F_BM_default=None, + include_construction=True, length_to_diameter=2): + Tank.__init__(self, ID=ID, ins=ins, outs=outs, thermo=thermo, + init_with=init_with, F_BM_default=F_BM_default, + include_construction=include_construction, + vessel_type=vessel_type, tau=tau, V_wf=V_wf, + vessel_material=vessel_material, kW_per_m3=kW_per_m3,) + self.length_to_diameter = length_to_diameter + + def _init_lca(self): + item_name = self.vessel_material.replace(' ', '_') + self.construction = [ + Construction(item_name.lower(), linked_unit=self, item=item_name, quantity_unit='kg'), + ] + + + def _design(self): + BSTStorageTank._design(self) + D = self.design_results + + Diameter = (4*D['Total volume']/pi/self.length_to_diameter)**(1/3) + Diameter *= _m_to_ft # convert from m to ft + L = Diameter * self.length_to_diameter # ft + D.update(self._horizontal_vessel_design(self.ins[0].P*_Pa_to_psi, Diameter, L)) + D['Material'] = self.vessel_material + if self.include_construction: self.construction[0].quantity = D['Weight']*_lb_to_kg + + + def _horizontal_vessel_design(self, pressure, diameter, length) -> dict: + pressure = pressure + diameter = diameter + length = length + # Calculate vessel weight and wall thickness + if self.vessel_material == 'Carbon steel': + rho_M = material_densities_lb_per_ft3[self.vessel_material] + else: + rho_M = material_densities_lb_per_ft3['Stainless steel 304'] + if pressure < 14.68: + warn('vacuum pressure vessel ASME codes not implemented yet; ' + 'wall thickness may be inaccurate and stiffening rings may be ' + 'required', category=DesignWarning) + VW, VWT = flash_vessel_design.compute_vessel_weight_and_wall_thickness( + pressure, diameter, length, rho_M) + bounds_warning(self, 'Horizontal vessel weight', VW, 'lb', + self._bounds['Horizontal vessel weight'], 'cost') + bounds_warning(self, 'Horizontal vessel diameter', diameter, 'ft', + self._bounds['Horizontal vessel diameter'], 'cost') + Design = {} + Design['Vessel type'] = 'Horizontal' + Design['Length'] = length # ft + Design['Diameter'] = diameter # ft + Design['Weight'] = VW # lb + Design['Wall thickness'] = VWT # in + return Design + + @property + def vessel_material(self): + return self._vessel_material + @vessel_material.setter + def vessel_material(self, i): + exist_material = getattr(self, '_vessel_material', None) + BSTStorageTank.vessel_material.fset(self, i) + if i and exist_material == i: return # type doesn't change, no need to reload construction items + self._init_lca() \ No newline at end of file diff --git a/qsdsan/sanunits/_toilet.py b/qsdsan/sanunits/_toilet.py new file mode 100644 index 00000000..a8a03e06 --- /dev/null +++ b/qsdsan/sanunits/_toilet.py @@ -0,0 +1,1190 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +''' +QSDsan: Quantitative Sustainable Design for sanitation and resource recovery systems + +This module is developed by: + + Yalin Li + + Lewis Rowles + + Lane To + +This module is under the University of Illinois/NCSA Open Source License. +Please refer to https://github.com/QSD-Group/QSDsan/blob/main/LICENSE.txt +for license details. +''' + + +# %% + +import numpy as np +from math import ceil +from warnings import warn +from .. import SanUnit, WasteStream, Construction +from ..processes import Decay +from ..utils import ospath, load_data, data_path, dct_from_str, price_ratio + +__all__ = ('Toilet', 'MURT', 'PitLatrine', 'UDDT', ) + + +# %% + +toilet_path = ospath.join(data_path, 'sanunit_data/_toilet.tsv') + +class Toilet(SanUnit, Decay, isabstract=True): + ''' + Abstract class containing common parameters and design algorithms for toilets + based on `Trimmer et al. `_ + + Parameters + ---------- + degraded_components : tuple + IDs of components that will degrade (simulated by first-order decay). + N_user : int, float + Number of people per toilet. + Note that this number can be a float when calculated from `N_tot_user` and `N_toilet`. + N_toilet : int + Number of parallel toilets. + In calculation, `N_toilet` will be calculated as `ceil(N_tot_user/N_user)`. + N_tot_user : int + Total number of users. + + .. note:: + + If `N_tot_user` is provided (i.e., not "None"), + then updating `N_user` will recalculate `N_toilet`, and vice versa. + + if_toilet_paper : bool + If toilet paper is used. + if_flushing : bool + If water is used for flushing. + if_cleansing : bool + If water is used for cleansing. + if_desiccant : bool + If desiccant is used for moisture and odor control. + if_air_emission : bool + If emission to air occurs + (i.e., if the pit is completely sealed off from the atmosphere). + if_ideal_emptying : bool + If the toilet appropriately emptied to avoid contamination to the + environmental. + CAPEX : float + Capital cost of a single toilet. + OPEX_over_CAPEX : float + Fraction of annual operating cost over total capital cost. + price_ratio : float + Calculated capital cost will be multiplied by this number + to consider the effect in cost difference from different locations. + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.processes.Decay ` + + ''' + _N_ins = 6 + _outs_size_is_fixed = False + density_dct = { + 'Sand': 1442, + 'Gravel': 1600, + 'Brick': 1750, + 'Plastic': 0.63, + 'Steel': 7900, + 'StainlessSteelSheet': 2.64 + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + degraded_components=('OtherSS',), N_user=1, N_toilet=1, N_tot_user=None, + if_toilet_paper=True, if_flushing=True, if_cleansing=False, + if_desiccant=False, if_air_emission=True, if_ideal_emptying=True, + CAPEX=None, OPEX_over_CAPEX=None, price_ratio=1.): + + SanUnit.__init__(self, ID, ins, outs, thermo, init_with, F_BM_default=1) + self.degraded_components = tuple(degraded_components) + self._N_user = self._N_toilet = self._N_tot_user = None + self.N_user = N_user + self.N_toilet = N_toilet + self.N_tot_user = N_tot_user + self.if_toilet_paper = if_toilet_paper + self.if_flushing = if_flushing + self.if_cleansing = if_cleansing + self.if_desiccant = if_desiccant + self.if_air_emission = if_air_emission + self.if_ideal_emptying = if_ideal_emptying + self.CAPEX = CAPEX + self.OPEX_over_CAPEX = OPEX_over_CAPEX + self.price_ratio = price_ratio + + data = load_data(path=toilet_path) + for para in data.index: + value = float(data.loc[para]['expected']) + if para in ('desiccant_V', 'desiccant_rho'): + setattr(self, para, value) + else: + setattr(self, '_'+para, value) + del data + + self._empty_ratio = 0.59 + + + def _run(self): + ur, fec, tp, fw, cw, des = self.ins + tp.imass['Tissue'] = int(self.if_toilet_paper)*self.toilet_paper + fw.imass['H2O'] = int(self.if_flushing)*self.flushing_water + cw.imass['H2O'] = int(self.if_cleansing)*self.cleansing_water + des.imass['WoodAsh'] = int(self.if_desiccant)*self.desiccant + + def _scale_up_outs(self): + ''' + Scale up the effluent based on the number of user per toilet and + toilet number. + ''' + N_tot_user = self.N_tot_user or self.N_toilet*self.N_user + for i in self.outs: + if not i.F_mass == 0: + i.F_mass *= N_tot_user + + + def _cost(self): + self.baseline_purchase_costs['Total toilets'] = self.CAPEX * self.N_toilet * self.price_ratio + add_OPEX = self.baseline_purchase_costs['Total toilets']*self.OPEX_over_CAPEX/365/24 + self._add_OPEX = {'Additional OPEX': add_OPEX} + + + @staticmethod + def get_emptying_emission(waste, CH4, N2O, empty_ratio, CH4_factor, N2O_factor): + ''' + Calculate emissions due to non-ideal emptying based on + `Trimmer et al. `_, + + Parameters + ---------- + stream : WasteStream + Excreta stream that is not appropriately emptied (before emptying). + CH4 : WasteStream + Fugitive CH4 gas (before emptying). + N2O : WasteStream + Fugitive N2O gas (before emptying). + empty_ratio : float + Fraction of excreta that is appropriately emptied.. + CH4_factor : float + Factor to convert COD removal to CH4 emission. + N2O_factor : float + Factor to convert COD removal to N2O emission. + + Returns + ------- + stream : WasteStream + Excreta stream that is not appropriately emptied (after emptying). + CH4 : WasteStream + Fugitive CH4 gas (after emptying). + N2O : WasteStream + Fugitive N2O gas (after emptying). + ''' + COD_rmvd = waste.COD*(1-empty_ratio)/1e3*waste.F_vol + CH4.imass['CH4'] += COD_rmvd * CH4_factor + N2O.imass['N2O'] += COD_rmvd * N2O_factor + waste.mass *= empty_ratio + + @property + def N_user(self): + '''[int, float] Number of people per toilet.''' + return self._N_user or self.N_tot_user/self.N_toilet + @N_user.setter + def N_user(self, i): + if i is not None: + N_user = self._N_user = int(i) + old_toilet = self._N_toilet + if old_toilet and self.N_tot_user: + new_toilet = ceil(self.N_tot_user/N_user) + warn(f'With the provided `N_user`, the previous `N_toilet` of {old_toilet} ' + f'is recalculated from `N_tot_user` and `N_user` as {new_toilet}.') + self._N_toilet = None + else: + self._N_user = i + + @property + def N_toilet(self): + '''[int] Number of parallel toilets.''' + return self._N_toilet or ceil(self.N_tot_user/self.N_user) + @N_toilet.setter + def N_toilet(self, i): + if i is not None: + N_toilet = self._N_toilet = ceil(i) + old_user = self._N_user + if old_user and self.N_tot_user: + new_user = self.N_tot_user/N_toilet + warn(f'With the provided `N_toilet`, the previous `N_user` of {old_user} ' + f'is recalculated from `N_tot_user` and `N_toilet` as {new_user}.') + self._N_user = None + else: + self._N_toilet = i + + @property + def N_tot_user(self): + '''[int] Number of total users.''' + return self._N_tot_user + @N_tot_user.setter + def N_tot_user(self, i): + if i is not None: + self._N_tot_user = int(i) + else: + self._N_tot_user = None + + @property + def toilet_paper(self): + ''' + [float] Amount of toilet paper used + (if ``if_toilet_paper`` is True), [kg/cap/hr]. + ''' + return self._toilet_paper + @toilet_paper.setter + def toilet_paper(self, i): + self._toilet_paper = i + + @property + def flushing_water(self): + ''' + [float] Amount of water used for flushing + (if ``if_flushing_water`` is True), [kg/cap/hr]. + ''' + return self._flushing_water + @flushing_water.setter + def flushing_water(self, i): + self._flushing_water = i + + @property + def cleansing_water(self): + ''' + [float] Amount of water used for cleansing + (if ``if_cleansing_water`` is True), [kg/cap/hr]. + ''' + return self._cleansing_water + @cleansing_water.setter + def cleansing_water(self, i): + self._cleansing_water = i + + @property + def desiccant(self): + ''' + [float] Amount of desiccant used (if ``if_desiccant`` is True), [kg/cap/hr]. + + .. note:: + + Value set by ``desiccant_V`` and ``desiccant_rho``. + + ''' + return self.desiccant_V*self.desiccant_rho + + @property + def N_volatilization(self): + ''' + [float] Fraction of input N that volatilizes to the air + (if ``if_air_emission`` is True). + ''' + return self._N_volatilization + @N_volatilization.setter + def N_volatilization(self, i): + self._N_volatilization = i + + @property + def empty_ratio(self): + ''' + [float] Fraction of excreta that is appropriately emptied. + + .. note:: + + Will be 1 (i.e., 100%) if ``if_ideal_emptying`` is True. + + ''' + if self.if_ideal_emptying: + return 1. + return self._empty_ratio + @empty_ratio.setter + def empty_ratio(self, i): + if self.if_ideal_emptying: + warn(f'`if_ideal_emptying` is True, the set value {i} is ignored.') + self._empty_ratio = i + + @property + def MCF_aq(self): + '''[float] Methane correction factor for COD lost due to inappropriate emptying.''' + return self._MCF_aq + @MCF_aq.setter + def MCF_aq(self, i): + self._MCF_aq = i + + @property + def N2O_EF_aq(self): + '''[float] Fraction of N emitted as N2O due to inappropriate emptying.''' + return self._N2O_EF_aq + @N2O_EF_aq.setter + def N2O_EF_aq(self, i): + self._N2O_EF_aq = i + + @property + def if_N2O_emission(self): + '''[bool] Whether to consider N degradation and fugitive N2O emission.''' + return self.if_air_emission + @if_N2O_emission.setter + def if_N2O_emission(self, i): + raise ValueError('Setting `if_N2O_emission` for `PitLatrine` is not supported, ' + 'please set `if_air_emission` instead.') + + +# %% + +murt_path = ospath.join(data_path, 'sanunit_data/_murt.tsv') + +@price_ratio() +class MURT(Toilet): + ''' + Multi-unit reinvented toilet. + + The following components should be included in system thermo object for simulation: + Tissue, WoodAsh, H2O, NH3, NonNH3, P, K, Mg, CH4, N2O. + + The following impact items should be pre-constructed for life cycle assessment: + Ceramic, Fan. + + Parameters + ---------- + ins : Iterable(stream) + waste_in: mixed excreta. + Outs : Iterable(stream) + waste_out: degraded mixed excreta. + CH4: fugitive CH4. + N2O: fugitive N2O. + N_squatting_pan_per_toilet : int + The number of squatting pan per toilet. + N_urinal_per_toilet : int + The number of urinals per toilet. + if_include_front_end : bool + If False, will not consider the capital and operating costs of this unit. + + See Also + -------- + :ref:`qsdsan.sanunits.Toilet ` + ''' + _N_outs = 3 + _units = { + 'Collection period': 'd', + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + degraded_components=('OtherSS',), N_user=1, N_tot_user=1, + N_toilet=None, if_toilet_paper=True, if_flushing=True, if_cleansing=False, + if_desiccant=True, if_air_emission=True, if_ideal_emptying=True, + CAPEX=0, OPEX_over_CAPEX=0, lifetime=10, + N_squatting_pan_per_toilet=1, N_urinal_per_toilet=1, + if_include_front_end=True, **kwargs): + + Toilet.__init__( + self, ID, ins, outs, thermo=thermo, init_with=init_with, + degraded_components=degraded_components, + N_user=N_user, N_tot_user=N_tot_user, N_toilet=N_toilet, + if_toilet_paper=if_toilet_paper, if_flushing=if_flushing, + if_cleansing=if_cleansing, if_desiccant=if_desiccant, + if_air_emission=if_air_emission, if_ideal_emptying=if_ideal_emptying, + CAPEX=CAPEX, OPEX_over_CAPEX=OPEX_over_CAPEX + ) + self.N_squatting_pan_per_toilet = N_squatting_pan_per_toilet + self.N_urinal_per_toilet = N_urinal_per_toilet + self.if_include_front_end = if_include_front_end + self._mixed_in = WasteStream(f'{self.ID}_mixed_in') + + data = load_data(path=murt_path) + for para in data.index: + value = float(data.loc[para]['expected']) + setattr(self, para, value) + del data + + for attr, value in kwargs.items(): + setattr(self, attr, value) + + def _init_lca(self): + self.construction = [ + Construction(item='Ceramic', linked_unit=self, quantity_unit='kg'), + Construction(item='Fan', linked_unit=self, quantity_unit='kg'), + ] + + + def _run(self): + Toilet._run(self) + mixed_out, CH4, N2O = self.outs + CH4.phase = N2O.phase = 'g' + + mixed_in = self._mixed_in + mixed_in.mix_from(self.ins) + tot_COD_kg = sum(float(getattr(i, 'COD')) * i.F_vol for i in self.ins) / 1e3 + # breakpoint() + # Air emission + if self.if_air_emission: + # N loss due to ammonia volatilization + NH3_rmd, NonNH3_rmd = \ + self.allocate_N_removal(mixed_in.TN/1e3*mixed_in.F_vol*self.N_volatilization, + mixed_in.imass['NH3']) + mixed_in.imass ['NH3'] -= NH3_rmd + mixed_in.imass['NonNH3'] -= NonNH3_rmd + + # Energy/N loss due to degradation + mixed_in._COD = tot_COD_kg * 1e3 / mixed_in.F_vol # accounting for COD loss in leachate + Decay._first_order_run(self, waste=mixed_in, treated=mixed_out, CH4=CH4, N2O=N2O) + else: + mixed_out.copy_like(mixed_in) + CH4.empty() + N2O.empty() + + # Aquatic emission when not ideally emptied + if not self.if_ideal_emptying: + self.get_emptying_emission( + waste=mixed_out, CH4=CH4, N2O=N2O, + empty_ratio=self.empty_ratio, + CH4_factor=self.COD_max_decay*self.MCF_aq*self.max_CH4_emission, + N2O_factor=self.N2O_EF_decay*44/28) + self._scale_up_outs() + + + def _design(self): + design = self.design_results + constr = self.construction + if self.if_include_front_end: + design['Number of users per toilet'] = self.N_user + design['Parallel toilets'] = N = self.N_toilet + design['Collection period'] = self.collection_period + design['Ceramic'] = Ceramic_quant = ( + self.squatting_pan_weight * self.N_squatting_pan_per_toilet+ + self.urinal_weight * self.N_urinal_per_toilet + ) + design['Fan'] = Fan_quant = 1 # assume fan quantity is 1 + constr[0].quantity = Ceramic_quant * N + constr[1].quantity = Fan_quant * N + self.add_construction(add_cost=False) + else: + design.clear() + for i in constr: i.quantity = 0 + + + def _cost(self): + C = self.baseline_purchase_costs + if self.if_include_front_end: + N_toilet = self.N_toilet + C['Ceramic Toilets'] = ( + self.squatting_pan_cost * self.N_squatting_pan_per_toilet + + self.urinal_cost * self.N_urinal_per_toilet + ) * N_toilet + C['Fan'] = self.fan_cost * N_toilet + C['Misc. parts'] = ( + self.led_cost + + self.anticor_floor_cost + + self.circuit_change_cost + + self.pipe_cost + ) * N_toilet + + ratio = self.price_ratio + for equipment, cost in C.items(): + C[equipment] = cost * ratio + else: + self.baseline_purchase_costs.clear() + + sum_purchase_costs = sum(v for v in C.values()) + self.add_OPEX = ( + self._calc_replacement_cost() + + self._calc_maintenance_labor_cost() + + sum_purchase_costs * self.OPEX_over_CAPEX / (365 * 24) + ) + + def _calc_replacement_cost(self): + return 0 + + def _calc_maintenance_labor_cost(self): + return 0 + + @property + def collection_period(self): + '''[float] Time interval between storage tank collection, [d].''' + return self._collection_period + @collection_period.setter + def collection_period(self, i): + self._collection_period = float(i) + + @property + def tau(self): + '''[float] Retention time of the unit, same as `collection_period`.''' + return self.collection_period + @tau.setter + def tau(self, i): + self.collection_period = i + + +# %% + +pit_path = ospath.join(data_path, 'sanunit_data/_pit_latrine.tsv') + +class PitLatrine(Toilet): + ''' + Single pit latrine based on `Trimmer et al. `_, + a subclass of :class:`qsdsan.sanunits.Toilet`. + + The following components should be included in system thermo object for simulation: + Tissue, WoodAsh, H2O, NH3, NonNH3, P, K, CH4, N2O. + + The following impact items should be pre-constructed for life cycle assessment: + Cement, Sand, Gravel, Brick, Plastic, Steel, Wood, Excavation. + + Parameters + ---------- + ins : stream obj + Excreta. + outs : Iterable(stream obj) + Recyclable mixed excreta, stream leached to soil, fugitive CH4, and fugitive N2O. + lifetime : int + Lifetime of this pit latrine, [yr]. + if_leaching : bool + If infiltration to soil occurs + (i.e., if the pit walls and floors are permeable). + if_shared : bool + If the toilet is shared. + if_pit_above_water_table : bool + If the pit is above local water table. + + Examples + -------- + `bwaise systems `_ + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.sanunits.Toilet ` + ''' + _N_outs = 4 + _units = { + 'Emptying period': 'yr', + 'Single pit volume': 'm3', + 'Single pit area': 'm2', + 'Single pit depth': 'm' + } + + # # Legacy code to add checkers + # _P_leaching = Frac_D(name='P_leaching') + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + degraded_components=('OtherSS',), N_user=1, N_toilet=1, N_tot_user=None, + if_toilet_paper=True, if_flushing=True, if_cleansing=False, + if_desiccant=False, if_air_emission=True, if_ideal_emptying=True, + CAPEX=449, OPEX_over_CAPEX=0.05, + lifetime=8, if_leaching=True, if_shared=True, + if_pit_above_water_table=True, **kwargs): + + Toilet.__init__(self, ID, ins, outs, thermo, init_with, + degraded_components, N_user, N_toilet, N_tot_user, + if_toilet_paper, if_flushing, if_cleansing, if_desiccant, + if_air_emission, if_ideal_emptying, CAPEX, OPEX_over_CAPEX) + + self.lifetime = lifetime + self.if_leaching = if_leaching + self.if_pit_above_water_table = if_pit_above_water_table + self.if_shared = if_shared + self._pit_depth = 4.57 # m + self._pit_area = 0.8 # m2 + self._liq_leaching = None + self._mixed_in = WasteStream(f'{self.ID}_mixed_in') + + data = load_data(path=pit_path) + for para in data.index: + if para in ('MCF_decay', 'N2O_EF_decay'): + value = dct_from_str(data.loc[para]['expected'], dtype='float') + else: + value = float(data.loc[para]['expected']) + setattr(self, '_'+para, value) + del data + + for attr, value in kwargs.items(): + setattr(self, attr, value) + + def _init_lca(self): + self.construction = [ + Construction('cement', linked_unit=self, item='Cement', quantity_unit='kg'), + Construction('sand', linked_unit=self, item='Sand', quantity_unit='kg'), + Construction('gravel', linked_unit=self, item='Gravel', quantity_unit='kg'), + Construction('brick', linked_unit=self, item='Brick', quantity_unit='kg'), + Construction('liner', linked_unit=self, item='Plastic', quantity_unit='kg'), + Construction('steel', linked_unit=self, item='Steel', quantity_unit='kg'), + Construction('wood', linked_unit=self, item='Wood', quantity_unit='m3'), + Construction('excavation', linked_unit=self, item='Excavation', quantity_unit='m3'), + ] + + def _run(self): + Toilet._run(self) + mixed_out, leachate, CH4, N2O = self.outs + CH4.phase = N2O.phase = 'g' + + mixed_in = self._mixed_in + mixed_in.mix_from(self.ins) + tot_COD_kg = sum(float(getattr(i, '_COD') or getattr(i, 'COD'))*i.F_vol for i in self.ins)/1e3 + + # All composite variables in mg/L + # Leaching, COD changes due to leaching is not considered + if self.if_leaching: + # Additional assumption not in ref [1] + leachate.imass['H2O'] = mixed_in.imass['H2O'] * self.liq_leaching + leachate.imass['NH3'], leachate.imass['NonNH3'] = \ + self.allocate_N_removal(mixed_in.TN/1e3*mixed_in.F_vol*self.N_leaching, + mixed_in.imass['NH3']) + leachate.imass['P'] = mixed_in.imass['P'] * self.P_leaching + leachate.imass['K'] = mixed_in.imass['K'] * self.K_leaching + mixed_in.mass -= leachate.mass + + # Air emission + if self.if_air_emission: + # N loss due to ammonia volatilization + NH3_rmd, NonNH3_rmd = \ + self.allocate_N_removal(mixed_in.TN/1e3*mixed_in.F_vol*self.N_volatilization, + mixed_in.imass['NH3']) + mixed_in.imass ['NH3'] -= NH3_rmd + mixed_in.imass['NonNH3'] -= NonNH3_rmd + + # Energy/N loss due to degradation + mixed_in._COD = tot_COD_kg * 1e3 / mixed_in.F_vol # accounting for COD loss in leachate + Decay._first_order_run(self, waste=mixed_in, treated=mixed_out, CH4=CH4, N2O=N2O) + else: + mixed_out.copy_like(mixed_in) + CH4.empty() + N2O.empty() + + # Aquatic emission when not ideally emptied + if not self.if_ideal_emptying: + self.get_emptying_emission( + waste=mixed_out, CH4=CH4, N2O=N2O, + empty_ratio=self.empty_ratio, + CH4_factor=self.COD_max_decay*self.MCF_aq*self.max_CH4_emission, + N2O_factor=self.N2O_EF_decay*44/28) + + # Drain extra water, assume density of water to be 1 kg/L + sludge = self.sludge_accum_rate/(365*24) + diff = mixed_out.F_mass - sludge + if diff > 0: + mixed_COD = mixed_out._COD * mixed_out.F_vol + mixed_out.imass['H2O'] -= diff + mixed_out.imass['H2O'] = max(0, mixed_out.imass['H2O']) + mixed_out._COD = mixed_COD / mixed_out.F_vol + + self._scale_up_outs() + + + def _design(self): + design = self.design_results + design['Number of users per toilet'] = self.N_user + design['Parallel toilets'] = N = self.N_toilet + design['Emptying period'] = self.emptying_period + design['Single pit volume'] = self.pit_V + design['Single pit area'] = self.pit_area + design['Single pit depth'] = self.pit_depth + + density = self.density_dct + constr = self.construction + constr[0].quantity = 700 * N # cement + constr[1].quantity = 2.2 * density['Sand'] * N + constr[2].quantity = 0.8 * density['Gravel'] * N + constr[3].quantity = 54*0.0024 * density['Brick'] * N + constr[4].quantity = 16 * density['Plastic'] * N + constr[5].quantity = 0.00425 * density['Steel'] * N + constr[6].quantity = 0.19 * N # wood + constr[7].quantity = self.pit_V * N # excavation + + self.add_construction(add_cost=False) + + + @property + def pit_depth(self): + '''[float] Depth of the pit, [m].''' + return self._pit_depth + @pit_depth.setter + def pit_depth(self, i): + self._pit_depth = i + + @property + def pit_area(self): + '''[float] Area of the pit, [m2].''' + return self._pit_area + @pit_area.setter + def pit_area(self, i): + self._pit_area = i + + # With baseline assumptions, this is about the same as the total volume of the excreta + @property + def pit_V(self): + '''[float] Volume of the pit, [m3].''' + return self.pit_area*self.pit_depth + + @property + def emptying_period(self): + '''[float] Time interval between pit emptying, [yr].''' + return self._emptying_period + @emptying_period.setter + def emptying_period(self, i): + self._emptying_period = i + + @property + def sludge_accum_rate(self): + '''[float] Sludge accumulation rate, [L/cap/yr].''' + return self._sludge_accum_rate + @sludge_accum_rate.setter + def sludge_accum_rate(self, i): + self._sludge_accum_rate = i + + @property + def liq_leaching(self): + ''' + [float] Fraction of input water that leaches to the soil + (if if_leaching is True). If not set, then return the maximum of + fraction of N, P, K leaching + ''' + return self._liq_leaching or \ + max(self.N_leaching, self.P_leaching, self.K_leaching) + @liq_leaching.setter + def liq_leaching(self, i): + self._liq_leaching = i + + @property + def N_leaching(self): + ''' + [float] Fraction of input N that leaches to the soil + (if if_leaching is True). + ''' + return self._N_leaching + @N_leaching.setter + def N_leaching(self, i): + self._N_leaching = i + # Legacy code to add checkers + # @Frac_C(self) + # def N_leaching(): return i + + @property + def P_leaching(self): + ''' + [float] Fraction of input P that leaches to the soil + (if if_leaching is True). + ''' + return self._P_leaching + @P_leaching.setter + def P_leaching(self, i): + self._P_leaching = i + + @property + def K_leaching(self): + ''' + [float] Fraction of input K that leaches to the soil + (if if_leaching is True). + ''' + return self._K_leaching + @K_leaching.setter + # This is faster than using descriptors or decorators, but (I think) less elegant + def K_leaching(self, i): + if i < 0: + raise ValueError('Value for K_leaching cannot be negative') + self._K_leaching = i + + def _return_MCF_EF(self): + # self._MCF and self._N2O_EF are dict for + # single_above_water, communal_above_water, below_water + if self.if_pit_above_water_table: + if not self.if_shared: + return 'single_above_water' + else: + return 'communal_above_water' + else: + return 'below_water' + + @property + def MCF_decay(self): + '''[float] Methane correction factor for COD degraded during storage.''' + return float(self._MCF_decay[self._return_MCF_EF()]) + @MCF_decay.setter + def MCF_decay(self, i): + self._MCF_decay[self._return_MCF_EF()] = i + + @property + def N2O_EF_decay(self): + '''[float] Fraction of N emitted as N2O during storage.''' + return float(self._N2O_EF_decay[self._return_MCF_EF()]) + @N2O_EF_decay.setter + def N2O_EF_decay(self, i): + self._N2O_EF_decay[self._return_MCF_EF()] = i + + @property + def tau(self): + '''[float] Retention time of the unit, same as `emptying_period` but in day.''' + return self.emptying_period*365 + @tau.setter + def tau(self, i): + self.emptying_period = i/365 + + +# %% + +uddt_path = ospath.join(data_path, 'sanunit_data/_uddt.tsv') + +class UDDT(Toilet): + ''' + Urine-diverting dry toilet with liquid storage tank and dehydration vault + for urine and feces storage, respectively, based on + `Trimmer et al. `_, + a subclass of :class:`qsdsan.sanunits.Toilet`. + + The following components should be included in system thermo object for simulation: + Tissue, WoodAsh, H2O, NH3, NonNH3, P, K, Mg, CH4, N2O. + + The following impact items should be pre-constructed for life cycle assessment: + Cement, Sand, Gravel, Brick, Plastic, Steel, StainlessSteelSheet, Wood. + + Parameters + ---------- + ins : stream obj + Excreta stream. + outs : Iterable(stream obj) + Recyclable liquid urine, recyclable solid feces, struvite scaling (irrecoverable), + hydroxyapatite scaling (irrecoverable), fugitive CH4, and fugitive N2O. + lifetime : int + Lifetime of this UDDT, [yr]. + T : float + Temperature, [K]. + safety_factor : float + Safety factor for pathogen removal during onsite treatment, + must be larger than 1. + if_treatment : bool + If has onsite treatment. + + Examples + -------- + `bwaise systems `_ + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.sanunits.Toilet ` + ''' + _N_outs = 6 + _units = { + 'Collection period': 'd', + 'Single tank volume': 'm3', + 'Single vault volume': 'm3', + 'Treatment time': 'd', + 'Treatment volume': 'm3' + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + degraded_components=('OtherSS',), N_user=1, N_toilet=1, N_tot_user=None, + if_toilet_paper=True, if_flushing=False, if_cleansing=False, + if_desiccant=True, if_air_emission=True, if_ideal_emptying=True, + CAPEX=553, OPEX_over_CAPEX=0.1, lifetime=8, + T=273.15+25, safety_factor=1, if_prep_loss=True, if_treatment=False, + **kwargs): + + Toilet.__init__(self, ID, ins, outs, thermo, init_with, + degraded_components, N_user, N_toilet, N_tot_user, + if_toilet_paper, if_flushing, if_cleansing, if_desiccant, + if_air_emission, if_ideal_emptying, CAPEX, OPEX_over_CAPEX) + self.lifetime = lifetime + self.T = T + self._safety_factor = safety_factor + self.if_prep_loss = if_prep_loss + self.if_treatment = if_treatment + self._mixed_sol = WasteStream(f'{self.ID}_mixed_sol') + + data = load_data(path=uddt_path) + for para in data.index: + value = float(data.loc[para]['expected']) + setattr(self, '_'+para, value) + del data + + self._tank_V = 60/1e3 # m3 + for attr, value in kwargs.items(): + setattr(self, attr, value) + + + def _init_lca(self): + self.construction = [ + Construction('cement', linked_unit=self, item='Cement', quantity_unit='kg'), + Construction('sand', linked_unit=self, item='Sand', quantity_unit='kg'), + Construction('gravel', linked_unit=self, item='Gravel', quantity_unit='kg'), + Construction('brick', linked_unit=self, item='Brick', quantity_unit='kg'), + Construction('liner', linked_unit=self, item='Plastic', quantity_unit='kg'), + Construction('steel', linked_unit=self, item='Steel', quantity_unit='kg'), + Construction('ss_sheet', linked_unit=self, item='StainlessSteelSheet', quantity_unit='kg'), + Construction('wood', linked_unit=self, item='Wood', quantity_unit='m3'), + ] + + def _run(self): + Toilet._run(self) + liq, sol, struvite, HAP, CH4, N2O = self.outs + liq.copy_like(self.ins[0]) + + # Assume all of additives (toilet paper and desiccant) and water + # (cleansing and flushing) go to feces + _COD = self.ins[1]._COD or self.ins[1].COD + sol_COD = _COD*self.ins[1].F_vol/1e3 + mixed_sol = self._mixed_sol + mixed_sol.mix_from(self.ins[1:]) + mixed_sol._COD = sol_COD*1e3/mixed_sol.F_vol + sol.copy_like(mixed_sol) + + struvite.phase = HAP.phase = 's' + CH4.phase = N2O.phase = 'g' + + # Modified from ref [1], assume this only happens when air emission occurs + # (to be consistent with pit latrine) + if self.if_air_emission: + # N loss due to ammonia volatilization + NH3_rmd, NonNH3_rmd = \ + self.allocate_N_removal(liq.TN/1e3*liq.F_vol*self.N_volatilization, + liq.imass['NH3']) + liq.imass ['NH3'] -= NH3_rmd + liq.imass['NonNH3'] -= NonNH3_rmd + + # Energy/N loss due to degradation + Decay._first_order_run(self, waste=mixed_sol, treated=sol, CH4=CH4, N2O=N2O) + else: + sol.copy_like(mixed_sol) + CH4.empty() + N2O.empty() + + # N and P losses due to struvite and hydroxyapatite (HAp) + if self.if_prep_loss: + # Struvite + NH3_mol = liq.imol['NH3'] + P_mol = liq.imol['P'] + Mg_mol = liq.imol['Mg'] + Ksp = 10**(-self.struvite_pKsp) + # Ksp = (initial N - struvite)(initial P - struvite)(initial Mg - struvite) + coeff = [1, -(NH3_mol+P_mol+Mg_mol), + (NH3_mol*P_mol + P_mol*Mg_mol + Mg_mol*NH3_mol), + (Ksp - NH3_mol*P_mol*Mg_mol)] + struvite_mol = 0 + for i in np.roots(coeff): + if 0 < i < min(NH3_mol, P_mol, Mg_mol): + struvite_mol = i + struvite.imol['Struvite'] = \ + max(0, min(NH3_mol, P_mol, Mg_mol, struvite_mol)) + liq.imol['NH3'] -= struvite_mol + liq.imol['P'] -= struvite_mol + liq.imol['Mg'] -= struvite_mol + # HAP + left_P = liq.imol['P'] - 3*(liq.imol['Ca']/5) + # Remaining P enough to precipitate all Ca as HAP + if left_P > 0: + HAP.imol['HAP'] = liq.imol['Ca']/5 + liq.imol['P'] = left_P + liq.imol['Ca'] = 0 + else: + HAP.imol['HAP'] = liq.imol['P']/3 + liq.imol['Ca'] -= 5*(liq.imol['P']/3) + liq.imol['P'] = 0 + else: + struvite.empty() + HAP.empty() + + # Onsite treatment + if self.if_treatment: + NH3_mmol = liq.imol['NH3'] * 1e3 + ur_DM = 1 - liq.imass['H2O']/liq.F_mass + pKa = 0.09018 + (2729.92/self.T) + f_NH3_Emerson = 1 / (10**(pKa-self.ur_pH)+1) + alpha = 0.82 - 0.011*np.sqrt(NH3_mmol+1700*ur_DM) + beta = 1.17 + 0.02 * np.sqrt(NH3_mmol+1100*ur_DM) + f_NH3_Pitzer = f_NH3_Emerson * \ + (alpha + ((1-alpha)*(f_NH3_Emerson**beta))) + NH3_conc = NH3_mmol * f_NH3_Pitzer + + # Time (in days) to reach desired inactivation level + self.treatment_tau = ((3.2 + self.log_removal) \ + / (10**(-3.7+0.062*(self.T-273.15)) * (NH3_conc**0.7))) \ + * 1.14*self.safety_factor + # Total volume in m3 + self.treatment_V = self.treatment_tau * liq.F_vol*24 + else: + self.treatment_tau = self.treatment_V = 0 + + # Feces water loss if desiccant is added + if self.if_desiccant: + sol_COD = sol._COD*sol.F_vol/1e3 + MC_min = self.fec_moi_min + r = self.fec_moi_red_rate + t = self.collection_period + mixed = sol.copy() + mixed.mix_from(self.ins[1:]) + fec_moi_int = mixed.imass['H2O']/mixed.F_mass + fec_moi = MC_min + (fec_moi_int-MC_min)/(r*t)*(1-np.exp(-r*t)) + dry_mass = sol.F_mass-sol.imass['H2O'] + tot_mass = (sol.F_mass-sol.imass['H2O']) / (1-fec_moi) + sol.imass['H2O'] = tot_mass - dry_mass + sol._COD = sol_COD*1e3/sol.F_vol + + self._vault_V = sol.F_vol*self.collection_period*24 # in day + + # Non-ideal emptying + if not self.if_ideal_emptying: + self.get_emptying_emission( + waste=liq, CH4=CH4, N2O=N2O, + empty_ratio=self.empty_ratio, + CH4_factor=self.COD_max_decay*self.MCF_aq*self.max_CH4_emission, + N2O_factor=self.N2O_EF_decay*44/28) + self.get_emptying_emission( + waste=sol, CH4=CH4, N2O=N2O, + empty_ratio=self.empty_ratio, + CH4_factor=self.COD_max_decay*self.MCF_aq*self.max_CH4_emission, + N2O_factor=self.N2O_EF_decay*44/28) + + self._scale_up_outs() + + + def _design(self): + design = self.design_results + design['Number of users per toilet'] = self.N_user + design['Parallel toilets'] = N = self.N_toilet + design['Collection period'] = self.collection_period + design['Single tank volume'] = self.tank_V + design['Single vault volume'] = self.vault_V + design['Treatment time'] = self.treatment_tau + design['Treatment volume'] = self.treatment_V + + density = self.density_dct + constr = self.construction + constr[0].quantity = 200 * N # cement + constr[1].quantity = 0.6 * density['Sand'] * N + constr[2].quantity = 0.2 * density['Gravel'] * N + constr[3].quantity = 682*0.0024 * density['Brick'] * N + constr[4].quantity = 4 * density['Plastic'] * N + constr[5].quantity = 0.00351 * density['Steel'] * N + constr[6].quantity = 28.05*density['StainlessSteelSheet'] * N + constr[7].quantity = 0.222 * N # wood + + self.add_construction(add_cost=False) + + + @property + def safety_factor(self): + return self._safety_factor + @safety_factor.setter + def safety_factor(self, i): + if i < 1: + raise ValueError(f'safety_factor must be larger than 1, not {i}.') + self._safety_factor = i + + @property + def collection_period(self): + '''[float] Time interval between storage tank collection, [d].''' + return self._collection_period + @collection_period.setter + def collection_period(self, i): + self._collection_period = i + + @property + def treatment_tau(self): + '''[float] Time for onsite treatment (if treating), [d].''' + return self._treatment_tau + @treatment_tau.setter + def treatment_tau(self, i): + self._treatment_tau = i + + @property + def treatment_V(self): + '''[float] Volume needed to achieve treatment target (if treating), [d].''' + return self._treatment_V + @treatment_V.setter + def treatment_V(self, i): + self._treatment_V = i + + @property + def tank_V(self): + '''[float] Volume of the urine storage tank, [m3].''' + return self._tank_V + @tank_V.setter + def tank_V(self, i): + self._tank_V = i + + @property + def vault_V(self): + '''[float] Volume of the feces dehydration vault, [m3].''' + return self._vault_V + + @property + def struvite_pKsp(self): + '''[float] Precipitation constant of struvite.''' + return self._struvite_pKsp + @struvite_pKsp.setter + def struvite_pKsp(self, i): + self._struvite_pKsp = i + + @property + def prep_sludge(self): + ''' + [float] + Fraction of total precipitate appearing as sludge that can + settle and be removed. + ''' + return self._prep_sludge + @prep_sludge.setter + def prep_sludge(self, i): + self._prep_sludge = i + + @property + def log_removal(self): + '''Desired level of pathogen inactivation.''' + return self._log_removal + @log_removal.setter + def log_removal(self, i): + self._log_removal = i + + @property + def ur_pH(self): + '''Urine pH.''' + return self._ur_pH + @ur_pH.setter + def ur_pH(self, i): + self._ur_pH = i + + @property + def fec_moi_min(self): + '''[float] Minimum moisture content of feces.''' + return self._fec_moi_min + @fec_moi_min.setter + def fec_moi_min(self, i): + self._fec_moi_min = i + + @property + def fec_moi_red_rate(self): + '''[float] Exponential reduction rate of feces moisture.''' + return self._fec_moi_red_rate + @fec_moi_red_rate.setter + def fec_moi_red_rate(self, i): + self._fec_moi_red_rate = i + + @property + def tau(self): + '''[float] Retention time of the unit, same as `collection_period`.''' + return self.collection_period + @tau.setter + def tau(self, i): + self.collection_period = i \ No newline at end of file diff --git a/qsdsan/sanunits/_treatment_bed.py b/qsdsan/sanunits/_treatment_bed.py new file mode 100644 index 00000000..597add0d --- /dev/null +++ b/qsdsan/sanunits/_treatment_bed.py @@ -0,0 +1,412 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +''' +QSDsan: Quantitative Sustainable Design for sanitation and resource recovery systems + +This module is developed by: + Yalin Li + +This module is under the University of Illinois/NCSA Open Source License. +Please refer to https://github.com/QSD-Group/QSDsan/blob/main/LICENSE.txt +for license details. +''' + +import numpy as np +from math import ceil +from warnings import warn +from .. import SanUnit, Construction +from ..processes._decay import Decay +from ..utils import ospath, load_data, data_path, dct_from_str + +__all__ = ('DryingBed', 'LiquidTreatmentBed') + + +# %% + +drying_bed_path = ospath.join(data_path, 'sanunit_data/_drying_bed.tsv') + +class DryingBed(SanUnit, Decay): + ''' + Unplanted and planted drying bed for solids based on + `Trimmer et al. `_ + + To enable life cycle assessment, the following impact items should be pre-constructed: + `Concrete`, `Steel`. + + Parameters + ---------- + ins : WasteStream + Solid for drying. + outs : WasteStream + Dried solids, evaporated water, fugitive CH4, and fugitive N2O. + design_type : str + Can be "unplanted" or "planted". The default unplanted process has + a number of "covered", "uncovered", and "storage" beds. The storage + bed is similar to the covered bed, but with higher wall height. + + Examples + -------- + `bwaise systems `_ + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.processes.Decay ` + ''' + _N_ins = 1 + _N_outs = 4 + _units = { + 'Single covered bed volume': 'm3', + 'Single uncovered bed volume': 'm3', + 'Single storage bed volume': 'm3', + 'Single planted bed volume': 'm3', + 'Total cover area': 'm2', + 'Total column length': 'm' + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + design_type='unplanted', degraded_components=('OtherSS',), **kwargs): + Decay.__init__(self, ID, ins, outs, thermo=thermo, + init_with=init_with, F_BM_default=1, + degraded_components=degraded_components, + if_capture_biogas=False, + if_N2O_emission=True,) + N_unplanted = {'covered': 19, + 'uncovered': 30, + 'storage': 19, + 'planted': 0} + if design_type == 'unplanted': + self._N_bed = N_unplanted + self.design_type = 'unplanted' + else: + self._N_bed = dict.fromkeys(N_unplanted.keys(), 0) + self._N_bed['planted'] = 2 + self.design_type = 'planted' + + data = load_data(path=drying_bed_path) + for para in data.index: + if para == 'N_bed': continue + if para in ('sol_frac', 'bed_L', 'bed_W', 'bed_H'): + value = dct_from_str(data.loc[para]['expected'], dtype='float') + else: + value = float(data.loc[para]['expected']) + setattr(self, '_'+para, value) + del data + + for attr, value in kwargs.items(): + setattr(self, attr, value) + + def _init_lca(self): + self.construction = [ + Construction('concrete', linked_unit=self, item='Concrete', quantity_unit='m3'), + Construction('steel', linked_unit=self, item='Steel', quantity_unit='kg'), + ] + + def _run(self): + waste = self.ins[0] + sol, evaporated, CH4, N2O = self.outs + evaporated.phase = CH4.phase = N2O.phase = 'g' + + # COD/N degradation in settled solids + Decay._first_order_run(self, waste=waste, treated=sol) + sol_COD = sol._COD/1e3*sol.F_vol + + # Adjust water content in the dried solids + sol_frac = self.sol_frac + solid_content = 1 - sol.imass['H2O']/sol.F_mass + if solid_content > sol_frac: + msg = f'Solid content of the solid after COD removal is {solid_content:.2f}, '\ + f'larger than the set sol_frac of {sol_frac:.2f} for the {self.design_type} ' \ + 'process type, the set value is ignored.' + warn(msg, stacklevel=3) + evaporated.empty() + else: + sol.imass['H2O'] = (sol.F_mass-sol.imass['H2O'])/sol_frac + evaporated.imass['H2O'] = waste.imass['H2O'] - sol.imass['H2O'] + sol._COD = sol_COD*1e3/sol.F_vol + + + def _design(self): + design = self.design_results + + L = np.fromiter(self.bed_L.values(), dtype=float) + W = np.fromiter(self.bed_W.values(), dtype=float) + H = np.fromiter(self.bed_H.values(), dtype=float) + V = L * W * H + N_bed = self.N_bed + N = np.fromiter(N_bed.values(), dtype=int) + for n, i in enumerate(N_bed.keys()): + design[f'Number of {i} bed'] = N_bed[i] + design[f'Single {i} bed volume'] = V[n] + cover_array = np.array((1, 0, 1, 0)) # covered, uncovered, storage, planted + design['Total cover area'] = tot_cover_area = \ + (cover_array*N*L*W/(np.cos(self.cover_slope/180*np.pi))).sum() + design['Total column length'] = tot_column_length = \ + (cover_array*N*2*self.column_per_side*self.column_H).sum() + + concrete = (N*self.concrete_thickness*(L*W+2*L*H+2*W*H)).sum() + steel = tot_cover_area*self.cover_unit_mass + \ + tot_column_length*self.column_unit_mass + + constr = self.construction + constr[0].quantity = concrete + constr[1].quantity = steel + + for i in self.construction: + self.F_BM[i.item.ID] = 1 + + @property + def tau(self): + '''[float] Retention time, [d].''' + return self._tau + @tau.setter + def tau(self, i): + self._tau = float(i) + + @property + def sol_frac(self): + '''[float] Final solid content of the dried solids.''' + return self._sol_frac[self.design_type] + @sol_frac.setter + def sol_frac(self, i): + self._sol_frac[self.design_type] = float(i) + + @property + def design_type(self): + '''[str] Drying bed type, can be either "unplanted" or "planted".''' + return self._design_type + @design_type.setter + def design_type(self, i): + if i in ('unplanted', 'planted'): + self._design_type = i + self.line =f'{i.capitalize()} drying bed' + else: + raise ValueError(f'design_type can only be "unplanted" or "planted", ' + f"not {i}.") + + @property + def N_bed(self): + ''' + [dict] Number of the different types of drying beds, + float will be converted to the smallest integer. + ''' + for i, j in self._N_bed.items(): + self._N_bed[i] = ceil(j) + return self._N_bed + @N_bed.setter + def N_bed(self, i): + int_i = {k: ceil(v) for k, v in i.items()} + self._N_bed.update(int_i) + + @property + def bed_L(self): + '''[dict] Length of the different types of drying beds, [m].''' + return self._bed_L + @bed_L.setter + def bed_L(self, i): + self._bed_L.update(i) + + @property + def bed_W(self): + '''[dict] Width of the different types of drying beds, [m].''' + return self._bed_W + @bed_W.setter + def bed_W(self, i): + self._bed_W.update(i) + + @property + def bed_H(self): + '''[dict] Wall height of the different types of drying beds, [m].''' + return self._bed_H + @bed_H.setter + def bed_H(self, i): + self._bed_H.update(i) + + @property + def column_H(self): + '''[float] Column height for covered bed, [m].''' + return self._column_H + @column_H.setter + def column_H(self, i): + self._column_H = float(i) + + @property + def column_per_side(self): + '''[int] Number of columns per side of covered bed, float will be converted to the smallest integer.''' + return self._column_per_side + @column_per_side.setter + def column_per_side(self, i): + self._column_per_side = ceil(i) + + @property + def column_unit_mass(self): + '''[float] Unit mass of the column, [kg/m].''' + return self._column_unit_mass + @column_unit_mass.setter + def column_unit_mass(self, i): + self._column_unit_mass = float(i) + + @property + def concrete_thickness(self): + '''[float] Thickness of the concrete wall.''' + return self._concrete_thickness + @concrete_thickness.setter + def concrete_thickness(self, i): + self._concrete_thickness = float(i) + + @property + def cover_slope(self): + '''[float] Slope of the bed cover, [°].''' + return self._cover_slope + @cover_slope.setter + def cover_slope(self, i): + self._cover_slope = float(i) + + @property + def cover_unit_mass(self): + '''[float] Unit mass of the bed cover, [kg/m2].''' + return self._cover_unit_mass + @cover_unit_mass.setter + def cover_unit_mass(self, i): + self._cover_unit_mass = float(i) + + +# %% + +liquid_bed_path = ospath.join(data_path, 'sanunit_data/_liquid_treatment_bed.tsv') + + +class LiquidTreatmentBed(SanUnit, Decay): + ''' + For secondary treatment of liquid based on + `Trimmer et al. `_ + + To enable life cycle assessment, the following impact items should be pre-constructed: + Concrete. + + Parameters + ---------- + ins : WasteStream + Waste for treatment. + outs : WasteStream + Treated waste, fugitive CH4, and fugitive N2O. + + Examples + -------- + `bwaise systems `_ + + References + ---------- + [1] Trimmer et al., Navigating Multidimensional Social–Ecological System + Trade-Offs across Sanitation Alternatives in an Urban Informal Settlement. + Environ. Sci. Technol. 2020, 54 (19), 12641–12653. + https://doi.org/10.1021/acs.est.0c03296. + + See Also + -------- + :ref:`qsdsan.processes.Decay ` + ''' + _N_ins = 1 + _N_outs = 3 + _run = Decay._first_order_run + _units = { + 'Residence time': 'd', + 'Bed length': 'm', + 'Bed width': 'm', + 'Bed height': 'm', + 'Single bed volume': 'm3' + } + + def __init__(self, ID='', ins=None, outs=(), thermo=None, init_with='WasteStream', + **kwargs): + Decay.__init__(self, ID, ins, outs, thermo=thermo, + init_with=init_with, F_BM_default=1, + if_capture_biogas=False, + if_N2O_emission=False,) + + data = load_data(path=liquid_bed_path) + for para in data.index: + value = float(data.loc[para]['expected']) + setattr(self, '_'+para, value) + del data + + for attr, value in kwargs.items(): + setattr(self, attr, value) + + def _init_lca(self): + self.construction = [ + Construction('concrete', linked_unit=self, item='Concrete', quantity_unit='m3'), + ] + + def _design(self): + design = self.design_results + design['Residence time'] = self.tau + design['Bed number'] = N = self.N_bed + design['Bed length'] = L = self.bed_L + design['Bed width'] = W = self.bed_W + design['Bed height'] = H = self.bed_H + design['Single bed volume'] = L*W*H + + concrete = N*self.concrete_thickness*(L*W+2*L*H+2*W*H) + self.construction[0].quantity = concrete + self.add_construction() + + + def _cost(self): + pass + + + @property + def tau(self): + '''[float] Residence time, [d].''' + return self._tau + @tau.setter + def tau(self, i): + self._tau = i + + @property + def N_bed(self): + '''[int] Number of treatment beds, float will be converted to the smallest integer.''' + return self._N_bed + @N_bed.setter + def N_bed(self, i): + self._N_bed = ceil(i) + + @property + def bed_L(self): + '''[float] Bed length, [m].''' + return self._bed_L + @bed_L.setter + def bed_L(self, i): + self._bed_L = i + + @property + def bed_W(self): + '''[float] Bed width, [m].''' + return self._bed_W + @bed_W.setter + def bed_W(self, i): + self._bed_W = i + + @property + def bed_H(self): + '''[float] Bed height, [m].''' + return self._bed_H + @bed_H.setter + def bed_H(self, i): + self._bed_H = i + + @property + def concrete_thickness(self): + '''[float] Thickness of the concrete wall.''' + return self._concrete_thickness + @concrete_thickness.setter + def concrete_thickness(self, i): + self._concrete_thickness = i \ No newline at end of file