"""Models for IPP.""" # pylint: disable=R0912,R0915 from __future__ import annotations from dataclasses import asdict, dataclass from datetime import datetime, timedelta, timezone from typing import Any from yarl import URL from .parser import parse_ieee1284_device_id, parse_make_and_model PRINTER_STATES = {3: "idle", 4: "printing", 5: "stopped"} @dataclass class Info: """Object holding information from IPP.""" name: str printer_name: str printer_uri_supported: list[str] uptime: int command_set: str | None = None location: str | None = None manufacturer: str | None = None model: str | None = None printer_info: str | None = None serial: str | None = None uuid: str | None = None version: str | None = None more_info: str | None = None @staticmethod def from_dict(data: dict[str, Any]) -> Info: """Return Info object from IPP response.""" cmd = None name = "IPP Printer" name_parts = [] serial = None _printer_name = printer_name = data.get("printer-name", "") make_model = data.get("printer-make-and-model", "") device_id = data.get("printer-device-id", "") uri_supported = data.get("printer-uri-supported", []) uuid = data.get("printer-uuid") if not isinstance(uri_supported, list): uri_supported = [str(uri_supported)] for uri in uri_supported: if (URL(uri).path.lstrip("/")) == _printer_name.lstrip("/"): _printer_name = "" break make, model = parse_make_and_model(make_model) parsed_device_id = parse_ieee1284_device_id(device_id) if parsed_device_id.get("MFG") is not None and len(parsed_device_id["MFG"]) > 0: make = parsed_device_id["MFG"] name_parts.append(make) if parsed_device_id.get("MDL") is not None and len(parsed_device_id["MDL"]) > 0: model = parsed_device_id["MDL"] name_parts.append(model) if parsed_device_id.get("CMD") is not None and len(parsed_device_id["CMD"]) > 0: cmd = parsed_device_id["CMD"] if parsed_device_id.get("SN") is not None and len(parsed_device_id["SN"]) > 0: serial = parsed_device_id["SN"] if len(make_model) > 0: name = make_model elif len(name_parts) == 2: name = " ".join(name_parts) elif len(_printer_name) > 0: name = _printer_name return Info( command_set=cmd, location=data.get("printer-location", ""), name=name, manufacturer=make, model=model, printer_name=printer_name, printer_info=data.get("printer-info"), printer_uri_supported=uri_supported, serial=serial, uptime=data.get("printer-up-time", 0), uuid=uuid[9:] if uuid else None, # strip urn:uuid: from uuid version=data.get("printer-firmware-string-version"), more_info=data.get("printer-more-info"), ) @dataclass class Marker: """Object holding marker (ink) info from IPP.""" marker_id: int marker_type: str name: str color: str level: int low_level: int high_level: int @dataclass class Uri: """Object holding URI info from IPP.""" uri: str authentication: str | None security: str | None @dataclass class State: """Object holding the IPP printer state.""" printer_state: str reasons: str | None message: str | None @staticmethod def from_dict(data: dict[str, Any]) -> State: """Return State object from IPP response.""" state = data.get("printer-state", 0) if (reasons := data.get("printer-state-reasons")) == "none": reasons = None return State( printer_state=PRINTER_STATES.get(state, state), reasons=reasons, message=data.get("printer-state-message"), ) @dataclass class Printer: """Object holding the IPP printer information.""" info: Info markers: list[Marker] state: State uris: list[Uri] booted_at: datetime def as_dict(self) -> dict[str, Any]: """Return dictionary version of this printer.""" return { "info": asdict(self.info), "state": asdict(self.state), "markers": [asdict(marker) for marker in self.markers], "uris": [asdict(uri) for uri in self.uris], "booted_at": self.booted_at, } def update_from_dict(self, data: dict[str, Any]) -> Printer: """Return updated Printer object from IPP response data.""" last_uptime = self.info.uptime self.info = Info.from_dict(data) self.markers = Printer.merge_marker_data(data) self.state = State.from_dict(data) self.uris = Printer.merge_uri_data(data) if self.info.uptime < last_uptime: self.booted_at = _utcnow() - timedelta(seconds=self.info.uptime) return self @staticmethod def from_dict(data: dict[str, Any]) -> Printer: """Return Printer object from IPP response data.""" info = Info.from_dict(data) return Printer( info=info, markers=Printer.merge_marker_data(data), state=State.from_dict(data), uris=Printer.merge_uri_data(data), booted_at=(_utcnow() - timedelta(seconds=info.uptime)), ) @staticmethod def merge_marker_data( # noqa: PLR0912, C901 data: dict[str, Any], ) -> list[Marker]: """Return Marker data from IPP response.""" marker_names = [] marker_colors = [] marker_levels = [] marker_types = [] marker_highs = [] marker_lows = [] if not data.get("marker-names"): return [] if isinstance(data["marker-names"], list): marker_names = data["marker-names"] elif isinstance(data["marker-names"], str): marker_names = [data["marker-names"]] if not (mlen := len(marker_names)): return [] for _ in range(mlen): marker_colors.append("") marker_levels.append(-2) marker_types.append("unknown") marker_highs.append(100) marker_lows.append(0) if isinstance(data.get("marker-colors"), list): for index, list_value in enumerate(data["marker-colors"]): if index < mlen: marker_colors[index] = list_value elif isinstance(data.get("marker-colors"), str) and mlen == 1: marker_colors[0] = data["marker-colors"] if isinstance(data.get("marker-levels"), list): for index, list_value in enumerate(data["marker-levels"]): if index < mlen: marker_levels[index] = list_value elif isinstance(data.get("marker-levels"), int) and mlen == 1: marker_levels[0] = data["marker-levels"] if isinstance(data.get("marker-high-levels"), list): for index, list_value in enumerate(data["marker-high-levels"]): if index < mlen: marker_highs[index] = list_value elif isinstance(data.get("marker-high-levels"), int) and mlen == 1: marker_highs[0] = data["marker-high-levels"] if isinstance(data.get("marker-low-levels"), list): for index, list_value in enumerate(data["marker-low-levels"]): if index < mlen: marker_lows[index] = list_value elif isinstance(data.get("marker-low-levels"), int) and mlen == 1: marker_lows[0] = data["marker-low-levels"] if isinstance(data.get("marker-types"), list): for index, list_value in enumerate(data["marker-types"]): if index < mlen: marker_types[index] = list_value elif isinstance(data.get("marker-types"), str) and mlen == 1: marker_types[0] = data["marker-types"] markers = [ Marker( marker_id=marker_id, marker_type=marker_types[marker_id], name=marker_names[marker_id], color=marker_colors[marker_id], level=marker_levels[marker_id], high_level=marker_highs[marker_id], low_level=marker_lows[marker_id], ) for marker_id in range(mlen) ] markers.sort(key=lambda x: x.name) return markers @staticmethod def merge_uri_data(data: dict[str, Any]) -> list[Uri]: # noqa: PLR0912 """Return URI data from IPP response.""" _uris: list[str] = [] auth: list[str | None] = [] security: list[str | None] = [] if not data.get("printer-uri-supported"): return [] if isinstance(data["printer-uri-supported"], list): _uris = data["printer-uri-supported"] elif isinstance(data["printer-uri-supported"], str): _uris = [data["printer-uri-supported"]] if not (ulen := len(_uris)): return [] for _ in range(ulen): auth.append(None) security.append(None) if isinstance(data.get("uri-authentication-supported"), list): for k, list_value in enumerate(data["uri-authentication-supported"]): if k < ulen: auth[k] = _str_or_none(list_value) elif isinstance(data.get("uri-authentication-supported"), str) and ulen == 1: auth[0] = _str_or_none(data["uri-authentication-supported"]) if isinstance(data.get("uri-security-supported"), list): for k, list_value in enumerate(data["uri-security-supported"]): if k < ulen: security[k] = _str_or_none(list_value) elif isinstance(data.get("uri-security-supported"), str) and ulen == 1: security[0] = _str_or_none(data["uri-security-supported"]) return [ Uri( uri=_uris[uri_id], authentication=auth[uri_id], security=security[uri_id], ) for uri_id in range(ulen) ] def _utcnow() -> datetime: """Return the current date and time in UTC.""" return datetime.now(tz=timezone.utc) def _str_or_none(value: str) -> str | None: """Return string while handling string representations of None.""" if value == "none": return None return value