[Inference Providers] Add text-to-video support for Novita (#2922)

* Add text-to-video support for Novita

* remove test

* catch timeout for text-to-video

* silent typing warning

* fix

* revert

Author: hanouticelina

src/huggingface_hub/inference/_providers/__init__.py

@@ -14,7 +14,7 @@
 from .hf_inference import HFInferenceBinaryInputTask, HFInferenceConversational, HFInferenceTask
 from .hyperbolic import HyperbolicTextGenerationTask, HyperbolicTextToImageTask
 from .nebius import NebiusConversationalTask, NebiusTextGenerationTask, NebiusTextToImageTask
-from .novita import NovitaConversationalTask, NovitaTextGenerationTask
+from .novita import NovitaConversationalTask, NovitaTextGenerationTask, NovitaTextToVideoTask
 from .replicate import ReplicateTask, ReplicateTextToSpeechTask
 from .sambanova import SambanovaConversationalTask
 from .together import TogetherConversationalTask, TogetherTextGenerationTask, TogetherTextToImageTask
@@ -95,6 +95,7 @@
     "novita": {
         "text-generation": NovitaTextGenerationTask(),
         "conversational": NovitaConversationalTask(),
+        "text-to-video": NovitaTextToVideoTask(),
     },
     "replicate": {
         "text-to-image": ReplicateTask("text-to-image"),

src/huggingface_hub/inference/_providers/novita.py

@@ -1,7 +1,13 @@
+from typing import Any, Dict, Optional, Union
+
+from huggingface_hub.inference._common import _as_dict
 from huggingface_hub.inference._providers._common import (
     BaseConversationalTask,
     BaseTextGenerationTask,
+    TaskProviderHelper,
+    filter_none,
 )
+from huggingface_hub.utils import get_session
 
 
 _PROVIDER = "novita"
@@ -24,3 +30,27 @@ def __init__(self):
     def _prepare_route(self, mapped_model: str) -> str:
         # there is no v1/ route for novita
         return "/v3/openai/chat/completions"
+
+
+class NovitaTextToVideoTask(TaskProviderHelper):
+    def __init__(self):
+        super().__init__(provider=_PROVIDER, base_url=_BASE_URL, task="text-to-video")
+
+    def _prepare_route(self, mapped_model: str) -> str:
+        return f"/v3/hf/{mapped_model}"
+
+    def _prepare_payload_as_dict(self, inputs: Any, parameters: Dict, mapped_model: str) -> Optional[Dict]:
+        return {"prompt": inputs, **filter_none(parameters)}
+
+    def get_response(self, response: Union[bytes, Dict]) -> Any:
+        response_dict = _as_dict(response)
+        if not (
+            isinstance(response_dict, dict)
+            and "video" in response_dict
+            and isinstance(response_dict["video"], dict)
+            and "video_url" in response_dict["video"]
+        ):
+            raise ValueError("Expected response format: { 'video': { 'video_url': string } }")
+
+        video_url = response_dict["video"]["video_url"]
+        return get_session().get(video_url).content