Captando água de um canal: um modelo baseado em agentes usando Python
A modelagem baseada em agentes (ABM) é uma técnica poderosa que pode ser usada em muitas áreas, incluindo a hidrologia. A ABM permite simular o comportamento complexo de sistemas naturais e sociais, permitindo que os cientistas compreendam melhor as interações entre os diferentes elementos de um sistema. Na hidrologia, a ABM tem sido usada para simular o comportamento dos rios, bacias hidrográficas e outros sistemas hídricos.
Neste artigo, exploraremos a importância da ABM na hidrologia e apresentaremos um modelo baseado em agentes para captar água de um canal. Utilizaremos a linguagem Python para programação, o que nos dará liberdade de implementar funções personalizadas e flexíveis, sem a necessidade de ficarmos dependentes de algum software.
Como demonstração, utilizaremos um exemplo em que usuários retiram água de um canal linear. Entre as funcionalidades que veremos aqui, temos: a criação de agentes, implementação de regras de interação, modelo baseado em grafo (usuários a montante afetam a quantidade de água disponível dos usuários a jusante) e implementação de regras de ativação de agentes personalizada. Acesse este link para ver o Jupyter Notebook deste tutorial. Esse artigo também foi publicado em Inglês.
Setup
Usaremos o pacote Mesa para construir nosso modelo. Mesa é uma biblioteca em Python para modelagem baseada em agentes. Ela possui classes base para construir as principais funcionalidades de um ABM.
Se você não tiver o Mesa instalado, siga a documentação. Também recomendo o uso do Anaconda para gerenciamento de pacotes.
Resumidamente, crie um novo ambiente virtual no anaconda e ative-o.
#Anaconda
conda activate mesaenv
#Terminal
source bin/activateEm seguida, instale os quatro pacotes python que usaremos aqui: mesa, matplotlib, networkx e jupyter (para executar este notebook).
python3 -m pip install mesa
python3 -m pip install matplotlib
python3 -m pip install networkx
python3 -m pip install jupyterClasses Agent e Model
Começaremos criando um modelo simples e vamos implementar funções aos poucos. Criamos duas classes, uma que representa os usuários que retiram água do canal, e outra que lida com variáveis a nível global do modelo. Eles são representados pelas classes WaterUserAgent e WaterModel.
from mesa import Agent, Model
class WaterUserAgent(Agent):
"""An agent that interacts with the water canal by withdrawing water for he/she own use."""
_last_id = 0
def __init__(self, model):
super().__init__(WaterUserAgent._last_id+1, model)
WaterUserAgent._last_id += 1
self.water_to_withdraw = 1
class WaterModel(Model):
"""
Create a new Irrigation model with the given parameters.
Args:
N: int number that represent the number of water users to create per year
"""
def __init__(self, N=10):
self.num_agents = N
# Create agents
for i in range(self.num_agents):
w = WaterUserAgent(self)Importamos Agent e Model, classes que herdaremos no código. No método init de WaterUserAgent definimos que cada usuário possui um atributo chamado water_to_withdraw que representa a quantidade de água que ele retirará do canal. Já na classe WaterModel, recebemos o número de agentes a serem criados e criamos os agentes (mas ainda não adicionamos eles no modelo).
Definindo o espaço
Como todo canal hídrico, as interações entre os agentes a montante e o canal afetam os usuários a jusante. Por exemplo, no caso extremo em que o primeiro agente retira toda a água disponível do canal, todos os demais agentes não terão água para retirar. Usaremos o NetworkGrid que é uma forma baseada em grafos para definir o espaço. Desta forma, criamos uma conexão lógica em sequência. Vamos precisar de ajuda da NetworkX, uma biblioteca em Python que cria grafos (uma espécie de rede). Criamos um grafo linear. Sendo assim, cada nó do grafo é conectado com o nó antecessor e subsequente.
A partir de um grafo linear criamos a lógica de um canal. Mudando o grafo, podemos modelar bacias hidrográficas inteiras!
Vamos construir nosso grafo linear:
import networkx as nx
import matplotlib.pyplot as plt
G = nx.path_graph(100)
nx.draw_spectral(G, with_labels=True, font_size=8)
Devido ao número grande de nós, a visualização não ficou bonita, mas ele funciona perfeitamente. Nós criamos um grafo com 100 nós. Cada nó é uma possível localização que um agente pode se alocar. Apenas uma gente pode se alocar em um nó por vez. Sendo assim, o número de nós restringe o número máximo de usuários no modelo. Como para fins de exemplo estamos modelando um pequeno canal com poucos usuários de água, 100 está bom.
Assim que nosso grafo linear estiver pronto, vamos atualizar a classe WaterModel! Criaremos também uma função que aloca usuários de água em um nó aleatório.
from mesa.space import NetworkGrid
import random
class WaterModel(Model):
"""
Create a new Irrigation model with the given parameters.
Args:
N: int number that represent the number of water users to create per year
G: generated linear graph using networkx
"""
def __init__(self, N=2):
self.num_agents = N
self.grid = NetworkGrid(G)
self.number_of_farmers_to_create = 10
# Create agents
for i in range(self.num_agents):
w = WaterUserAgent(self)
def create_farmers_random_position(self):
"""Create water users at random position in a linear graph."""
while (i < self.number_of_farmers_to_create):
random_node = random.sample(list(G.nodes()), 1)
if (len(self.grid.get_cell_list_contents(random_node)) == 0):
f = WaterUserAgent(self)
self.schedule.add(f)
self.grid.place_agent(f, random_node[0])
i += 1
def step(self):
""" Execute the step for all agents, one at a time. At the end, advance model by one step """
# Preparation
self.create_farmers_random_position()
# Run step
self.schedule.step()Acima, nós estamos criando agentes e os posicionando em um nó aleatório. Estamos chamando o método que faz isso no método step, ou seja, a cada iteração do modelo novos agentes serão criados.
Reordenando a ativação dos agentes (Schedule Personalizado)
Após criamos os agentes, precisamos criar uma sequência de passos que será executada para cada agente quando ele for ativado. Para isto, criaremos outro método step, desta vez para a classe dos Agentes. Nós precisaremos chamar o step no Schedule, para executar as funções deste método para todos os agentes, um por vez. O Schedule faz parte do módulo de tempo do mesa. Ele controla quando cada agente executa suas ações. Como a ordem de ação dos agentes impacta significativamente os resultados, ele precisa ser explicitado.
A ação que queremos que os agentes executem a cada iteração é retirar água do canal. Por padrão, o scheduler ativa os agentes aleatoriamente. Mas num canal hídrico, não é assim que deve funcionar. Usuários a montante afetarão os a jusante quanto à disponibilidade de água. Portanto, devemos redefinir a ordem em que os agentes são ativados: agentes mais a montante são ativados primeiro. Para redefinir a ordem em que os agentes com base em sua posição no canal, devemos criar um Scheduler personalizado.
Vamos colocar a criação da função step em espera e trabalhar no Scheduler.
Primeiro, criaremos um scheduler personalizado que herda a classe BaseSchedule, que é o scheduler mais simples que ativa os agentes um por vez e assume que todos eles têm um método step. Em seguida, criaremos um método que obtenha a ordem correta para ativar os agentes, classificando todos os agentes com base em sua posição. Por fim, criamos um método que reordena a ativação dos agentes.
Mesa usa um Dicionário Ordenado (Ordered Dictionary), que é uma subclasse de dicionário em Python que “lembra” a ordem em que os itens foram adicionados através de uma chave para cada dicionário.
Caso você tenha vários tipos de agentes, você também pode redefinir a ordem em que cada classe de agente é ativada. Veja esta issue para mais detalhes.
Abaixo criamos o nosso scheduler personalizado.
from mesa.time import BaseScheduler
from collections import OrderedDict
class CanalScheduler(BaseScheduler):
def __init__(self, model):
super().__init__(model)
def sort_agents(self):
sorted_list_of_agents = sorted(self.agents, key=lambda x: x.pos)
new_queue = OrderedDict(dict((i,j) for i,j in enumerate(sorted_list_of_agents)))
return new_queue
def step(self):
self._agents = self.sort_agents()
super().step()O scheduler precisa ter um método step. Este método que é chamado no quando rodamos aself.schedule.step() no WaterModel. Ela chama sort_agentsque reordena os agentes com base na sua posição.
Agora que temos nosso scheduler personalizado, finalmente criaremos o método run_model que chama o método step do nosso scheduler. Para testar, vamos criar o método step para que cada agente imprima uma mensagem quando for acionado para verificar se nossa implementação está correta.
class WaterUserAgent(Agent):
"""An agent that interacts with the water canal by withdrawing water for he/she own use."""
_last_id = 0
def __init__(self, model):
super().__init__(WaterUserAgent._last_id+1, model)
WaterUserAgent._last_id += 1
self.water_to_withdraw = 1
def step(self):
print("Olá, Eu sou o Water Agent nº {} e estou na posição {}.".format(
self.unique_id, self.pos))
class WaterModel(Model):
"""
Create a new Irrigation model with the given parameters.
Args:
N: int number that represent the number of water users to create per year
G: generated linear graph using networkx
"""
def __init__(self, N=2):
self.num_agents = N
self.grid = NetworkGrid(G)
self.number_of_farmers_to_create = 2
# Define scheduler
self.schedule = CanalScheduler(
WaterModel)
# Create agents
for i in range(self.num_agents):
w = WaterUserAgent(self)
def create_farmers_random_position(self):
"""Create water users at random position in a linear graph."""
i = 0
while (i < self.number_of_farmers_to_create):
random_node = random.sample(list(G.nodes()), 1)
if (len(self.grid.get_cell_list_contents(random_node)) == 0):
f = WaterUserAgent(self)
self.schedule.add(f)
self.grid.place_agent(f, random_node[0])
i += 1
def step(self):
""" Execute the step for all agents, one at a time. At the end, advance model by one step """
# Preparation
self.create_farmers_random_position()
# Run step
self.schedule.step()
def run_model(self, step_count=3):
for i in range(1,step_count+1):
print("-------------- \n" +
"Initiating year n. " + str(i) + "\n" +
"--------------")
self.step()model.run_model()
--------------
Initiating year n. 1
--------------
Hi, I am Water Agent n. 5 and I am in position 26. I am withdrawing 4.245385761173964 m³/year
Available Water: 95.75461423882604
Hi, I am Water Agent n. 4 and I am in position 50. I am withdrawing 7.98141842703323 m³/year
Available Water: 87.7731958117928
Hi, I am Water Agent n. 6 and I am in position 68. I am withdrawing 2.270441679844418 m³/year
Available Water: 85.50275413194838
--------------
Initiating year n. 2
--------------
Hi, I am Water Agent n. 7 and I am in position 12. I am withdrawing 4.615901240437739 m³/year
Available Water: 95.38409875956226
Hi, I am Water Agent n. 5 and I am in position 26. I am withdrawing 4.245385761173964 m³/year
Available Water: 91.1387129983883
Hi, I am Water Agent n. 4 and I am in position 50. I am withdrawing 7.98141842703323 m³/year
Available Water: 83.15729457135507
Hi, I am Water Agent n. 8 and I am in position 51. I am withdrawing 5.55253531624829 m³/year
Available Water: 77.60475925510677
Hi, I am Water Agent n. 9 and I am in position 64. I am withdrawing 6.911028324209107 m³/year
Available Water: 70.69373093089766
Hi, I am Water Agent n. 6 and I am in position 68. I am withdrawing 2.270441679844418 m³/year
Available Water: 68.42328925105323
...Como a ativação dos agentes segue uma sequência lógica quanto às suas posições, parece que o modelo está indo bem.
Nosso modelo está pronto! Antes de prosseguirmos para a visualização, faremos algumas observações:
- A escolha de sortear
water_to_withdrawde uma distribuição uniforme é para fins de exemplo. Em um caso real, sugiro você deve tentar ajustar com a lei de potência, distribuições lognormal ou gama, pois irão considerar adequadamente a assimetria nos dados. - Definimos a duração do step como 1 ano, o que corresponde a duração de uma colheita para a maioria das culturas para os usuários irrigantes, isso também simplifica o balanço hídrico. Como algumas culturas precisam de mais água do que outras em épocas diferentes, um step mais discretizado seria ideal
- Na maioria dos países, uma agência governamental gerencia e supervisiona as retiradas de água. Aqui, sua imaginação para adicionar recursos a este ABM pode voar. Você pode definir diferentes usuários de água (irrigantes, para abastecimento humano …) e introduzir novos elementos como reservatórios e o órgão regulador.
Coletar dados e visualizá-los
Para obter dados de nosso ABM, usaremos a função interna do mesa DataCollector. Incluindo a coleta de dados, a versão final do nosso ABM ficará assim:
from mesa.datacollection import DataCollector
class WaterUserAgent(Agent):
"""An agent that interacts with the water canal by withdrawing water for he/she own use."""
_last_id = 0
def __init__(self, model):
super().__init__(WaterUserAgent._last_id+1, model)
WaterUserAgent._last_id += 1
self.water_to_withdraw = random.uniform(1, 9)
def withdraw_water(self):
model.water_available -= self.water_to_withdraw
def step(self):
print("Hi, I am Water Agent n. {} and I am in position {}. I am withdrawing {} m³/year".format(
self.unique_id, self.pos, self.water_to_withdraw))
self.withdraw_water()
print("Available Water: {}".format(model.water_available))
class WaterModel(Model):
"""
Create a new Irrigation model with the given parameters.
Args:
N: int number that represent the number of water users to create per year
G: generated linear graph using networkx
water_available: Total water to supply the canal (in m³/year)
"""
def __init__(self,
N,
G,
water_available):
self.num_agents = N
self.grid = NetworkGrid(G)
self.number_of_farmers_to_create = N
self.water_available_to_reset = water_available
# Define scheduler
self.schedule = CanalScheduler(WaterModel)
# Create agents
for i in range(self.num_agents):
w = WaterUserAgent(self)
# Data Collector
self.datacollector = DataCollector(
agent_reporters={
"Position":
lambda x: x.pos,
"Withdrawn water (m³/year)":
lambda x: x.water_to_withdraw,
},
)
self.datacollector.collect(self)
def create_farmers_random_position(self):
"""Create water users at random position in a linear graph."""
i = 0
while (i < self.number_of_farmers_to_create):
random_node = random.sample(list(G.nodes()), 1)
if (len(self.grid.get_cell_list_contents(random_node)) == 0):
f = WaterUserAgent(self)
self.schedule.add(f)
self.grid.place_agent(f, random_node[0])
i += 1
def step(self):
""" Execute the step for all agents, one at a time. At the end, advance model by one step """
# Preparation
self.create_farmers_random_position()
self.water_available = self.water_available_to_reset
# Run step
self.schedule.step()
# Save data
self.datacollector.collect(self)
def run_model(self, step_count=3):
for i in range(1,step_count+1):
print("-------------- \n" +
"Initiating year n. " + str(i) + "\n" +
"--------------")
self.step()water_available = 100
model = WaterModel(random.randint(1, 5), G, water_available)
model.run_model(10)
agents_results = model.datacollector.get_agent_vars_dataframe()import matplotlib.pyplot as plt
def water_balance_in_canal(agent_results):
index = []
water_volume = []
for i in range(agent_results.index.get_level_values(0).min(), agent_results.index.get_level_values(0).max()+1):
results_current_step = agent_results.xs(i, level=0)
index.append(i)
water_volume.append(results_current_step['Withdrawn water (m³/year)'].sum())
water_balance = [water_available - x for x in water_volume]
fig, ax = plt.subplots(dpi=100)
ax.axhline(0, linestyle='--', color='red')
ax.plot(index, water_balance, color='blue')
ax.set_xlabel('Step')
ax.set_ylabel('Balanço hídrico anual')
water_balance_in_canal(agents_results)
Que modificações você faria e quais funcionalidades adicionaria para esse caso ficar ainda mais real?
