
import inspect
import unittest

import torch
import torch.nn as nn

from accelerate.hooks import (
    AlignDevicesHook,
    ModelHook,
    SequentialHook,
    add_hook_to_module,
    attach_align_device_hook,
    remove_hook_from_module,
    remove_hook_from_submodules,
)
from accelerate.test_utils import require_multi_gpu




class 					_SCREAMING_SNAKE_CASE  (      nn.Module ):
						def __init__(			self ):
										super().__init__()
										snake_case__						=				nn.Linear(3		,	4 )
										snake_case__						=				nn.BatchNormad(4 )
										snake_case__						=				nn.Linear(4		,	5 )



						def A_			(			self		,	lowerCamelCase ):
										return self.lineara(self.batchnorm(self.lineara(__a ) ) )





class 					_SCREAMING_SNAKE_CASE  (      UpperCamelCase_ ):
						def A_			(			self		,	lowerCamelCase		,	*lowerCamelCase		,	**lowerCamelCase ):
										return (args[0] + 1,) + args[1:], kwargs





class 					_SCREAMING_SNAKE_CASE  (      UpperCamelCase_ ):
						def A_			(			self		,	lowerCamelCase		,	lowerCamelCase ):
										return output + 1





class 					_SCREAMING_SNAKE_CASE  (      unittest.TestCase ):
						def A_			(			self ):
										snake_case__						=				ModelForTest()
										snake_case__						=				ModelHook()

										add_hook_to_module(__a		,	__a )
										self.assertEqual(test_model._hf_hook		,	__a )
										self.assertTrue(hasattr(__a		,	"_old_forward" ) )

										# Check adding the hook did not change the name or the signature
										self.assertEqual(test_model.forward.__name__		,	"forward" )
										self.assertListEqual(list(inspect.signature(test_model.forward ).parameters )		,	["x"] )

										remove_hook_from_module(__a )
										self.assertFalse(hasattr(__a		,	"_hf_hook" ) )
										self.assertFalse(hasattr(__a		,	"_old_forward" ) )
						def A_			(			self ):
										snake_case__						=				ModelForTest()
										snake_case__						=				ModelHook()

										add_hook_to_module(__a		,	__a )
										add_hook_to_module(__a		,	__a		,	append=__a )

										self.assertEqual(isinstance(test_model._hf_hook		,	__a )		,	__a )
										self.assertEqual(len(test_model._hf_hook.hooks )		,	2 )
										self.assertTrue(hasattr(__a		,	"_old_forward" ) )

										# Check adding the hook did not change the name or the signature
										self.assertEqual(test_model.forward.__name__		,	"forward" )
										self.assertListEqual(list(inspect.signature(test_model.forward ).parameters )		,	["x"] )

										remove_hook_from_module(__a )
										self.assertFalse(hasattr(__a		,	"_hf_hook" ) )
										self.assertFalse(hasattr(__a		,	"_old_forward" ) )
						def A_			(			self ):
										snake_case__						=				ModelForTest()
										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				test_model(x + 1 )
										snake_case__						=				test_model(x + 2 )

										snake_case__						=				PreForwardHook()
										add_hook_to_module(__a		,	__a )
										snake_case__						=				test_model(__a )
										self.assertTrue(torch.allclose(__a		,	__a		,	atol=1e-5 ) )

										# Attaching a hook to a model when it already has one replaces, does not chain
										snake_case__						=				PreForwardHook()
										add_hook_to_module(__a		,	__a )
										snake_case__						=				test_model(__a )
										self.assertTrue(torch.allclose(__a		,	__a		,	atol=1e-5 ) )

										# You need to use the sequential hook to chain two or more hooks
										snake_case__						=				SequentialHook(PreForwardHook()		,	PreForwardHook() )
										add_hook_to_module(__a		,	__a )

										snake_case__						=				test_model(__a )
										assert torch.allclose(__a		,	__a		,	atol=1e-5 )
						def A_			(			self ):
										snake_case__						=				ModelForTest()
										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				test_model(__a )

										snake_case__						=				PostForwardHook()
										add_hook_to_module(__a		,	__a )
										snake_case__						=				test_model(__a )
										self.assertTrue(torch.allclose(__a		,	output + 1		,	atol=1e-5 ) )

										# Attaching a hook to a model when it already has one replaces, does not chain
										snake_case__						=				PostForwardHook()
										add_hook_to_module(__a		,	__a )
										snake_case__						=				test_model(__a )
										self.assertTrue(torch.allclose(__a		,	output + 1		,	atol=1e-5 ) )

										# You need to use the sequential hook to chain two or more hooks
										snake_case__						=				SequentialHook(PostForwardHook()		,	PostForwardHook() )
										add_hook_to_module(__a		,	__a )

										snake_case__						=				test_model(__a )
										assert torch.allclose(__a		,	output + 2		,	atol=1e-5 )
						def A_			(			self ):
										snake_case__						=				ModelForTest()
										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				test_model(__a )

										snake_case__						=				PostForwardHook()
										add_hook_to_module(__a		,	__a )
										snake_case__						=				test_model(__a )
										self.assertTrue(torch.allclose(__a		,	output + 1 ) )
										self.assertTrue(outputa.requires_grad )

										snake_case__						=				True
										snake_case__						=				test_model(__a )
										self.assertFalse(outputa.requires_grad )
						@require_multi_gpu
						def A_			(			self ):
										snake_case__						=				ModelForTest()
										# Everything is on CPU
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# This will move each submodule on different devices
										add_hook_to_module(model.lineara		,	AlignDevicesHook(execution_device=0 ) )
										add_hook_to_module(model.batchnorm		,	AlignDevicesHook(execution_device=0 ) )
										add_hook_to_module(model.lineara		,	AlignDevicesHook(execution_device=1 ) )

										self.assertEqual(model.lineara.weight.device		,	torch.device(0 ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device(0 ) )
										self.assertEqual(model.batchnorm.running_mean.device		,	torch.device(0 ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device(1 ) )

										# We can still make a forward pass. The input does not need to be on any particular device
										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	torch.device(1 ) )

										# We can add a general hook to put back output on same device as input.
										add_hook_to_module(__a		,	AlignDevicesHook(io_same_device=__a ) )
										snake_case__						=				torch.randn(2		,	3 ).to(0 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	torch.device(0 ) )
						def A_			(			self ):
										snake_case__						=				ModelForTest()

										# Everything is on CPU
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# This will move each submodule on different devices
										snake_case__						=				{'execution_device': 0 if torch.cuda.is_available() else 'cpu', 'offload': True}

										add_hook_to_module(model.lineara		,	AlignDevicesHook(**__a ) )
										add_hook_to_module(model.batchnorm		,	AlignDevicesHook(**__a ) )
										add_hook_to_module(model.lineara		,	AlignDevicesHook(**__a ) )

										# Parameters have been offloaded, so on the meta device
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										# Buffers are not included in the offload by default, so are on the execution device
										snake_case__						=				torch.device(hook_kwargs["execution_device"] )
										self.assertEqual(model.batchnorm.running_mean.device		,	__a )

										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	__a )

										# Removing hooks loads back the weights in the model.
										remove_hook_from_module(model.lineara )
										remove_hook_from_module(model.batchnorm )
										remove_hook_from_module(model.lineara )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# Now test with buffers included in the offload
										snake_case__						=				{
										    'execution_device': 0 if torch.cuda.is_available() else 'cpu',
										    'offload': True,
										    'offload_buffers': True,
										}

										add_hook_to_module(model.lineara		,	AlignDevicesHook(**__a ) )
										add_hook_to_module(model.batchnorm		,	AlignDevicesHook(**__a ) )
										add_hook_to_module(model.lineara		,	AlignDevicesHook(**__a ) )

										# Parameters have been offloaded, so on the meta device, buffers included
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.running_mean.device		,	torch.device("meta" ) )

										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	__a )

										# Removing hooks loads back the weights in the model.
										remove_hook_from_module(model.lineara )
										remove_hook_from_module(model.batchnorm )
										remove_hook_from_module(model.lineara )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
						def A_			(			self ):
										snake_case__						=				ModelForTest()

										# Everything is on CPU
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# This will move each submodule on different devices
										snake_case__						=				0 if torch.cuda.is_available() else 'cpu'
										attach_align_device_hook(__a		,	execution_device=__a		,	offload=__a )

										# Parameters have been offloaded, so on the meta device
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										# Buffers are not included in the offload by default, so are on the execution device
										snake_case__						=				torch.device(__a )
										self.assertEqual(model.batchnorm.running_mean.device		,	__a )

										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	__a )

										# Removing hooks loads back the weights in the model.
										remove_hook_from_submodules(__a )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# Now test with buffers included in the offload
										attach_align_device_hook(__a		,	execution_device=__a		,	offload=__a		,	offload_buffers=__a )

										# Parameters have been offloaded, so on the meta device, buffers included
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.running_mean.device		,	torch.device("meta" ) )

										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	__a )

										# Removing hooks loads back the weights in the model.
										remove_hook_from_submodules(__a )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )



						def A_			(			self ):
										snake_case__						=				ModelForTest()

										# Everything is on CPU
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# This will move each submodule on different devices
										snake_case__						=				0 if torch.cuda.is_available() else 'cpu'
										attach_align_device_hook(
										    __a		,	execution_device=__a		,	offload=__a		,	weights_map=model.state_dict() )

										# Parameters have been offloaded, so on the meta device
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										# Buffers are not included in the offload by default, so are on the execution device
										snake_case__						=				torch.device(__a )
										self.assertEqual(model.batchnorm.running_mean.device		,	__a )

										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	__a )

										# Removing hooks loads back the weights in the model.
										remove_hook_from_submodules(__a )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )

										# Now test with buffers included in the offload
										attach_align_device_hook(
										    __a		,	execution_device=__a		,	offload=__a		,	weights_map=model.state_dict()		,	offload_buffers=__a		,	)

										# Parameters have been offloaded, so on the meta device, buffers included
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("meta" ) )
										self.assertEqual(model.batchnorm.running_mean.device		,	torch.device("meta" ) )

										snake_case__						=				torch.randn(2		,	3 )
										snake_case__						=				model(__a )
										self.assertEqual(output.device		,	__a )

										# Removing hooks loads back the weights in the model.
										remove_hook_from_submodules(__a )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.batchnorm.weight.device		,	torch.device("cpu" ) )
										self.assertEqual(model.lineara.weight.device		,	torch.device("cpu" ) )





from itertools import zip_longest

import requests
from bsa import BeautifulSoup
from pandas import DataFrame





def 	SCREAMING_SNAKE_CASE__    (      __lowerCAmelCase = "laptop"							):
    snake_case__						=				F"""https://www.amazon.in/laptop/s?k={product}"""
    snake_case__						=				{
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36\n        (KHTML, like Gecko)Chrome/44.0.2403.157 Safari/537.36",
        "Accept-Language": "en-US, en;q=0.5",
    }
    snake_case__						=				BeautifulSoup(requests.get(__lowerCAmelCase    ,		headers=__lowerCAmelCase							).text							)
    # Initialize a Pandas dataframe with the column titles
    snake_case__						=				DataFrame(
        columns=[
            "Product Title",
            "Product Link",
            "Current Price of the product",
            "Product Rating",
            "MRP of the product",
            "Discount",
        ]							)
    # Loop through each entry and store them in the dataframe
    for item, _ in zip_longest(
        soup.find_all(
            "div"    ,		attrs={"class": "s-result-item", "data-component-type": "s-search-result"}    ,		)    ,		soup.find_all("div"    ,		attrs={"class": "a-row a-size-base a-color-base"}							)    ,		):
        try:
            snake_case__						=				item.ha.text
            snake_case__						=				"https://www.amazon.in/" + item.ha.a["href"]
            snake_case__						=				item.find("span"    ,		attrs={"class": "a-offscreen"}							).text
            try:
                snake_case__						=				item.find("span"    ,		attrs={"class": "a-icon-alt"}							).text
            except AttributeError:
                snake_case__						=				"Not available"
            try:
                snake_case__						=				(
                    "₹"
                    + item.find(
                        "span"    ,		attrs={"class": "a-price a-text-price"}							).text.split("₹"							)[1]
                )
            except AttributeError:
                snake_case__						=				""
            try:
                snake_case__						=				float(
                    (
                        (
                            float(product_mrp.strip("₹"							).replace(","    ,		""							)							)
                            - float(product_price.strip("₹"							).replace(","    ,		""							)							)
                        )
                        / float(product_mrp.strip("₹"							).replace(","    ,		""							)							)
                    )
                    * 100							)
            except ValueError:
                snake_case__						=				float("nan"							)
        except AttributeError:
            pass
        snake_case__						=				[
            product_title,
            product_link,
            product_price,
            product_rating,
            product_mrp,
            discount,
        ]
    snake_case__						=				" "
    snake_case__						=				" "
    data_frame.index += 1
    return data_frame


if __name__ == "__main__":
     __magic_name__            =  '''headphones'''
     get_amazon_product_data(product).to_csv(F'''Amazon Product Data for {product}.csv''')



