Processes

Processes: stocks of material in engineered flows.

GrainStoreElement(label, visual=None, *, stock=1000.0, on_order=10000.0, desired_stock=6000.0, release_fraction=0.025, receiving_delay=10.0, adjustment_delay=5.0)

Bases: Element

Retail grain store, whose stock is restocked from grain on order through a receiving delay.

Grain on order arrives in the stock at the receiving rate on_order / receiving_delay, and the stock releases the fraction release_fraction of itself per unit time to the consumers. The procurement rate (desired_stock - stock) / adjustment_delay adds grain on order, and it is negative when the stock exceeds the desired stock, which cancels orders.

Parameters:

Name	Type	Description	Default
label	str	The label of the store.	required
visual	VisualSpec | None	How the store is drawn, or None for default_visual_.	None
stock	Annotated[float, StateSpec(quantity=grain)]	The initial grain stock.	1000.0
on_order	Annotated[float, StateSpec(quantity=grain)]	The initial grain on order.	10000.0
desired_stock	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The desired grain stock.	6000.0
release_fraction	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The fraction of the stock released per unit time.	0.025
receiving_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The average time for grain on order to arrive.	10.0
adjustment_delay	Annotated[float, ParameterSpec(domain=open(0, oo))]	The time to close the gap between the desired and the current stock.	5.0

Examples:

>>> from skcomplex.systems.engineering.processes import GrainStoreElement
>>> GrainStoreElement('store').desired_stock
6000.0

Source code in src/skcomplex/systems/engineering/processes/_grain_storage.py

def __init__(
    self,
    label: str,
    visual: VisualSpec | None = None,
    *,
    stock: Annotated[float, StateSpec(quantity='grain')] = 1000.0,
    on_order: Annotated[float, StateSpec(quantity='grain')] = 10000.0,
    desired_stock: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 6000.0,
    release_fraction: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.025,
    receiving_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 10.0,
    adjustment_delay: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 5.0,
) -> None:
    super().__init__(label, visual=visual)
    self.stock = stock
    self.on_order = on_order
    self.desired_stock = desired_stock
    self.release_fraction = release_fraction
    self.receiving_delay = receiving_delay
    self.adjustment_delay = adjustment_delay

bathtub(water=50.0, *, inflow=5.0, outflow=5.0)

Return a bathtub, one stock of water with an inflow from the faucet and an outflow through the drain.

The drain empties the tub at outflow gallons per minute while there is water, and the faucet opens to inflow gallons per minute after five minutes. With equal flows the level falls for five minutes and then stays constant.

Parameters:

Name	Type	Description	Default
water	float	The initial water in the tub, in gallons.	50.0
inflow	float	The inflow after five minutes, in gallons per minute.	5.0
outflow	float	The outflow, in gallons per minute.	5.0

Returns:

Name	Type	Description
system	System	The system, with the water tub.water in gallons and the flows tub.faucet and tub.drain in gallons per minute, over minutes.

Examples:

>>> from skcomplex.systems.engineering.processes import bathtub
>>> bathtub(water=20.0).elements[0].water
20.0

Source code in src/skcomplex/systems/engineering/processes/_processes.py

@register_system(
    field='engineering/processes',
    tags=('stocks and flows', 'system dynamics', 'ode'),
    references=('Meadows 2008',),
    defaults={'until': 10.0, 'dt': 0.5},
)
def bathtub(water: float = 50.0, *, inflow: float = 5.0, outflow: float = 5.0) -> System:
    """Return a bathtub, one stock of water with an inflow from the faucet and an outflow through the drain.

    The drain empties the tub at `outflow` gallons per minute while there is water, and the faucet opens to `inflow`
    gallons per minute after five minutes. With equal flows the level falls for five minutes and then stays constant.

    Args:
        water:
            The initial water in the tub, in gallons.

        inflow:
            The inflow after five minutes, in gallons per minute.

        outflow:
            The outflow, in gallons per minute.

    Returns:
        system:
            The system, with the water `tub.water` in gallons and the flows `tub.faucet` and `tub.drain` in gallons
            per minute, over minutes.

    Examples:
        >>> from skcomplex.systems.engineering.processes import bathtub
        >>> bathtub(water=20.0).elements[0].water
        20.0
    """
    return System(elements=[TubElement('tub', water=water, inflow=inflow, outflow=outflow)])

cascade_classifier(n_stages=10, *, k=0.6, feed_stage=None, feed=1.0)

Return a cascade classifier, a column of stages that separates a narrow size class of particles by an air flow.

Material leaves every stage at unit rate, a fraction k upwards and 1 - k downwards. The top stage releases into the fine product and the bottom stage into the coarse product. Material of one size class is fed into one stage at a constant rate. At steady state the share of the feed that ends in the fine product is F_f = (1 - ((1 - k) / k)^(z + 1 - i)) / (1 - ((1 - k) / k)^(z + 1)) for z stages and feeding at stage i counted from the top, Barsky's fractional extraction.

Parameters:

Name	Type	Description	Default
n_stages	int	The number of stages z.	10
k	float	The distribution coefficient, the fraction of material moving up from a stage.	0.6
feed_stage	int | None	The stage that receives the feed, counted from 1 at the top, or None for the middle stage.	None
feed	float	The feed rate.	1.0

Returns:

Name	Type	Description
system	System	The system, with the material stage_<i>.m of each stage from the top, and the accumulated products fine.m and coarse.m.

Examples:

>>> from skcomplex.systems.engineering.processes import cascade_classifier
>>> len(cascade_classifier(n_stages=4).interactions)
8

Source code in src/skcomplex/systems/engineering/processes/_processes.py

@register_system(
    field='engineering/processes',
    tags=('separation', 'cascade', 'stocks and flows', 'ode'),
    references=('Barsky 2020',),
    defaults={'until': 50.0, 'dt': 0.5},
)
def cascade_classifier(
    n_stages: int = 10,
    *,
    k: float = 0.6,
    feed_stage: int | None = None,
    feed: float = 1.0,
) -> System:
    """Return a cascade classifier, a column of stages that separates a narrow size class of particles by an air flow.

    Material leaves every stage at unit rate, a fraction `k` upwards and `1 - k` downwards. The top stage releases
    into the fine product and the bottom stage into the coarse product. Material of one size class is fed into one
    stage at a constant rate. At steady state the share of the feed that ends in the fine product is
    `F_f = (1 - ((1 - k) / k)^(z + 1 - i)) / (1 - ((1 - k) / k)^(z + 1))` for `z` stages and feeding at stage `i`
    counted from the top, Barsky's fractional extraction.

    Args:
        n_stages:
            The number of stages `z`.

        k:
            The distribution coefficient, the fraction of material moving up from a stage.

        feed_stage:
            The stage that receives the feed, counted from 1 at the top, or `None` for the middle stage.

        feed:
            The feed rate.

    Returns:
        system:
            The system, with the material `stage_<i>.m` of each stage from the top, and the accumulated products
            `fine.m` and `coarse.m`.

    Examples:
        >>> from skcomplex.systems.engineering.processes import cascade_classifier
        >>> len(cascade_classifier(n_stages=4).interactions)
        8
    """
    entry = (n_stages + 1) // 2 if feed_stage is None else feed_stage
    stages = [StageElement(f'stage_{index}', feed=feed if index == entry else 0.0) for index in range(1, n_stages + 1)]
    labels = ['fine', *(stage.label for stage in stages), 'coarse']
    interactions = [
        TransferInteraction(
            f'up_{index}',
            source=labels[index],
            target=labels[index - 1],
            state='m',
            amount=lambda s: k * s.source.m,
        )
        for index in range(1, n_stages + 1)
    ] + [
        TransferInteraction(
            f'down_{index}',
            source=labels[index],
            target=labels[index + 1],
            state='m',
            amount=lambda s: (1 - k) * s.source.m,
        )
        for index in range(1, n_stages + 1)
    ]
    return System(elements=[StageElement('fine'), *stages, StageElement('coarse')], interactions=interactions)

food_relief(affected=100000.0, *, ration=0.0005, onset=10.0, shipping_delay=14.0, cover=7.0, stock_time=7.0)

Return a food relief system, a pipeline that ships food to a warehouse and distributes it to people in need.

A disaster at onset leaves affected people who need a daily ration. The relief agency ships food at the need plus corrections that close the gaps between the warehouse stock and its desired cover of the need, and between the food in transit and the need over the shipping delay. Shipped food enters the pipeline and arrives at the warehouse through a third-order material delay. The warehouse distributes the need while its stock lasts. Food is conserved, so almost nothing reaches the people in the first days after the disaster, and the gap closes as the pipeline fills. The structure follows the relief food example of Bala, Arshad and Noh (2017, Section 4.11), with illustrative parameters in tonnes and days.

Parameters:

Name	Type	Description	Default
affected	float	The number of people in need after the disaster.	100000.0
ration	float	The daily ration in tonnes per person.	0.0005
onset	float	The day of the disaster.	10.0
shipping_delay	float	The average time from shipping to arrival, in days.	14.0
cover	float	The desired warehouse stock in days of need.	7.0
stock_time	float	The time to correct the warehouse stock, in days.	7.0

Returns:

Name	Type	Description
system	System	The system, with the food stocks relief.in_transit and relief.warehouse in tonnes, and the flows relief.shipping, relief.arrivals, and relief.distribution in tonnes per day, over days. The observables of the system are relief.need and relief.shipments.

Examples:

>>> from skcomplex.systems.engineering.processes import food_relief
>>> food_relief(shipping_delay=7.0).elements[0].shipping_delay
7.0

Source code in src/skcomplex/systems/engineering/processes/_systems.py

@register_system(
    field='engineering/processes',
    tags=('humanitarian logistics', 'supply chain', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Bala et al. 2017',),
    defaults={'until': 120.0, 'dt': 0.5},
    observables=lambda system: [
        ObservableSpec('need', ReliefPipelineElement._need, element='relief'),
        ObservableSpec('shipments', ReliefPipelineElement._shipments, element='relief'),
    ],
)
def food_relief(
    affected: float = 1e5,
    *,
    ration: float = 5e-4,
    onset: float = 10.0,
    shipping_delay: float = 14.0,
    cover: float = 7.0,
    stock_time: float = 7.0,
) -> System:
    """Return a food relief system, a pipeline that ships food to a warehouse and distributes it to people in need.

    A disaster at `onset` leaves `affected` people who need a daily ration. The relief agency ships food at the need
    plus corrections that close the gaps between the warehouse stock and its desired cover of the need, and between
    the food in transit and the need over the shipping delay. Shipped food enters the pipeline and arrives at the
    warehouse through a third-order material delay. The warehouse distributes the need while its stock lasts. Food is
    conserved, so almost nothing reaches the people in the first days after the disaster, and the gap closes as the
    pipeline fills. The structure follows the relief food example of Bala, Arshad and Noh (2017, Section 4.11), with
    illustrative parameters in tonnes and days.

    Args:
        affected:
            The number of people in need after the disaster.

        ration:
            The daily ration in tonnes per person.

        onset:
            The day of the disaster.

        shipping_delay:
            The average time from shipping to arrival, in days.

        cover:
            The desired warehouse stock in days of need.

        stock_time:
            The time to correct the warehouse stock, in days.

    Returns:
        system:
            The system, with the food stocks `relief.in_transit` and `relief.warehouse` in tonnes, and the flows
            `relief.shipping`, `relief.arrivals`, and `relief.distribution` in tonnes per day, over days. The
            observables of the system are `relief.need` and `relief.shipments`.

    Examples:
        >>> from skcomplex.systems.engineering.processes import food_relief
        >>> food_relief(shipping_delay=7.0).elements[0].shipping_delay
        7.0
    """
    relief = ReliefPipelineElement(
        'relief',
        affected=affected,
        ration=ration,
        onset=onset,
        shipping_delay=shipping_delay,
        cover=cover,
        stock_time=stock_time,
    )
    return System(elements=[relief])

grain_storage(stock=1000.0, on_order=10000.0, *, desired_stock=6000.0, release_fraction=0.025, receiving_delay=10.0, adjustment_delay=5.0)

Return a retail grain store, whose orders through a supply line make its stock oscillate.

The grain stock gains the receiving rate on_order / receiving_delay and loses the release rate release_fraction stock, and the grain on order gains the procurement rate (desired_stock - stock) / adjustment_delay and loses the receiving rate. The orders ignore the grain already on order, so the supply line overshoots and the stock, the receiving rate, and the procurement oscillate in weeks before they settle, with the stock below the desired one at desired_stock / (1 + release_fraction adjustment_delay), 5333 tonnes for the defaults, where the release balances the receiving. The defaults are those of the grain storage example of Bala, Arshad and Noh (2017), in tonnes and weeks.

Parameters:

Name	Type	Description	Default
stock	float	The initial grain stock.	1000.0
on_order	float	The initial grain on order.	10000.0
desired_stock	float	The desired grain stock.	6000.0
release_fraction	float	The fraction of the stock released per week.	0.025
receiving_delay	float	The average time for grain on order to arrive, in weeks.	10.0
adjustment_delay	float	The time to close the gap between the desired and the current stock, in weeks.	5.0

Returns:

Name	Type	Description
system	System	The system, with the stocks store.stock and store.on_order in tonnes, and the flows store.procurement, store.receiving, and store.release in tonnes per week, over weeks.

Examples:

>>> from skcomplex.systems.engineering.processes import grain_storage
>>> grain_storage(receiving_delay=5.0).elements[0].receiving_delay
5.0

Source code in src/skcomplex/systems/engineering/processes/_grain_storage.py

@register_system(
    field='engineering/processes',
    tags=('agriculture', 'storage', 'stocks and flows', 'system dynamics', 'oscillation', 'ode'),
    references=('Bala et al. 2017',),
    defaults={'until': 100.0, 'dt': 0.25},
)
def grain_storage(
    stock: float = 1000.0,
    on_order: float = 10000.0,
    *,
    desired_stock: float = 6000.0,
    release_fraction: float = 0.025,
    receiving_delay: float = 10.0,
    adjustment_delay: float = 5.0,
) -> System:
    """Return a retail grain store, whose orders through a supply line make its stock oscillate.

    The grain stock gains the receiving rate `on_order / receiving_delay` and loses the release rate
    `release_fraction stock`, and the grain on order gains the procurement rate
    `(desired_stock - stock) / adjustment_delay` and loses the receiving rate. The orders ignore the grain already on
    order, so the supply line overshoots and the stock, the receiving rate, and the procurement oscillate in weeks
    before they settle, with the stock below the desired one at
    `desired_stock / (1 + release_fraction adjustment_delay)`, 5333 tonnes for the defaults, where the release
    balances the receiving. The defaults are those of the grain storage example of Bala, Arshad and Noh (2017), in
    tonnes and weeks.

    Args:
        stock:
            The initial grain stock.

        on_order:
            The initial grain on order.

        desired_stock:
            The desired grain stock.

        release_fraction:
            The fraction of the stock released per week.

        receiving_delay:
            The average time for grain on order to arrive, in weeks.

        adjustment_delay:
            The time to close the gap between the desired and the current stock, in weeks.

    Returns:
        system:
            The system, with the stocks `store.stock` and `store.on_order` in tonnes, and the flows
            `store.procurement`, `store.receiving`, and `store.release` in tonnes per week, over weeks.

    Examples:
        >>> from skcomplex.systems.engineering.processes import grain_storage
        >>> grain_storage(receiving_delay=5.0).elements[0].receiving_delay
        5.0
    """
    store = GrainStoreElement(
        'store',
        stock=stock,
        on_order=on_order,
        desired_stock=desired_stock,
        release_fraction=release_fraction,
        receiving_delay=receiving_delay,
        adjustment_delay=adjustment_delay,
    )
    return System(elements=[store])

irrigation(*, river_inflow=0.5, seasonality=0.9, reservoir_capacity=80.0, potential_et=0.4, canal_loss=0.2, pump_capacity=0.2)

Return an irrigation system, a reservoir and an aquifer that keep the root zone of the crops supplied.

A river fills the reservoir with a yearly cycle, and the reservoir loses water to evaporation and spills above its capacity. The irrigation demand is the potential crop evapotranspiration plus the refill of the root zone towards a target fraction of its field capacity. The reservoir releases the demand grossed up for the canal losses, as far as its stock allows, and the losses seep into the aquifer. When the release falls short, the farmers pump the unmet demand from the aquifer up to the pump capacity. The crops draw water from the root zone at the potential rate scaled by a water-stress table, and the root zone percolates to the aquifer, which drains as baseflow. Water is conserved, so in the dry season the reservoir falls, the pumping rises, and the aquifer is drawn down. The structure follows the irrigation example of Bala, Arshad and Noh (2017, Section 4.11), with illustrative parameters in millions of cubic metres and days.

Parameters:

Name	Type	Description	Default
river_inflow	float	The mean river inflow to the reservoir, per day.	0.5
seasonality	float	The relative amplitude of the yearly inflow cycle.	0.9
reservoir_capacity	float	The capacity of the reservoir.	80.0
potential_et	float	The crop water requirement, the potential evapotranspiration of the irrigated area per day.	0.4
canal_loss	float	The fraction of the release lost from the canals to the aquifer.	0.2
pump_capacity	float	The largest groundwater pumping per day.	0.2

Returns:

Name	Type	Description
system	System	The system, with the water stocks scheme.reservoir, scheme.soil, and scheme.groundwater, and the flows scheme.inflow, scheme.evaporation, scheme.spill, scheme.irrigation, scheme.seepage, scheme.groundwater_pumping, scheme.crop_water, scheme.percolation, and scheme.baseflow per day, over days. The observables of the system are scheme.demand, scheme.release, and scheme.pumping.

Examples:

>>> from skcomplex.systems.engineering.processes import irrigation
>>> irrigation(pump_capacity=0.0).elements[0].pump_capacity
0.0

Source code in src/skcomplex/systems/engineering/processes/_systems.py

@register_system(
    field='engineering/processes',
    tags=('water resources', 'agriculture', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Bala et al. 2017',),
    defaults={'until': 730.0, 'dt': 1.0},
    observables=lambda system: [
        ObservableSpec('demand', IrrigationSchemeElement._demand, element='scheme'),
        ObservableSpec('release', IrrigationSchemeElement._reservoir_release, element='scheme'),
        ObservableSpec('pumping', IrrigationSchemeElement._pumping, element='scheme'),
    ],
)
def irrigation(
    *,
    river_inflow: float = 0.5,
    seasonality: float = 0.9,
    reservoir_capacity: float = 80.0,
    potential_et: float = 0.4,
    canal_loss: float = 0.2,
    pump_capacity: float = 0.2,
) -> System:
    """Return an irrigation system, a reservoir and an aquifer that keep the root zone of the crops supplied.

    A river fills the reservoir with a yearly cycle, and the reservoir loses water to evaporation and spills above its
    capacity. The irrigation demand is the potential crop evapotranspiration plus the refill of the root zone towards
    a target fraction of its field capacity. The reservoir releases the demand grossed up for the canal losses, as far
    as its stock allows, and the losses seep into the aquifer. When the release falls short, the farmers pump the
    unmet demand from the aquifer up to the pump capacity. The crops draw water from the root zone at the potential
    rate scaled by a water-stress table, and the root zone percolates to the aquifer, which drains as baseflow. Water
    is conserved, so in the dry season the reservoir falls, the pumping rises, and the aquifer is drawn down. The
    structure follows the irrigation example of Bala, Arshad and Noh (2017, Section 4.11), with illustrative
    parameters in millions of cubic metres and days.

    Args:
        river_inflow:
            The mean river inflow to the reservoir, per day.

        seasonality:
            The relative amplitude of the yearly inflow cycle.

        reservoir_capacity:
            The capacity of the reservoir.

        potential_et:
            The crop water requirement, the potential evapotranspiration of the irrigated area per day.

        canal_loss:
            The fraction of the release lost from the canals to the aquifer.

        pump_capacity:
            The largest groundwater pumping per day.

    Returns:
        system:
            The system, with the water stocks `scheme.reservoir`, `scheme.soil`, and `scheme.groundwater`, and the
            flows `scheme.inflow`, `scheme.evaporation`, `scheme.spill`, `scheme.irrigation`, `scheme.seepage`,
            `scheme.groundwater_pumping`, `scheme.crop_water`, `scheme.percolation`, and `scheme.baseflow` per day,
            over days. The observables of the system are `scheme.demand`, `scheme.release`, and `scheme.pumping`.

    Examples:
        >>> from skcomplex.systems.engineering.processes import irrigation
        >>> irrigation(pump_capacity=0.0).elements[0].pump_capacity
        0.0
    """
    scheme = IrrigationSchemeElement(
        'scheme',
        river_inflow=river_inflow,
        seasonality=seasonality,
        reservoir_capacity=reservoir_capacity,
        potential_et=potential_et,
        canal_loss=canal_loss,
        pump_capacity=pump_capacity,
    )
    return System(elements=[scheme])

rice_milling(*, harvest=2000.0, seasonality=0.8, procurement_delay=1.0, outturn=0.65, capacity=2000.0, expansion_time=12.0)

Return a rice milling system, a mill that buys paddy, mills it into rice and by-products, and sells them.

Paddy arrives from the farms through a third-order procurement delay, at a rate that follows the harvest season around its mean. The mill processes its paddy stock in a month, as far as its milling capacity allows, and splits the milled paddy by mass into rice, the outturn, husk, bran, and milling losses. Rice, husk, and bran leave their inventories for the market at a fixed coverage. The capacity adjusts to the smoothed indicated milling rate, so in the harvest season the capacity lags and paddy accumulates, and in the lean season the capacity idles. Grain is conserved, every change of the paddy, rice, husk, and bran stocks is a flow, and over a year the milled paddy and the rice sales match the mean harvest and its outturn. The structure follows the rice milling case of Bala, Arshad and Noh (2017), with illustrative parameters in tonnes and months, since the case data are specific to their study.

Parameters:

Name	Type	Description	Default
harvest	float	The mean paddy procurement in tonnes per month.	2000.0
seasonality	float	The relative amplitude of the yearly harvest cycle.	0.8
procurement_delay	float	The average time for purchased paddy to reach the mill, in months.	1.0
outturn	float	The fraction of the milled paddy that becomes rice.	0.65
capacity	float	The initial milling capacity in tonnes of paddy per month.	2000.0
expansion_time	float	The time for the capacity to adjust to the perceived milling rate, in months.	12.0

Returns:

Name	Type	Description
system	System	The system, with the grain stocks mill.paddy, mill.rice, mill.husk, and mill.bran in tonnes, the capacity mill.capacity, and the flows mill.paddy_purchases, mill.milled_rice, mill.husk_output, mill.bran_output, mill.milling_losses, mill.rice_sales, mill.husk_sales, and mill.bran_sales in tonnes per month, over months. The observables of the system are mill.procurement, mill.indicated_milling, and mill.milling.

Examples:

>>> from skcomplex.systems.engineering.processes import rice_milling
>>> rice_milling(capacity=1500.0).elements[0].capacity
1500.0

Source code in src/skcomplex/systems/engineering/processes/_systems.py

@register_system(
    field='engineering/processes',
    tags=('agriculture', 'milling', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Bala et al. 2017',),
    defaults={'until': 120.0, 'dt': 0.25},
    observables=lambda system: [
        ObservableSpec('procurement', RiceMillElement._procurement, element='mill'),
        ObservableSpec('indicated_milling', RiceMillElement._indicated_milling, element='mill'),
        ObservableSpec('milling', RiceMillElement._milling, element='mill'),
    ],
)
def rice_milling(
    *,
    harvest: float = 2000.0,
    seasonality: float = 0.8,
    procurement_delay: float = 1.0,
    outturn: float = 0.65,
    capacity: float = 2000.0,
    expansion_time: float = 12.0,
) -> System:
    """Return a rice milling system, a mill that buys paddy, mills it into rice and by-products, and sells them.

    Paddy arrives from the farms through a third-order procurement delay, at a rate that follows the harvest season
    around its mean. The mill processes its paddy stock in a month, as far as its milling capacity allows, and splits
    the milled paddy by mass into rice, the outturn, husk, bran, and milling losses. Rice, husk, and bran leave their
    inventories for the market at a fixed coverage. The capacity adjusts to the smoothed indicated milling rate, so in
    the harvest season the capacity lags and paddy accumulates, and in the lean season the capacity idles. Grain is
    conserved, every change of the paddy, rice, husk, and bran stocks is a flow, and over a year the milled paddy and
    the rice sales match the mean harvest and its outturn. The structure follows the rice milling case of Bala,
    Arshad and Noh (2017), with illustrative parameters in tonnes and months, since the case data are specific to
    their study.

    Args:
        harvest:
            The mean paddy procurement in tonnes per month.

        seasonality:
            The relative amplitude of the yearly harvest cycle.

        procurement_delay:
            The average time for purchased paddy to reach the mill, in months.

        outturn:
            The fraction of the milled paddy that becomes rice.

        capacity:
            The initial milling capacity in tonnes of paddy per month.

        expansion_time:
            The time for the capacity to adjust to the perceived milling rate, in months.

    Returns:
        system:
            The system, with the grain stocks `mill.paddy`, `mill.rice`, `mill.husk`, and `mill.bran` in tonnes, the
            capacity `mill.capacity`, and the flows `mill.paddy_purchases`, `mill.milled_rice`, `mill.husk_output`,
            `mill.bran_output`, `mill.milling_losses`, `mill.rice_sales`, `mill.husk_sales`, and `mill.bran_sales` in
            tonnes per month, over months. The observables of the system are `mill.procurement`,
            `mill.indicated_milling`, and `mill.milling`.

    Examples:
        >>> from skcomplex.systems.engineering.processes import rice_milling
        >>> rice_milling(capacity=1500.0).elements[0].capacity
        1500.0
    """
    mill = RiceMillElement(
        'mill',
        paddy=harvest,
        rice=outturn * harvest,
        husk=HUSK_SHARE * harvest,
        bran=BRAN_SHARE * harvest,
        capacity=capacity,
        harvest=harvest,
        seasonality=seasonality,
        procurement_delay=procurement_delay,
        outturn=outturn,
        expansion_time=expansion_time,
    )
    return System(elements=[mill])

solid_waste(population=1000000.0, *, growth_rate=0.02, per_capita=0.2, recycling_share=0.2, composting_share=0.3, landfill_capacity=2000000.0)

Return a solid waste system, the waste of a growing city collected, recycled, composted, and landfilled.

The population grows exponentially and generates waste in proportion to its size. The waste waits uncollected until the collection, limited by a collection capacity that the city expands towards the generation with a delay, removes it. The collected waste is recycled, composted, or sent to the landfill in fixed shares. Recycled material and compost leave for industry and farms. The landfill accepts waste until it nears its capacity, and a table function then turns the residual waste to open dumping. Waste is conserved, so the landfill fills up to its capacity and dumping takes over. The structure follows the solid waste management case of Bala, Arshad and Noh (2017), with illustrative parameters in tonnes and years for a city of a million people.

Parameters:

Name	Type	Description	Default
population	float	The initial population.	1000000.0
growth_rate	float	The yearly growth rate of the population.	0.02
per_capita	float	The waste generated per person, in tonnes per year.	0.2
recycling_share	float	The share of the collected waste that is recycled.	0.2
composting_share	float	The share of the collected waste that is composted.	0.3
landfill_capacity	float	The capacity of the landfill in tonnes.	2000000.0

Returns:

Name	Type	Description
system	System	The system, with the population city.population, the waste stocks city.uncollected, city.recycled, city.composted, city.landfill, and city.dumped in tonnes, the capacity city.collection_capacity, and the flows city.population_growth, city.waste_generation, city.recycling, city.composting, city.landfilling, city.dumping, city.recycled_sales, and city.compost_sales per year, over years. The observables of the system are city.generation, city.collection, and city.acceptance.

Examples:

>>> from skcomplex.systems.engineering.processes import solid_waste
>>> solid_waste(landfill_capacity=1e6).elements[0].landfill_capacity
1000000.0

Source code in src/skcomplex/systems/engineering/processes/_systems.py

@register_system(
    field='engineering/processes',
    tags=('waste management', 'urban', 'stocks and flows', 'system dynamics', 'ode'),
    references=('Bala et al. 2017',),
    defaults={'until': 40.0, 'dt': 0.1},
    observables=lambda system: [
        ObservableSpec('generation', WasteSystemElement._generation, element='city'),
        ObservableSpec('collection', WasteSystemElement._collection, element='city'),
        ObservableSpec('acceptance', WasteSystemElement._acceptance, element='city'),
    ],
)
def solid_waste(
    population: float = 1e6,
    *,
    growth_rate: float = 0.02,
    per_capita: float = 0.2,
    recycling_share: float = 0.2,
    composting_share: float = 0.3,
    landfill_capacity: float = 2e6,
) -> System:
    """Return a solid waste system, the waste of a growing city collected, recycled, composted, and landfilled.

    The population grows exponentially and generates waste in proportion to its size. The waste waits uncollected
    until the collection, limited by a collection capacity that the city expands towards the generation with a delay,
    removes it. The collected waste is recycled, composted, or sent to the landfill in fixed shares. Recycled material
    and compost leave for industry and farms. The landfill accepts waste until it nears its capacity, and a table
    function then turns the residual waste to open dumping. Waste is conserved, so the landfill fills up to its
    capacity and dumping takes over. The structure follows the solid waste management case of Bala, Arshad and Noh
    (2017), with illustrative parameters in tonnes and years for a city of a million people.

    Args:
        population:
            The initial population.

        growth_rate:
            The yearly growth rate of the population.

        per_capita:
            The waste generated per person, in tonnes per year.

        recycling_share:
            The share of the collected waste that is recycled.

        composting_share:
            The share of the collected waste that is composted.

        landfill_capacity:
            The capacity of the landfill in tonnes.

    Returns:
        system:
            The system, with the population `city.population`, the waste stocks `city.uncollected`, `city.recycled`,
            `city.composted`, `city.landfill`, and `city.dumped` in tonnes, the capacity `city.collection_capacity`,
            and the flows `city.population_growth`, `city.waste_generation`, `city.recycling`, `city.composting`,
            `city.landfilling`, `city.dumping`, `city.recycled_sales`, and `city.compost_sales` per year, over years.
            The observables of the system are `city.generation`, `city.collection`, and `city.acceptance`.

    Examples:
        >>> from skcomplex.systems.engineering.processes import solid_waste
        >>> solid_waste(landfill_capacity=1e6).elements[0].landfill_capacity
        1000000.0
    """
    city = WasteSystemElement(
        'city',
        population=population,
        collection_capacity=INITIAL_COLLECTION_SHARE * per_capita * population,
        growth_rate=growth_rate,
        per_capita=per_capita,
        recycling_share=recycling_share,
        composting_share=composting_share,
        landfill_capacity=landfill_capacity,
    )
    return System(elements=[city])