ISocketMessage
hash
[ str, required ]
The hash of the current task.
This is used internally and should not be used elsewhere.
status
[ SocketStatus, required ]
Status of the current task.
total
[ int, required ]
Total number of tasks.
pending
[ int, required ]
Number of pending tasks.
message
[ str, required ]
Message of the current status.
data
[ ISocketResponse, required ]
Response data.
Reference
SocketStatus
class SocketStatus(str, Enum):
PENDING = "pending"
WORKING = "working"
FINISHED = "finished"
EXCEPTION = "exception"
INTERRUPTED = "interrupted"
ISocketResponse
progress
[ float | None, default: None - ≥0, ≤1 ]
Progress of current task, if any.
intermediate
[ ISocketIntermediate | None, default: None ]
Intermediate responses, if any.
final
[ Dict[str, Any] | None, default: None ]
Final response, if any.
Although not typed here, final actually needs to follow the following schema in most cases:
export type IPythonResults = (
| { type: "text"; value: { text: string; safe: boolean; reason: string }[] }
| { type: "image"; value: { w: number; h: number; url: string; safe: boolean; reason: string }[] }
) & { extra?: Dictionary<any> };
Luckily, we've already provided the ResponseMiddleware to handle the conversion from raw results to the above schema, so normally you don't need to worry about it.
injections
[ Dict[str, Any] | None, default: None ]
Injections, if any.
See Injections for what an 'injection' is.
elapsedTimes
[ ElapsedTimes | None, default: None ]
Elapsed times.
ISocketIntermediate
imageList
[ List[str] | None, default: None ]
Intermediate images, if any.
textList
[ List[str] | None, default: None ]
Intermediate texts, if any.
ElapsedTimes
class ElapsedTimes(BaseModel):
createTime: Optional[float]
startTime: Optional[float]
endTime: Optional[float]
pending: Optional[float]
executing: Optional[float]
upload: Optional[float]
total: Optional[float]
def __init__(self, **data: Any):
super().__init__(**data)
self.createTime = time.time()
def start(self) -> None:
start = time.time()
self.startTime = start
if self.createTime is not None:
self.pending = start - self.createTime
def end(self) -> None:
end = time.time()
self.endTime = end
if self.startTime is not None:
self.executing = end - self.startTime
if self.upload is not None:
self.executing -= self.upload
if self.createTime is not None:
self.total = end - self.createTime