水资源管理主体

In progress

该部分文档仍在积极开发中。

水资源管理主体又包含不同尺度的市级主体 \(p\) 和省级主体 \(P\) 两类，都利用.shp地理空间数据创建，但辖区范围不同。省级主体管辖其空间范围内的所有市级主体，省级主体则管辖其空间范围内的所有农业灌溉主体 \(v\)。因此对作物 \(c\) 的灌溉面积 \(A\) 而言，有：

\[A_{P, c} = \sum_{p \in P} A_{p, c} = \sum_{v \in P} A_{v, c}\]

省级主体¶

Api

Bases: Manager

每个省的主体。

在黄河的水分配模型中，省起到的作用主要是向下一级分配水配额。因为“八七”分水方案是省尺度进行配额的，但省一级单位通常不了解有多少用水需求。因此，本模型假设省份控制总配额量，根据灌溉面积将其分配给各个地级市。

Source code in src/api/province.py

def __init__(self, *args, name: str, **kwargs) -> None:
    self.name_en = name
    super().__init__(*args, **kwargs)
    if name not in self.p.names:
        raise ValueError(f"Province {name} not in parameters.")

managed `property` ¶

managed

该省所管辖的市水资源管理单位。

quota `property` ¶

quota

计算当前的水资源配额，统计数据通常是亿立方米，需要转化单位为立方米。

water_prices `cached` `property` ¶

water_prices

水资源价格字典，分别指示地表水和地下水的价格。价格应该是一个正数，从数据中读取，单位是元/立方米。

'ground': 地下水价格
'surface': 地表水价格

create `classmethod` ¶

create(model, name_en)

使用单例模式创造一个省主体。

Parameters:

Name	Type	Description	Default
`model`	`MainModel`	当前模型。	required
`name_en`	`str`	省份的英文名。	required

Returns:

Type	Description
`Province`	一个省份的实例。

Source code in src/api/province.py

@classmethod
def create(cls, model: MainModel, name_en: str) -> Province:
    """使用单例模式创造一个省主体。

    Parameters:
        model:
            当前模型。
        name_en:
            省份的英文名。

    Returns:
        一个省份的实例。
    """
    # 对当前模型的每个省份，只有一个实例
    if model not in cls._instances:
        cls._instances[model] = {}
    # 如果已经有了这个省份的实例，直接返回
    if instance := cls._instances[model].get(name_en):
        return instance
    # 否则，创建一个新的实例
    instance: Province = model.agents.new(cls, singleton=True, name=name_en)
    cls._instances[model][name_en] = instance
    return instance

setup ¶

setup()

初始化一个省份的数据。

水资源配额数据。来源于黄河水资源分配方案。
属于该省份的主体数量数据。来源于统计局各省的乡村数量数据。

Source code in src/api/province.py

def setup(self):
    """初始化一个省份的数据。

    1. 水资源配额数据。来源于黄河水资源分配方案。
    2. 属于该省份的主体数量数据。来源于统计局各省的乡村数量数据。
    """
    self.add_dynamic_variable(
        name="quota",
        data=pd.read_csv(self.ds.quotas, index_col=0),
        function=update_province_csv,
    )
    self.add_dynamic_variable(
        name="num",
        data=pd.read_csv(self.ds.agents_num, index_col=0),
        function=update_province_csv,
    )

update_data ¶

update_data(agents_by='total_area', quotas_by='total_area')

更新配额数据和主体数量数据，按需分配给每个其管辖的地级市。权重变量应该是一个字符串，指向一个已经存在于所辖城市主体的动态变量。

Parameters:

Name	Type	Description	Default
`agents_by`	`str`	用于分配农民数量的权重变量。	`'total_area'`
`quotas_by`	`str`	用于分配水资源配额的权重变量。	`'total_area'`

Source code in src/api/province.py

def update_data(
    self,
    agents_by: str = "total_area",
    quotas_by: str = "total_area",
) -> None:
    """更新配额数据和主体数量数据，按需分配给每个其管辖的地级市。
    权重变量应该是一个字符串，指向一个已经存在于所辖城市主体的动态变量。

    Parameters:
        agents_by:
            用于分配农民数量的权重变量。
        quotas_by:
            用于分配水资源配额的权重变量。
    """
    scale = self.p.get("n_scale", 1)
    n_agents = int(self.dynamic_var("num") * scale)
    agents_arr = ceil_divide(n_agents, weights=self.managed.array(agents_by))
    self.managed.update("n_agents", agents_arr)
    self.assign(self.quota, "quota", by=quotas_by)
    # === logging ===
    logger.info(f"{self.name_en} assigned {n_agents} Farmers.")

update_graph ¶

update_graph(l_p=None, l_c=None)

更新社会网络。乡村是我们进行灌溉决策的基本单元，社会网络代表着不同乡村之间的联系。当乡村之间有联系时，他们可以感知到对方的决策信息。如果乡村 A 知道了乡村 B 大量超用水，那么乡村 A 可能会感到不满。这种不满会同时降低两者的社会满意程度。

Parameters:

Name	Type	Description	Default
`l_p`	`Optional[float]`	省之间，hub节点的概率。	`None`
`l_c`	`Optional[float]`	城市内部的联系概率。	`None`

Source code in src/api/province.py

def update_graph(
    self,
    l_p: Optional[float] = None,
    l_c: Optional[float] = None,
) -> list:
    """更新社会网络。
    乡村是我们进行灌溉决策的基本单元，
    社会网络代表着不同乡村之间的联系。
    当乡村之间有联系时，他们可以感知到对方的决策信息。
    如果乡村 A 知道了乡村 B 大量超用水，那么乡村 A 可能会感到不满。
    这种不满会同时降低两者的社会满意程度。

    Parameters:
        l_p:
            省之间，hub节点的概率。
        l_c:
            城市内部的联系概率。
    """
    if l_p is None:
        l_p = self.p["l_p"]
    if l_c is None:
        l_c = self.p["l_c"]
    hub_agents = ActorsList(self.model)
    links = []
    for city in self.managed:
        # 每个城市里的农民自己互相关联
        links.extend(city.link_farmers(l_c))
        # 每个群体里抽一个“中心节点”，可以和其它的外市主体有交集
        if hub := city.managed.random.choice(when_empty="return None"):
            hub_agents.append(hub)
    # 这里 Hub 节点怎么选择伙伴？目前用的是全联接
    links.extend(hub_agents.random.link("friend", p=l_p))
    # 为了方便测试，记录一下链接的数量
    logger.info(f"{self.name_en} has {len(links)} links.")
    return links

市级主体¶

经过筛选，本研究使用黄河流域地级市共计 59 个。

更新农业灌溉主体¶

API References

根据自身的耕地情况，随机产生农民主体，直到主体总数达到某值。

Parameters:

Name	Type	Description	Default
`num`	`int`	期望达到的主体数量。	required
`farmer_cls`	`Type[Farmer]`	生成主体的基类，默认是本模型自带的农民。	`Farmer`

Returns:

Type	Description
`ActorsList[Farmer]`	完成添加或死亡后，当前城市管理的农民主体列表。

Source code in src/api/city.py

def add_until(
    self, num: int, farmer_cls: Type[Farmer] = Farmer
) -> ActorsList[Farmer]:
    """根据自身的耕地情况，随机产生农民主体，直到主体总数达到某值。

    Parameters:
        num:
            期望达到的主体数量。
        farmer_cls:
            生成主体的基类，默认是本模型自带的农民。

    Returns:
        完成添加或死亡后，当前城市管理的农民主体列表。
    """
    # 与当前的差值
    diff = int(num - self.n_agents)
    farmers = self.managed
    if diff == 0:
        return farmers
    # 获取所有土地单元
    cells = self.get_cells()
    if diff < 0:  # 如果当前农民比期望的更多，杀死它们
        farmers.random.choice(abs(diff), as_list=True).trigger("die")
    elif diff > 0:  # 如果需要创建更多主体
        cells.random.new(
            farmer_cls,
            size=diff,
            prob="irr_area",
            double_check=True,
            replace=False,
            actor_attrs={"city": self},
        )
    return self.managed

更新主体灌溉面积¶

API References

为农民分配种植的作物。

Parameters:

Name	Type	Description	Default
`crops`	`List[str]`	作物类型列表，例如 ["Rice", "Wheat", "Maize"]。作物名称应该是可以识别的，例如在 `irr_area` 中有对应的列。	required
`nums`	`ndarray`	土地利用格局，种植不同作物的之和农民主体数量应该相等。如果数量不等就随机增加到那个数量（调用 `add_farmers_until()` 方法）。	required

Raises:

Type	Description
`ValueError`	如果土地利用格局的总和不等于农民主体数量。或者作物类型数不等于 nums 列表的长度。

Source code in src/api/city.py

def shuffle_farmers(self, crops: List[str], nums: np.ndarray) -> None:
    """为农民分配种植的作物。

    Parameters:
        crops:
            作物类型列表，例如 ["Rice", "Wheat", "Maize"]。
            作物名称应该是可以识别的，例如在 `irr_area` 中有对应的列。
        nums:
            土地利用格局，种植不同作物的之和农民主体数量应该相等。
            如果数量不等就随机增加到那个数量（调用 `add_farmers_until()` 方法）。

    Raises:
        ValueError:
            如果土地利用格局的总和不等于农民主体数量。
            或者作物类型数不等于 nums 列表的长度。
    """
    # 根据土地利用格局，将农民按照不同的作物进行分组
    if sum(nums) != self.n_agents:
        raise ValueError(
            f"Total lands {nums} not equal to the number of farmers {self.n_agents}."
        )
    if len(nums) != len(crops):
        raise ValueError(f"Length of crops {crops} not equal to lands {nums}.")
    farmers_g = self.managed.split(np.cumsum(nums)[:-1])
    # 对每一组农民，分配作物
    for crop, farmers in zip(crops, farmers_g):
        farmers.set("crop", crop)
        area = self.irr_area.get(crop, 0.0)
        farmers.random.assign(area, "irr_area", "return None")

水资源管理主体

省级主体¶

managed property ¶

quota property ¶

water_prices cached property ¶

create classmethod ¶

setup ¶

update_data ¶

update_graph ¶

市级主体¶

更新农业灌溉主体¶

更新主体灌溉面积¶

managed `property` ¶

quota `property` ¶

water_prices `cached` `property` ¶

create `classmethod` ¶