UniswapV3

def model_post_init(self, __context):
    network = self.__class__._network
    object.__setattr__(self, "_network", network)
    # overwrite the .rpc() classmethod
    object.__setattr__(self, "rpc", self._rpc)

@classmethod
def rpc(cls) -> "RPC":
    """
    This uses the default network, unless a network has been provided
    """
    from ._transport import _force_get_global_rpc

    if cls._network is None:
        return _force_get_global_rpc()
    response = _force_get_global_rpc(cls._network)
    return response

def add_func(self, func: "FuncSignature"):
    if func not in self.functions:
        self.functions.append(ContractFunc(func=func, contract=self))

def get_storage_at(
    self, *, slot: int | HexStr, block_number="latest", sync: bool = False
) -> MaybeAwaitable[HexStr]:
    return run(
        self._get_storage_at,
        slot=slot,
        block_number=block_number,
        sync=sync,
    )

def get_code(
    self,
    *,
    block_number: int | BLOCK_STRINGS | None = None,
    block_hash: HexStr | None = None,
    sync: bool = False,
) -> MaybeAwaitable[HexStr]:
    return run(
        self._get_code,
        block_number=block_number,
        block_hash=block_hash,
        sync=sync,
    )

def create2(self, salt: bytes, keccak_init_code: bytes) -> HexAddress:
    """
    EIP-104
    https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1014.md
    """
    pre = "0xff"
    b_pre = bytes.fromhex(pre[2:])
    b_address = bytes.fromhex(self.address[2:])

    b_result = keccak_256(b_pre + b_address + salt + keccak_init_code)
    result_address = "0x" + b_result[12:].hex()

    return HexAddress(HexStr(result_address))

def model_post_init(self, __context):
    network = self.__class__._network
    object.__setattr__(self, "_network", network)
    # overwrite the .rpc() classmethod
    object.__setattr__(self, "rpc", self._rpc)

@classmethod
def rpc(cls) -> "RPC":
    """
    This uses the default network, unless a network has been provided
    """
    from ._transport import _force_get_global_rpc

    if cls._network is None:
        return _force_get_global_rpc()
    response = _force_get_global_rpc(cls._network)
    return response

def add_func(self, func: "FuncSignature"):
    if func not in self.functions:
        self.functions.append(ContractFunc(func=func, contract=self))

def get_storage_at(
    self, *, slot: int | HexStr, block_number="latest", sync: bool = False
) -> MaybeAwaitable[HexStr]:
    return run(
        self._get_storage_at,
        slot=slot,
        block_number=block_number,
        sync=sync,
    )

def get_code(
    self,
    *,
    block_number: int | BLOCK_STRINGS | None = None,
    block_hash: HexStr | None = None,
    sync: bool = False,
) -> MaybeAwaitable[HexStr]:
    return run(
        self._get_code,
        block_number=block_number,
        block_hash=block_hash,
        sync=sync,
    )

def create2(self, salt: bytes, keccak_init_code: bytes) -> HexAddress:
    """
    EIP-104
    https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1014.md
    """
    pre = "0xff"
    b_pre = bytes.fromhex(pre[2:])
    b_address = bytes.fromhex(self.address[2:])

    b_result = keccak_256(b_pre + b_address + salt + keccak_init_code)
    result_address = "0x" + b_result[12:].hex()

    return HexAddress(HexStr(result_address))

async def get_all_indices(self, owner: HexAddress) -> list[int]:
    balance = await self.balance_of(OwnerRequest(owner=owner)).get()
    calls = []
    for i in range(balance):
        calls.append(
            self.token_of_owner_by_index(
                OwnerTokenRequest(
                    owner=owner,
                    index=primitives.uint256(i),
                )
            )
        )
    return await multicall.execute(*calls)

async def get_all_positions(self, owner: HexAddress) -> list[Position]:
    indices = await self.get_all_indices(owner)
    return await multicall.execute(
        *[self.positions(primitives.uint256(index)) for index in indices]
    )

def model_post_init(self, __context):
    network = self.__class__._network
    object.__setattr__(self, "_network", network)
    # overwrite the .rpc() classmethod
    object.__setattr__(self, "rpc", self._rpc)

@classmethod
def rpc(cls) -> "RPC":
    """
    This uses the default network, unless a network has been provided
    """
    from ._transport import _force_get_global_rpc

    if cls._network is None:
        return _force_get_global_rpc()
    response = _force_get_global_rpc(cls._network)
    return response

def add_func(self, func: "FuncSignature"):
    if func not in self.functions:
        self.functions.append(ContractFunc(func=func, contract=self))

def get_storage_at(
    self, *, slot: int | HexStr, block_number="latest", sync: bool = False
) -> MaybeAwaitable[HexStr]:
    return run(
        self._get_storage_at,
        slot=slot,
        block_number=block_number,
        sync=sync,
    )

def get_code(
    self,
    *,
    block_number: int | BLOCK_STRINGS | None = None,
    block_hash: HexStr | None = None,
    sync: bool = False,
) -> MaybeAwaitable[HexStr]:
    return run(
        self._get_code,
        block_number=block_number,
        block_hash=block_hash,
        sync=sync,
    )

def create2(self, salt: bytes, keccak_init_code: bytes) -> HexAddress:
    """
    EIP-104
    https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1014.md
    """
    pre = "0xff"
    b_pre = bytes.fromhex(pre[2:])
    b_address = bytes.fromhex(self.address[2:])

    b_result = keccak_256(b_pre + b_address + salt + keccak_init_code)
    result_address = "0x" + b_result[12:].hex()

    return HexAddress(HexStr(result_address))

async def get_price(self, token0: bool = True):
    slot0 = await self.slot0().get()
    _token0 = await self.token0().get()
    _token1 = await self.token1().get()
    token0_decimals = await ERC20(address=_token0).decimals().get()
    token1_decimals = await ERC20(address=_token1).decimals().get()
    return self.sqrt_price_x96_to_token_prices(
        slot0.sqrt_price,
        token0_decimals,
        token1_decimals,
        token0,
    )

@staticmethod
def sqrt_price_x96_to_token_prices(
    sqrt_price_x96: int,
    token0_decimals: int,
    token1_decimals: int,
    token0: bool = True,
) -> Decimal:
    num = Decimal(sqrt_price_x96 * sqrt_price_x96)
    price1 = (
        (num / Q192)
        * (Decimal(10) ** token0_decimals)
        / (Decimal(10) ** token1_decimals)
    )
    price0 = Decimal(1) / price1

    return price0 if token0 else price1

async def all_tick_values(self):
    tick_spacing = await self.get_tick_spacing()
    min_tick, max_tick = await self.tick_range()
    return list(range(min_tick, max_tick, tick_spacing))

async def get_token0(self):
    if not self._token0:
        self._token0 = ERC20(await self.token0().get())
    return self._token0

async def get_token1(self):
    if not self._token1:
        self._token1 = ERC20(await self.token1().get())
    return self._token1

async def get_tick_spacing(self):
    if not self._tick_spacing:
        self._tick_spacing = await self.tick_spacing().get()
    return self._tick_spacing

async def get_swaps(
    self, start_block, end_block
) -> dict[int, list[V3SwapEventType]]:
    ticks: dict[int, list[V3SwapEventType]] = {}
    tick_values = await self.all_tick_values()
    async for event in V3SwapEvent.set_filter(addresses=[self.address]).backfill(
        start_block, end_block
    ):
        tick = event.event.tick
        closest_tick = max(t for t in tick_values if t <= tick)
        if tick in ticks:
            ticks[closest_tick].append(event.event)
        else:
            ticks[closest_tick] = [event.event]
    return ticks

async def get_ticks(self, indices: Iterable[int]) -> list[ProcessedTick]:
    calls = []
    for tick in indices:
        calls.append(self.ticks(primitives.int24(tick)))
    ticks: list[Tick] = await multicall.execute(*calls)

    processed_ticks = []
    for idx, tick_data in zip(indices, ticks):
        processed_ticks.append(
            ProcessedTick(
                index=idx,
                tick=tick_data,
                token0=await self.get_token0(),
                token1=await self.get_token1(),
            )
        )
    return processed_ticks

async def tvl(self, block_number=None) -> list[ProcessedTick]:
    liquidity, initialized_ticks, active_tick_index = await asyncio.gather(
        self.liquidity().get(),
        self.get_initialized_ticks(),
        self.active_tick(),
    )
    ticks = await self.get_ticks(initialized_ticks)

    if active_tick_index in initialized_ticks:
        active_tick = [t for t in ticks if t.index == active_tick_index][0]
        active_tick.active_liquidity = liquidity
        current_liquidity = liquidity - active_tick.liquidity_net
        previous_tick_net = active_tick.liquidity_net
    else:
        current_liquidity = liquidity
        previous_tick_net = 0

    for tick in [tick for tick in ticks if tick.index < active_tick_index][::-1]:
        tick.active_liquidity = current_liquidity - previous_tick_net
        current_liquidity = tick.active_liquidity
        previous_tick_net = tick.liquidity_net

    current_liquidity = liquidity
    previous_tick_net = 0
    for tick in [tick for tick in ticks if tick.index > active_tick_index]:
        tick.active_liquidity = current_liquidity + tick.liquidity_net
        current_liquidity = tick.active_liquidity
    return ticks

async def get_tick_range(self, lower: int, upper: int) -> list[ProcessedTick]:
    if not self._tick_spacing:
        self._tick_spacing = await self.tick_spacing().get()
    assert self._tick_spacing
    return await self.get_ticks(range(lower, upper, self._tick_spacing))

async def tick_range(self):
    tick_spacing = await self.get_tick_spacing()
    min_tick = math.ceil(MIN_TICK / tick_spacing) * tick_spacing
    return min_tick, -min_tick

async def bitmap_range(self) -> tuple[int, int]:
    min_tick, max_tick = await self.tick_range()
    return (min_tick // 60 // 256), (max_tick // 60 // 256)

async def get_initialized_ticks(self) -> list[int]:
    min_bitmap, max_bitmap = await self.bitmap_range()
    tick_spacing = await self.get_tick_spacing()
    bitmaps = await multicall.execute(
        *[
            self.tick_bitmap(primitives.int16(idx))
            for idx in range(min_bitmap, max_bitmap + 1)
        ]
    )
    initialized_ticks = []
    for idx, row in zip(range(min_bitmap, max_bitmap + 1), bitmaps):
        initialized_ticks.extend(tick_from_bitmap(idx, row, tick_spacing))
    return initialized_ticks

async def active_tick(self) -> int:
    slot0 = await self.slot0().get()
    if not self._tick_spacing:
        self._tick_spacing = await self.tick_spacing().get()
    return slot0.tick // self._tick_spacing * self._tick_spacing

async def liquidity_at_current_tick(self, tick: int) -> tuple[float, float]:
    tick_spacing, slot0, liquidity = cast(
        tuple[int, Slot0, int],
        await multicall.execute(
            self.tick_spacing(),
            self.slot0(),
            self.liquidity(),
        ),
    )
    bottom_tick = tick
    top_tick = tick + tick_spacing

    price = tick_to_price(slot0.tick)
    sa = tick_to_price(bottom_tick // 2)
    sb = tick_to_price(top_tick // 2)
    sqrt_price = math.sqrt(price)

    amount0 = liquidity * (sb - sqrt_price) / (sqrt_price * sb)
    amount1 = liquidity * (sqrt_price - sa)

    return (amount0, amount1)

async def get_all_positions(self):
    events = []
    async for event in V3MintEvent.set_filter(addresses=[self.address]).backfill():
        events.append(event.event)
    return events