Skip to content

Commit

Permalink
Add counter for print & copy
Browse files Browse the repository at this point in the history
  • Loading branch information
@sebcaps
sebcaps committed Apr 10, 2023
1 parent 7022c24 commit 963735d
Show file tree
Hide file tree
Showing 2 changed files with 105 additions and 55 deletions.
4 changes: 3 additions & 1 deletion example.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,11 @@ async def main() -> None:
print("Toner Cyan details:", printer.toner_status()["cyan"])
# Get the details about a tray
print("Tray 1 Capacity:", printer.input_tray_status()["tray_1"]["capa"])
print ('Get counter data.')
print (f'Print : {printer.print_count()}, Copies: {printer.copy_count()}')
# Print all available details from the printer
print("All data:\n", printer.raw())

print("All counter data:\n", printer.raw_counter())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
156 changes: 102 additions & 54 deletions pysyncthru/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,10 @@

from .htmlparsers import ENDPOINT_HTML_PARSERS

ENDPOINT_API = "/sws/app/information/home/home.json"
ENDPOINT_API = "/sws/app/information"
PRINTER_ENDPOINT = "/home/home.json"
COUNTER_ENDPOINT = "/counters/counters.json"



class ConnectionMode(Enum):
Expand All @@ -27,20 +30,23 @@ class SyncthruState(Enum):
TESTING = 4
ERROR = 5

class DataType(Enum):
PRINTER = 1
COUNTER = 2

def construct_url(ip_address: str) -> str:
"""Construct the URL with a given IP address."""
if "http://" not in ip_address and "https://" not in ip_address:
ip_address = "{}{}".format("http://", ip_address)
ip_address = f'http://{ip_address}'
if ip_address[-1] == "/":
ip_address = ip_address[:-1]
return ip_address


class SyncThruAPINotSupported(Exception):
"""Error raised when a printer does not provide access to a JSON based API."""


class SyncThruHTMLNotSupported(Exception):
"""Error raised when API does not provide access to a HTML scraping."""
class SyncThru:
"""Interface to communicate with the Samsung Printer with SyncThru."""

Expand All @@ -58,62 +64,89 @@ def __init__(
"""Initialize the the printer."""
self.url = construct_url(ip)
self._session = session
self.data = {} # type: Dict[str, Any]
self.data_printer_status = {} # type: Dict[str, Any]
self.data_counter_status = {} # type: Dict[str, Any]
self.connection_mode = connection_mode

async def update(self) -> None:
"""
Retrieve the data from the printer and caches is in the class
Throws ValueError if host does not support SyncThru
"""
self.data = await self._current_data()
self.data_printer_status = await self._current_data(DataType.PRINTER.value)
self.data_counter_status = await self._current_data(DataType.COUNTER.value)


async def _current_data(self) -> Dict[str, Any]:
async def _current_data(self, datatype: DataType) -> Dict[str, Any]:
"""
Retrieve the data from the printer.
Throws ValueError if host does not support SyncThru
"""
data = {"status": {"hrDeviceStatus": SyncthruState.OFFLINE.value}}
if self.connection_mode in [ConnectionMode.AUTO, ConnectionMode.API]:
url = "{}{}".format(self.url, ENDPOINT_API)

try:
async with self._session.get(url) as response:
res = demjson3.decode(
await response.text(), strict=False
) # type: Dict[str, Any]
# if we get something back from this endpoint,
# we directly return it
return res
except (aiohttp.ClientError, asyncio.TimeoutError):
pass
except demjson3.JSONDecodeError:
# If no JSON data is provided but we want to only connect
# in this mode, raise an Exception
if self.connection_mode != ConnectionMode.AUTO:
raise SyncThruAPINotSupported(
"Invalid host, does not support SyncThru JSON API."
)

if self.connection_mode in [ConnectionMode.AUTO, ConnectionMode.HTML]:

any_connection_successful = False
for endpoint_url, parsers in ENDPOINT_HTML_PARSERS.items():
html_url = "{}{}".format(self.url, endpoint_url)
if datatype == DataType.PRINTER.value:
data = {"status": {"hrDeviceStatus": SyncthruState.OFFLINE.value}}
if self.connection_mode in [ConnectionMode.AUTO, ConnectionMode.API]:
url = f'{self.url}{ENDPOINT_API}{PRINTER_ENDPOINT}'
try:
async with self._session.get(url) as response:
res = demjson3.decode(
await response.text(), strict=False
) # type: Dict[str, Any]
# if we get something back from this endpoint,
# we directly return it
return res
except (aiohttp.ClientError, asyncio.TimeoutError):
pass
except demjson3.JSONDecodeError:
# If no JSON data is provided but we want to only connect
# in this mode, raise an Exception
if self.connection_mode != ConnectionMode.AUTO:
raise SyncThruAPINotSupported(
"Invalid host, does not support SyncThru JSON API."
)

if self.connection_mode in [ConnectionMode.AUTO, ConnectionMode.HTML]:

any_connection_successful = False
for endpoint_url, parsers in ENDPOINT_HTML_PARSERS.items():
html_url = "{}{}".format(self.url, endpoint_url)
try:
async with self._session.get(html_url) as response:
html_res = await response.text()
any_connection_successful = True
for parser in parsers:
parser(data).feed(html_res)
# if successful, set device status to unknown
except (aiohttp.ClientError, asyncio.TimeoutError):
pass
if (
any_connection_successful
and data["status"]["hrDeviceStatus"] == SyncthruState.OFFLINE.value
):
data["status"]["hrDeviceStatus"] = SyncthruState.UNKNOWN.value
elif datatype == DataType.COUNTER.value:
if self.connection_mode in [ConnectionMode.AUTO, ConnectionMode.API]:
url = f'{self.url}{ENDPOINT_API}{COUNTER_ENDPOINT}'
print (f'In counter section with url {url}')
try:
async with self._session.get(html_url) as response:
html_res = await response.text()
any_connection_successful = True
for parser in parsers:
parser(data).feed(html_res)
# if successful, set device status to unknown
async with self._session.get(url) as response:
res = demjson3.decode(
await response.text(), strict=False
) # type: Dict[str, Any]
# if we get something back from this endpoint,
# we directly return it
return res
except (aiohttp.ClientError, asyncio.TimeoutError):
pass
if (
any_connection_successful
and data["status"]["hrDeviceStatus"] == SyncthruState.OFFLINE.value
):
data["status"]["hrDeviceStatus"] = SyncthruState.UNKNOWN.value
except demjson3.JSONDecodeError:
# If no JSON data is provided but we want to only connect
# in this mode, raise an Exception
if self.connection_mode != ConnectionMode.AUTO:
raise SyncThruAPINotSupported(
"Invalid host, does not support SyncThru JSON API."
)
#TODO Handle HTML Connection Mode. For the moment raise an exception
else:
raise SyncThruHTMLNotSupported("HTML scrapping not supported for counter data")

return data

Expand All @@ -134,7 +167,7 @@ def is_unknown_state(self) -> bool:

def _identity_data(self, key: str) -> Optional[str]:
try:
return cast(Dict[str, str], self.data.get("identity", {})).get(key)
return cast(Dict[str, str], self.data_printer_status.get("identity", {})).get(key)
except (KeyError, AttributeError):
return None

Expand Down Expand Up @@ -165,13 +198,13 @@ def ip_address(self) -> Optional[str]:
def device_status(self) -> SyncthruState:
"""Fetch the raw device status"""
try:
return SyncthruState(int(self.data.get("status", {}).get("hrDeviceStatus")))
return SyncthruState(int(self.data_printer_status.get("status", {}).get("hrDeviceStatus")))
except (ValueError, TypeError):
return SyncthruState.INVALID

def device_status_details(self) -> str:
"""Return the detailed (display) status of the device as string."""
head = self.data.get("status", {})
head = self.data_printer_status.get("status", {})
status_display = [
head.get("status{}".format(i), "").strip() for i in [1, 2, 3, 4]
]
Expand All @@ -181,14 +214,21 @@ def device_status_details(self) -> str:
def capability(self) -> Dict[str, Any]:
"""Return the capabilities of the printer."""
try:
return self.data.get("capability", {})
return self.data_printer_status.get("capability", {})
except (KeyError, AttributeError):
return {}

def raw(self) -> Dict[str, Any]:
"""Return all details of the printer."""
try:
return self.data
return self.data_printer_status
except (KeyError, AttributeError):
return {}

def raw_counter(self) -> Dict[str, Any]:
"""Return all details of the printer."""
try:
return self.data_counter_status
except (KeyError, AttributeError):
return {}

Expand All @@ -197,7 +237,7 @@ def toner_status(self, filter_supported: bool = True) -> Dict[str, Any]:
toner_status = {}
for color in self.COLOR_NAMES:
try:
toner_stat = self.data.get("{}_{}".format(SyncThru.TONER, color), {})
toner_stat = self.data_printer_status.get("{}_{}".format(SyncThru.TONER, color), {})
if filter_supported and toner_stat.get("opt", 0) == 0:
continue
else:
Expand All @@ -215,7 +255,7 @@ def input_tray_status(self, filter_supported: bool = True) -> Dict[str, Any]:
"manual",
):
try:
tray_stat = self.data.get(tray.replace("_", ""), {})
tray_stat = self.data_printer_status.get(tray.replace("_", ""), {})
if filter_supported and tray_stat.get("opt", 0) != 1:
continue
else:
Expand All @@ -228,7 +268,7 @@ def output_tray_status(self) -> Dict[int, Dict[str, str]]:
"""Return the state of all output trays."""
tray_status = {}
try:
tray_stat = self.data.get("outputTray", [])
tray_stat = self.data_printer_status.get("outputTray", [])
for i, stat in enumerate(tray_stat):
tray_status[i] = {
"name": stat[0],
Expand All @@ -244,11 +284,19 @@ def drum_status(self, filter_supported: bool = True) -> Dict[str, Any]:
drum_status = {}
for color in self.COLOR_NAMES:
try:
drum_stat = self.data.get("{}_{}".format(SyncThru.DRUM, color), {})
drum_stat = self.data_printer_status.get("{}_{}".format(SyncThru.DRUM, color), {})
if filter_supported and drum_stat.get("opt", 0) == 0:
continue
else:
drum_status[color] = drum_stat
except (KeyError, AttributeError):
drum_status[color] = {}
return drum_status

def print_count(self) -> Dict [str, Any]:
"""Return number of prints"""
return self.data_counter_status.get("GXI_BILLING_PRINT_TOTAL_IMP_CNT")

def copy_count(self) -> Dict [str, Any]:
"""Return number of copies"""
return self.data_counter_status.get("GXI_BILLING_COPY_TOTAL_IMP_CNT")

0 comments on commit 963735d

Please sign in to comment.