Old Barco projectors (initial implementation)

barco_rlmw_http/barco_rlwm_http.py

@@ -0,0 +1,67 @@
+import re
+import asyncio
+import aiohttp
+from asyncio.exceptions import CancelledError
+
+PORT = 43680
+RE_STATUS = re.compile(r".stats.txt(.+).value='(.+)';")
+REMOTEKEY_TIMEOUT = aiohttp.ClientTimeout(total=1)
+
+
+class BarcoRLM_Control:
+    def __init__(self, projector_ip):
+        self.projector_ip = projector_ip
+        self.session = None
+
+    async def _request(self, method, path, *args, **kwargs):
+        if not self.session:
+            self.session = aiohttp.ClientSession()
+        url = f"http://{self.projector_ip}{path}"
+        resp = await self.session.request(method, url, *args, **kwargs)
+        resp.raise_for_status()
+        return await resp.text()
+
+    async def get_status(self):
+        resp = await self._request("GET", "/tgi/firststatus.tgi")
+        matches = RE_STATUS.findall(resp)
+        status = dict(matches)
+        return status
+
+    async def toggle_power(self):
+        await self._request("GET", "/tgi/general.tgi?powertog")
+
+    async def click_remote(self, key):
+        try:
+            await self._request("GET", f"/tgi/remote.tgi?{key}", timeout=REMOTEKEY_TIMEOUT, raise_for_status=False)
+        except TimeoutError:
+            pass
+    
+    async def set_shutter(self, shutter):
+        endpoint = f"/tgi/general.tgi?pause{onoff(shutter)}"
+        await self._request("GET", endpoint)
+
+    async def set_power(self, power):
+        key = onoff(power) + "ky"
+        await self.click_remote(key)
+
+    async def set_input(self, input):
+        endpoint = f"/tgi/input.tgi?{input}"
+        await self._request("GET", endpoint)
+
+
+def onoff(state: bool) -> str:
+    if state:
+        return "on"
+    else:
+        return "off"
+
+if __name__ == "__main__":
+
+    async def main():
+        barco = BarcoRLM_Control("192.168.192.12")
+        status = await barco.get_status()
+        print(status)
+
+        await barco.click_remote("kymenu")
+
+    asyncio.run(main())