129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
from projrequest import ProjectorConnection
|
|
from datetime import datetime
|
|
|
|
from typing import Dict, List, Any, ClassVar, Optional
|
|
from uuid import UUID
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
class BaseCommand(BaseModel):
|
|
model_config = ConfigDict(arbitrary_types_allowed=True)
|
|
timestamp: datetime = datetime.now()
|
|
type: str = ''
|
|
projector: ProjectorConnection = Field(exclude=True)
|
|
path: ClassVar[List[str]]
|
|
params: ClassVar[Dict[str, Any] | None] = None
|
|
|
|
def update(self) -> List | Dict | None:
|
|
resp = self.projector.get(path=self.path, params=self.params)
|
|
if resp is not None:
|
|
self.timestamp = datetime.fromtimestamp(float(resp['timestamp'])/1000)
|
|
self.type = resp['type']
|
|
return resp['body']
|
|
return None
|
|
|
|
def dump(self):
|
|
return self.model_dump_json(indent=2)
|
|
|
|
class DCPInfo(BaseModel):
|
|
ID: Optional[UUID] = None
|
|
Title: Optional[str] = None
|
|
Path: Optional[str] = None
|
|
Size: Optional[int] = None
|
|
ImportTime: Optional[datetime] = None
|
|
IsImported: Optional[bool] = None
|
|
VerifyStatus: Optional[bool] = None
|
|
ValidateStatus: Optional[bool] = None
|
|
IsPlayable: Optional[bool] = None
|
|
IsTransferred: Optional[bool] = None
|
|
|
|
class DCPInfoList(BaseCommand):
|
|
path = ['content', 'dcp', 'info', 'list']
|
|
params = { 'formatDate': 'false' }
|
|
dcpInfoList: Optional[List[DCPInfo]] = None
|
|
|
|
def get(self):
|
|
rv = self.update()
|
|
if rv:
|
|
self.dcpInfoList = [DCPInfo(**e) for e in rv]
|
|
|
|
class PowerStatus(BaseModel):
|
|
Device: Optional[str] = None
|
|
State: Optional[str] = None
|
|
|
|
class PowerStatusList(BaseCommand):
|
|
powerStatusList: Optional[List[PowerStatus]] = None
|
|
path = ['status', 'sms', 'powerstatus']
|
|
|
|
def get(self):
|
|
rv = self.update()
|
|
if isinstance(rv, dict):
|
|
self.powerStatusList = [PowerStatus(**e) for e in rv['PowerStatus']]
|
|
|
|
class ShowStatusDetailClass(BaseModel):
|
|
Type: Optional[str] = None
|
|
Id: Optional[UUID] = None
|
|
RemainingTime: Optional[int] = None
|
|
ElapsedTime: Optional[int] = None
|
|
TotalDuration: Optional[int] = None
|
|
CurrentEventId: Optional[UUID] = None
|
|
CurrentEventType: Optional[str] = None
|
|
IsStoppedByMalfunction: Optional[bool] = None
|
|
RewindTimeList: Optional[str] = None
|
|
MalfunctionTime: Optional[int] = None
|
|
|
|
class ShowStatus(BaseCommand):
|
|
PlayState: Optional[str] = None
|
|
ShowStatusDetail: Optional[ShowStatusDetailClass] = None
|
|
PlayBackMode: Optional[str] = None
|
|
AtmosPlayingStatus: Optional[str] = None
|
|
path = ['playback', 'showstatus']
|
|
|
|
def get(self):
|
|
rv = self.update()
|
|
if isinstance(rv, dict):
|
|
self.PlayState = rv['PlayState']
|
|
self.ShowStatusDetail = ShowStatusDetailClass(**rv['ShowStatusDetail'])
|
|
self.PlayBackMode = rv['PlayBackMode']
|
|
self.AtmosPlayingStatus = rv['AtmosPlayingStatus']
|
|
|
|
class ImportProgressClass(BaseModel):
|
|
TotalBytesToTransfer: Optional[int] = None
|
|
BytesTransferred: Optional[int] = None
|
|
PercentCompleted: Optional[int] = None
|
|
InProgress: Optional[int] = None
|
|
ImportPath: Optional[str] = None
|
|
CompletionStatus: Optional[str] = None
|
|
CompletionTime: Optional[str] = None
|
|
DCPTitle: Optional[str] = None
|
|
|
|
class ValidationProgressClass(BaseModel):
|
|
TotalBytesToValidate: Optional[int] = None
|
|
BytesValidated: Optional[int] = None
|
|
PercentCompleted: Optional[int] = None
|
|
InProgress: Optional[bool] = None
|
|
Id: Optional[UUID] = None
|
|
CompletionStatus: Optional[str] = None
|
|
CompletionTime: Optional[datetime] = None
|
|
|
|
class JobProgress(BaseModel):
|
|
Id: Optional[int] = None
|
|
ValidateAfterImport: Optional[bool] = None
|
|
AggregatePercentValidated: Optional[int] = None
|
|
State: Optional[str] = None
|
|
ImportProgress: Optional[ImportProgressClass] = None
|
|
ValidationProgressList: Optional[List[ValidationProgressClass]] = None
|
|
IngestedByFolder: Optional[bool] = None
|
|
ContentsTransferType: Optional[str] = None
|
|
|
|
class DCPImportJobList(BaseCommand):
|
|
IsPaused: Optional[bool] = None
|
|
JobProgressList: Optional[List[JobProgress]] = None
|
|
path = ['content', 'dcp', 'command']
|
|
params = {'action': 'ListImportJobs'}
|
|
|
|
def get(self):
|
|
rv = self.update()
|
|
if isinstance(rv, dict):
|
|
self.IsPaused = rv['IsPaused']
|
|
self.JobProgressList = [JobProgress(**e) for e in rv['JobProgressList']]
|