working with lamp information
This commit is contained in:
+13
-13
@@ -27,7 +27,7 @@ class MessageHeaderWrapper(BaseModel):
|
||||
class SMSMessage(BaseModel):
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
path: ClassVar[List[str]]
|
||||
params: ClassVar[Dict[str, str]]
|
||||
params: ClassVar[Dict[str, str]] = {}
|
||||
|
||||
MessageHeader: MessageHeaderWrapper
|
||||
|
||||
@@ -176,15 +176,15 @@ class SPLList(SMSMessage):
|
||||
class PrjLampInformation(SMSMessage):
|
||||
path = ['command', 'projector', 'maintenance', 'lamp', 'info', 'get']
|
||||
|
||||
ProjectorId: int
|
||||
LampId: int
|
||||
HoursOfUse: int
|
||||
LampSerialNumber: str
|
||||
LampMaximumWattage: int
|
||||
LampWattage: int
|
||||
LampLuminanceDecayRatio: int
|
||||
LampPower: float
|
||||
LampSelect: int
|
||||
LuminanceShutterControl: int
|
||||
LampOn: bool
|
||||
LampStatus: str
|
||||
ProjectorId: int | None
|
||||
LampId: int | None
|
||||
HoursOfUse: int | None
|
||||
LampSerialNumber: str | None
|
||||
LampMaximumWattage: int | None
|
||||
LampWattage: int | None
|
||||
LampLuminanceDecayRatio: int | None
|
||||
LampPower: float | None
|
||||
LampSelect: int | None
|
||||
LuminanceShutterControl: int | None
|
||||
LampOn: bool | None
|
||||
LampStatus: str | None
|
||||
|
||||
+24
-15
@@ -184,19 +184,19 @@ def main() -> int:
|
||||
identifiers=[f"projector_lamp_{l}"],
|
||||
manufacturer="Sony",
|
||||
model="Lamp",
|
||||
short_name=f"{sony_device.short_name}_lamp_{l}",
|
||||
short_name=f"lamp_{l}",
|
||||
via_device=sony_device.identifiers[0]
|
||||
)
|
||||
|
||||
lamp_device.sensors.append(HAsensor (
|
||||
name=f"Hours of Use",
|
||||
unique_id=f"{lamp_device.short_name}_hours",
|
||||
state_topic=f"{lamp_device.short_name}/state",
|
||||
state_topic=f"{sony_device.short_name}/{lamp_device.short_name}/state",
|
||||
value_template="{{ value_json.HoursOfUse }}",
|
||||
device_class="uptime",
|
||||
state_class="total_increasing",
|
||||
device_class="duration",
|
||||
state_class="measurement",
|
||||
unit_of_measurement="h",
|
||||
icon="mdi:timer-play-outline",
|
||||
icon="mdi:timer-sand-complete",
|
||||
request=PrjLampInformation,
|
||||
device=lamp_device
|
||||
))
|
||||
@@ -204,7 +204,7 @@ def main() -> int:
|
||||
lamp_device.sensors.append(HAsensor (
|
||||
name=f"Power",
|
||||
unique_id=f"{lamp_device.short_name}_power",
|
||||
state_topic=f"{lamp_device.short_name}/state",
|
||||
state_topic=f"{sony_device.short_name}/{lamp_device.short_name}/state",
|
||||
value_template="{{ value_json.LampPower }}",
|
||||
device_class="power",
|
||||
state_class="measurement",
|
||||
@@ -217,7 +217,7 @@ def main() -> int:
|
||||
lamp_device.sensors.append(HAsensor (
|
||||
name=f"Status",
|
||||
unique_id=f"{lamp_device.short_name}_status",
|
||||
state_topic=f"{lamp_device.short_name}/state",
|
||||
state_topic=f"{sony_device.short_name}/{lamp_device.short_name}/state",
|
||||
value_template="{{ value_json.LampStatus }}",
|
||||
state_class="measurement",
|
||||
icon="mdi:ab-testing",
|
||||
@@ -228,7 +228,7 @@ def main() -> int:
|
||||
lamp_device.sensors.append(HAsensor (
|
||||
name=f"Lamp ON",
|
||||
unique_id=f"{lamp_device.short_name}_lampon",
|
||||
state_topic=f"{lamp_device.short_name}/state",
|
||||
state_topic=f"{sony_device.short_name}/{lamp_device.short_name}/state",
|
||||
value_template="{{ value_json.LampOn }}",
|
||||
device_class="power",
|
||||
icon="mdi:lightbulb-on-outline",
|
||||
@@ -255,11 +255,17 @@ def main() -> int:
|
||||
try:
|
||||
resp: None | SMSMessage = None
|
||||
# update projector states
|
||||
lamp_num: int = 0
|
||||
for d in all_devices:
|
||||
states: Set[Type] = set([s.request for s in d.sensors])
|
||||
|
||||
for s in states:
|
||||
resp = projector.request(s)
|
||||
# include lamp id in the requests parameters
|
||||
if s == PrjLampInformation:
|
||||
resp = projector.request(s, reqParams={"lampId": lamp_num})
|
||||
lamp_num += 1
|
||||
else:
|
||||
resp = projector.request(s)
|
||||
# no respondse
|
||||
if resp is None:
|
||||
continue
|
||||
@@ -278,16 +284,19 @@ def main() -> int:
|
||||
resp.ShowStatusDetail.CurrentEventTitle = "Unknown or CPL"
|
||||
|
||||
# send only one message for sensor with this type of request
|
||||
for s in d.sensors:
|
||||
last_cls: Type = SMSMessage
|
||||
# filter sensors of this device per this type of request
|
||||
for sens in filter(lambda x: x.request == s, d.sensors):
|
||||
if last_cls == sens.request:
|
||||
continue
|
||||
if isinstance(resp, SMSMessage):
|
||||
sts: SMSMessage = resp
|
||||
mqtt.publish(topic=s.state_topic,
|
||||
mqtt.publish(topic=sens.state_topic,
|
||||
payload=sts.model_dump_json(
|
||||
indent=2,
|
||||
indent=1,
|
||||
exclude={"MessageHeader"},
|
||||
exclude_unset=True),
|
||||
retain=True)
|
||||
break
|
||||
exclude_unset=True))
|
||||
last_cls = sens.request
|
||||
except (Timeout, ConnectionError) as ce:
|
||||
LOGGER.debug("Projector is probably shutdown: %s", ce)
|
||||
except Exception as e:
|
||||
|
||||
@@ -43,7 +43,7 @@ class ProjectorConnection():
|
||||
return xmltodict.parse(response.text.replace("%2F", "/").replace("%3A", ":")) # Common containert for all messages
|
||||
return {}
|
||||
|
||||
def request(self, cls: Type[T], reqParams: Dict[str, Any]= {}) -> T | None:
|
||||
def request(self, cls: Type[T], reqParams: Dict[str, Any] = {}) -> T | None:
|
||||
rv: dict
|
||||
if TEST:
|
||||
with open(f'{getcwd()}/projmon/xmlresponses/showstatus.xml', 'r') as fd:
|
||||
|
||||
Reference in New Issue
Block a user