File size: 2,078 Bytes
d5e1b1b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
# Copyright (C) 2023 Deforum LLC
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

# Contact the authors: https://deforum.github.io/

from pydantic import BaseModel
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass
from enum import Enum

class Batch(BaseModel):
    deforum_settings : Optional[Union[Dict[str, Any],List[Dict[str, Any]]]]
    options_overrides : Optional[Dict[str, Any]]

class DeforumJobStatusCategory(str, Enum):
    ACCEPTED = "ACCEPTED"
    SUCCEEDED = "SUCCEEDED"
    FAILED = "FAILED"
    CANCELLED = "CANCELLED"

class DeforumJobPhase(str, Enum):
    QUEUED = "QUEUED"
    PREPARING = "PREPARING"
    GENERATING = "GENERATING"
    POST_PROCESSING = "POST_PROCESSING"
    DONE = "DONE"

class DeforumJobErrorType(str, Enum):
    NONE = "NONE"
    RETRYABLE = "RETRYABLE"
    TERMINAL = "TERMINAL"

@dataclass(frozen=True)
class DeforumJobStatus(BaseModel):
    id: str
    status : DeforumJobStatusCategory
    phase : DeforumJobPhase
    error_type : DeforumJobErrorType
    phase_progress : float
    started_at: float
    last_updated: float 
    execution_time: float           # time between job start and the last status update
    update_interval_time: float     # time between the last two status updates
    updates: int                    # number of status updates so far
    message: Optional[str]
    outdir: Optional[str]
    timestring: Optional[str]
    deforum_settings : Optional[List[Dict[str, Any]]]
    options_overrides : Optional[Dict[str, Any]]