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