From 501ba01bed771b2948470cfae711c5d04522cac5 Mon Sep 17 00:00:00 2001 From: Emanuele Trabattoni Date: Fri, 6 Feb 2026 22:13:51 +0100 Subject: [PATCH] use pydantic dataclasses, members are optional.. not the best --- projmon/commands.py | 178 +++++++++++++++++++++-------------------- projmon/projmon.py | 11 ++- projmon/projrequest.py | 3 + 3 files changed, 102 insertions(+), 90 deletions(-) diff --git a/projmon/commands.py b/projmon/commands.py index 4223bb4..0f06ac4 100644 --- a/projmon/commands.py +++ b/projmon/commands.py @@ -1,122 +1,128 @@ from projrequest import ProjectorConnection from datetime import datetime -from pydantic import BaseModel -from typing import Dict, List, Any +from typing import Dict, List, Any, ClassVar, Optional from uuid import UUID +from pydantic import BaseModel, ConfigDict, Field class BaseCommand(BaseModel): - projector: ProjectorConnection - timestamp: datetime - type: str - content: Any + 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 __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) + 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'])) + self.timestamp = datetime.fromtimestamp(float(resp['timestamp'])/1000) self.type = resp['type'] - self.content = resp['body'][resp['type']] + return resp['body'] + return None + + def dump(self): + return self.model_dump_json(indent=2) class DCPInfo(BaseModel): - ID: UUID - Title: str - Path: str - Size: int - ImportTime: datetime - IsImported: bool - VerifyStatus: bool - ValidateStatus: bool - IsPlayable: bool - IsTransferred: bool + 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): - dcpInfoList: List[DCPInfo] - path: List[str] = ['content', 'dcp', 'info', 'list'] - params: Dict[str, str] = { 'formatDate': 'false' } + path = ['content', 'dcp', 'info', 'list'] + params = { 'formatDate': 'false' } + dcpInfoList: Optional[List[DCPInfo]] = None def get(self): - self.update(path=self.path, params=self.params) - self.dcpInfoList = [DCPInfo(**e) for e in self.content] + rv = self.update() + if rv: + self.dcpInfoList = [DCPInfo(**e) for e in rv] class PowerStatus(BaseModel): - Device: str - State: str + Device: Optional[str] = None + State: Optional[str] = None class PowerStatusList(BaseCommand): - powerStatusList: List[PowerStatus] - path: List[str] = ['status', 'sms', 'powerstatus'] - + powerStatusList: Optional[List[PowerStatus]] = None + path = ['status', 'sms', 'powerstatus'] + def get(self): - self.update(path=self.path) - self.powerStatusList = [PowerStatus(**e) for e in self.content] + rv = self.update() + if isinstance(rv, dict): + self.powerStatusList = [PowerStatus(**e) for e in rv['PowerStatus']] class ShowStatusDetailClass(BaseModel): - Type: str - Id: UUID - RemainingTime: int - ElapsedTime: int - TotalDuration: int - CurrentEventId: UUID - CurrentEventType: str - IsStoppedByMalfunction: bool - RewindTimeList: str - MalfunctionTime: int + 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: str - ShowStatusDetail: ShowStatusDetailClass - PlayBackMode: str - AtmosPlayingStatus: str - path: List[str] = ['playback', 'showstatus'] + PlayState: Optional[str] = None + ShowStatusDetail: Optional[ShowStatusDetailClass] = None + PlayBackMode: Optional[str] = None + AtmosPlayingStatus: Optional[str] = None + path = ['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'] + 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: int - BytesTransferred: int - PercentCompleted: int - InProgress: int - ImportPath: str - CompletionStatus: str - CompletionTime: str - DCPTitle: str + 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: int - BytesValidated: int - PercentCompleted: int - InProgress: bool - Id: UUID - CompletionStatus: str - CompletionTime: datetime + 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: int - ValidateAfterImport: bool - AggregatePercentValidated: int - State: str - ImportProgress: ImportProgressClass - ValidationProgressList: List[ValidationProgressClass] - IngestedByFolder: bool - ContentsTransferType: str + 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: bool - JobProgressList: List[JobProgress] - path: List[str] = ['content', 'dcp', 'command'] - params: Dict[str, str] = {'action': 'ListImportJobs'} + IsPaused: Optional[bool] = None + JobProgressList: Optional[List[JobProgress]] = None + path = ['content', 'dcp', 'command'] + params = {'action': 'ListImportJobs'} def get(self): - self.update(self.path) - self.IsPaused = self.content['IsPaused'] - self.JobProgressList = [JobProgress(**e) for e in self.content['JobProgressList']] + rv = self.update() + if isinstance(rv, dict): + self.IsPaused = rv['IsPaused'] + self.JobProgressList = [JobProgress(**e) for e in rv['JobProgressList']] diff --git a/projmon/projmon.py b/projmon/projmon.py index 7201c90..e16b00c 100644 --- a/projmon/projmon.py +++ b/projmon/projmon.py @@ -5,6 +5,7 @@ import logging import json import projrequest as projrequest +from commands import ShowStatus, PowerStatusList from utils import * from influxdb_client_3 import InfluxDBClient3, Point @@ -38,15 +39,17 @@ def main() -> int: username=env['PROJECTOR_USER'], password=env['PROJECTOR_PASSWORD'] ) + ps = PowerStatusList(projector=projector) + ss = ShowStatus(projector=projector) while handler.running: try: now:float = time.time() - resp = projector.get(path=['status', 'storage', 'info'], params={"area":"DCP"}) - if resp is not None: - print(json.dumps(resp, indent=2)) + ps.get() + ss.get() + print(ps.dump()) + print(ss.dump()) - last: float = time.time() cycle_time: float = last - now LOGGER.debug(f"Cycle Time: {cycle_time:4.3f}") diff --git a/projmon/projrequest.py b/projmon/projrequest.py index 2c74b54..742e2f8 100644 --- a/projmon/projrequest.py +++ b/projmon/projrequest.py @@ -1,4 +1,7 @@ import requests +import urllib3 +# Suppress only the single warning from urllib3. +urllib3.disable_warnings(category=urllib3.exceptions.InsecureRequestWarning) import xmltodict from requests.auth import HTTPBasicAuth from datetime import datetime