


'''simple docstring'''







# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import torch

from accelerate import PartialState
from accelerate.utils.operations import broadcast, gather, gather_object, pad_across_processes, reduce


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	List[Any]  )  ->			Union[str, Any]:
       return (torch.arange(state.num_processes  ) + 1.0 + (state.num_processes * state.process_index)).to(state.device  )


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	List[str]  )  ->			List[Any]:
       _a    : List[str]     					=create_tensor(_UpperCAmelCase  )
       _a    : Tuple     					=gather(_UpperCAmelCase  )
       assert gathered_tensor.tolist() == list(range(1    ,state.num_processes**2 + 1  )  )


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Tuple  )  ->			Dict:
       _a    : Any     					=[state.process_index]
       _a    : List[Any]     					=gather_object(_UpperCAmelCase  )
       assert len(_UpperCAmelCase  ) == state.num_processes, F"{gathered_obj}, {len(_UpperCAmelCase  )} != {state.num_processes}"
       assert gathered_obj == list(range(state.num_processes  )  ), F"{gathered_obj} != {list(range(state.num_processes  )  )}"


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[Any]  )  ->			Tuple:
       _a    : Any     					=create_tensor(_UpperCAmelCase  )
       _a    : Tuple     					=broadcast(_UpperCAmelCase  )
       assert broadcasted_tensor.shape == torch.Size([state.num_processes]  )
       assert broadcasted_tensor.tolist() == list(range(1    ,state.num_processes + 1  )  )


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	List[str]  )  ->			int:
       # We need to pad the tensor with one more element if we are the main process
       # to ensure that we can pad
       if state.is_main_process:
              _a    : Dict     					=torch.arange(state.num_processes + 1  ).to(state.device  )
       else:
              _a    : List[Any]     					=torch.arange(state.num_processes  ).to(state.device  )
       _a    : Tuple     					=pad_across_processes(_UpperCAmelCase  )
       assert padded_tensor.shape == torch.Size([state.num_processes + 1]  )
       if not state.is_main_process:
              assert padded_tensor.tolist() == list(range(0    ,state.num_processes  )  ) + [0]


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	str  )  ->			List[str]:
       # For now runs on only two processes
       if state.num_processes != 2:
              return
       _a    : Optional[Any]     					=create_tensor(_UpperCAmelCase  )
       _a    : Optional[int]     					=reduce(_UpperCAmelCase    ,"""sum"""  )
       _a    : int     					=torch.tensor([4.0, 6]  ).to(state.device  )
       assert torch.allclose(_UpperCAmelCase    ,_UpperCAmelCase  ), F"{reduced_tensor} != {truth_tensor}"


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[int]  )  ->			Union[str, Any]:
       # For now runs on only two processes
       if state.num_processes != 2:
              return
       _a    : Tuple     					=create_tensor(_UpperCAmelCase  )
       _a    : Any     					=reduce(_UpperCAmelCase    ,"""mean"""  )
       _a    : Any     					=torch.tensor([2.0, 3]  ).to(state.device  )
       assert torch.allclose(_UpperCAmelCase    ,_UpperCAmelCase  ), F"{reduced_tensor} != {truth_tensor}"


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Tuple  )  ->			Union[str, Any]:
       # For xla_spawn (TPUs)
       main()


def 	SCREAMING_SNAKE_CASE_					(  )  ->			str:
       _a    : str     					=PartialState()
       state.print(F"State: {state}"  )
       state.print("""testing gather"""  )
       test_gather(_UpperCAmelCase  )
       state.print("""testing gather_object"""  )
       test_gather_object(_UpperCAmelCase  )
       state.print("""testing broadcast"""  )
       test_broadcast(_UpperCAmelCase  )
       state.print("""testing pad_across_processes"""  )
       test_pad_across_processes(_UpperCAmelCase  )
       state.print("""testing reduce_sum"""  )
       test_reduce_sum(_UpperCAmelCase  )
       state.print("""testing reduce_mean"""  )
       test_reduce_mean(_UpperCAmelCase  )


if __name__ == "__main__":
   main()






'''simple docstring'''







from __future__ import annotations

import requests


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	str  )  ->			dict:
       _a    : Any     					=F"https://hacker-news.firebaseio.com/v0/item/{story_id}.json?print=pretty"
       return requests.get(_UpperCAmelCase  ).json()


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	int = 10  )  ->			list[dict]:
       _a    : Union[str, Any]     					="""https://hacker-news.firebaseio.com/v0/topstories.json?print=pretty"""
       _a    : int     					=requests.get(_UpperCAmelCase  ).json()[:max_stories]
       return [get_hackernews_story(_UpperCAmelCase  ) for story_id in story_ids]


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	int = 10  )  ->			str:
       _a    : Union[str, Any]     					=hackernews_top_stories(_UpperCAmelCase  )
       return "\n".join("""* [{title}]({url})""".format(**_UpperCAmelCase  ) for story in stories  )


if __name__ == "__main__":
   print(hackernews_top_stories_as_markdown())


