Merge pull request #1 from guanana/extra_endpoints

Extra endpoints

.github/workflows/publish.yml

@@ -9,8 +9,10 @@
 
 name: Publish Python distribution to PyPI and Github
 
-on: push
-
+on:
+  push:
+    branches:
+    - main
 jobs:
   build:
     runs-on: ubuntu-latest

README.md

@@ -7,6 +7,9 @@ With PiKVM, you can easily perform various actions on your PiKVM devices, such a
 * Controlling ATX power
 * Managing Mass Storage Device (MSD) images
 * Interacting with General-Purpose Input/Output (GPIO) channels
+* Taking snapshots and receive image
+* Reading snapshots with OCR and receive text representation
+* Send keys to the server
 
 
 With PiKVM, you can automate tasks, integrate PiKVMs into your existing applications, 
@@ -20,14 +23,15 @@ To install PiKVM, simply run the following command in your terminal:
 pip install pikvm-lib
 ```
 
-## Usage
+## PiKVM device control
 
 After installing PiKVM, you can import it into your Python script and create an instance of the `PiKVM` class. The `PiKVM` class constructor takes the following parameters:
 
 * `hostname`: The hostname or IP address of the PiKVM device
 * `username`: The username for authentication
 * `password`: The password for authentication
 
+
 ```python
 from pikvm_lib import PiKVM
 
@@ -51,7 +55,7 @@ pikvm_instance.set_atx_power(action="on")
 For more information on how to use PiKVM, 
 please refer to the official documentation: [PiKVM official web](https://docs.pikvm.org/) and [PiKVM API Reference](https://docs.pikvm.org/api/)
 
-## Examples
+### Usage examples
 
 Here are some examples of how to use PiKVM to perform common tasks:
 
@@ -88,3 +92,59 @@ pikvm_instance.connect_msd()
 ```python
 pikvm_instance.switch_gpio_channel(channel=1, state=1)
 ```
+
+* **Take snapshot and receive OCR text:**
+
+```python
+pikvm_instance.get_streamer_snapshot(snapshot_path="/home/user/pikvm-snapshots",
+                            filename="test.txt", ocr=True)
+```
+* **Take snapshot and receive image:**
+
+```python
+pikvm_instance.get_streamer_snapshot(snapshot_path="/home/user/pikvm-snapshots",
+                            filename="test.jpeg", ocr=False)
+```
+## PiKVM websocket
+The PiKVMWebsocket class is a Python class that allows you to send keyboard events to a PiKVM server over WebSocket. 
+
+It provides methods for sending individual keys, key combinations, and text input. 
+
+The class also handles the connection to the PiKVM server and the parsing of the WebSocket messages.
+
+### Usage examples
+```python
+from pikvm_lib import PiKVMWebsocket
+
+hostname = "192.168.1.10"  # Replace with your PiKVM server's hostname or IP address
+username = "user"
+password = "password"
+
+# Create a PiKVMWebsocket object
+websocket = PiKVMWebsocket(hostname, username, password)
+
+# Send the Ctrl+Alt+Delete key combination
+websocket.send_ctrl_alt_sup()
+
+# Send the text "Hello, world!"
+websocket.send_input("Hello, world!")
+```
+
+```python
+from pikvm_lib import PiKVMWebsocket
+
+hostname = "192.168.1.10"  # Replace with your PiKVM server's hostname or IP address
+username = "user"
+password = "password"
+
+# Create a PiKVMWebsocket object
+websocket = PiKVMWebsocket(hostname, username, password)
+
+# Send the F2 key
+websocket.send_key("<F2>")
+
+# Send the Ctrl+B key
+websocket.send_key_press("ControlLeft", "true")
+websocket.send_input("b") # or websocket.send_key("KeyB")
+websocket.send_key_press("ControlLeft", "false")
+```

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pikvm_lib"
-version = "0.2.0"
+version = "0.3.0"
 authors = [
   { name="guanana", email="[email protected]" },
 ]
@@ -12,6 +12,7 @@ requires-python = ">=3.9"
 dependencies = [
     "requests>=2.31.0",
     "pyotp<=2.9.0",
+    "websocket-client>=1.7.0"
 ]
 
 
@@ -33,7 +34,7 @@ Homepage = "https://github.com/guanana/pikvm-lib"
 Issues = "https://github.com/guanana/pikvm-lib/issues"
 
 [tool.bumpver]
-current_version = "0.2.0"
+current_version = "0.3.0"
 version_pattern = "MAJOR.MINOR.PATCH"
 commit_message = "Bump version {old_version} -> {new_version}"
 tag_message = "{new_version}"

requirements.txt

@@ -16,3 +16,5 @@ requests==2.31.0
     # via pikvm-lib (pyproject.toml)
 urllib3==2.1.0
     # via requests
+websocket-client==1.7.0
+    # via pikvm-lib (pyproject.toml)

src/pikvm_lib/__init__.py

@@ -1,5 +1,6 @@
 import logging
 from .pikvm import PiKVM
+from .pikvm_websocket import PiKVMWebsocket
 
 
 # Create a logger

src/pikvm_lib/_streamer.py

@@ -0,0 +1,47 @@
+import os
+
+
+def get_streamer_state(self, path="/api/streamer"):
+    """
+    Gets the state of the streamer subsystem.
+
+    Parameters:
+    - path: The Streamer state endpoint path (default is "/api/streamer").
+
+    Returns:
+    - dict: Streamer subsystem state information.
+    """
+    return self._get_infos(path)
+
+
+def get_streamer_snapshot(self, path="/api/streamer/snapshot", snapshot_path=os.getcwd(), filename="snapshot.jpeg",
+                          ocr=False):
+    """
+    Gets screen snapshot
+
+    Parameters:
+    - path: The Streamer state endpoint path (default is "/api/streamer").
+    - snapshot_path: Folder to download the snapshot to (default is _current path_)
+    - filename: Name of the snapshot or text file name if ocr is enabled (default is "snapshot.jpeg")
+    - ocr: Enable OCR recognition and creates a text file instead
+
+    Returns:
+    - file: Path to the file.
+    """
+    options = "allow_offline=1"
+    if ocr:
+        if filename.endswith(".jpeg"):
+            self.logger.warning("Detected OCR but jpeg extension found, changing to txt")
+            filename = f"{filename.strip('.jpeg')}.txt"
+        options = "ocr=1&allow_offline=1"
+    else:
+        if filename.endswith(".txt"):
+            self.logger.warning("OCR off but txt extension found, changing to jpeg")
+            filename = f"{filename.strip('.txt')}.jpeg"
+
+    file = os.path.join(snapshot_path, filename)
+    f = open(file, 'wb')
+    f.write(self._get(path, options).content)
+    f.close()
+    self.logger.info(f"Writing snapshot to: {file}")
+    return file

src/pikvm_lib/keymap.csv

@@ -0,0 +1,52 @@
+characters,web_name
+\n,Enter
+\b,Backspace
+\t,Tab
+\s,Space
+-,Minus
+=,Equal
+[,BracketLeft
+],BracketRight
+\,Backslash
+;,Semicolon
+',Quote
+`,Backquote
+",",Comma
+.,Period
+/,Slash
+<Esc>,Escape
+<Enter>,Enter
+<CapsLock>,CapsLock
+<F1>,F1
+<F2>,F2
+<F3>,F3
+<F4>,F4
+<F5>,F5
+<F6>,F6
+<F7>,F7
+<F8>,F8
+<F9>,F9
+<F10>,F10
+<F11>,F11
+<F12>,F12
+<PrintScreen>,PrintScreen
+<Insert>,Insert
+<Home>,Home
+<PageUp>,PageUp
+<Delete>,Delete
+<End>,End
+<PageDown>,PageDown
+<ArrowRight>,ArrowRight
+<ArrowLeft>,ArrowLeft
+<ArrowDown>,ArrowDown
+<ArrowUp>,ArrowUp
+<ControlLeft>,ControlLeft
+<ShiftLeft>,ShiftLeft
+<AltLeft>,AltLeft
+<MetaLeft>,MetaLeft
+<ControlRight>,ControlRight
+<ShiftRight>,ShiftRight
+<AltRight>,AltRight
+<MetaRight>,MetaRight
+<Pause>,Pause
+<Power>,Power

src/pikvm_lib/keymap_shift.csv

@@ -0,0 +1,22 @@
+characters,web_name
+_,Minus
++,Equal
+{,BracketLeft
+},BracketRight
+|,Backslash
+:,Semicolon
+\",Quote
+~,Backquote
+<,Comma
+>,Period
+?,Slash
+!,Digit1
+@,Digit2
+#,Digit3
+$,Digit4
+%,Digit5
+^,Digit6
+&,Digit7
+*,Digit8
+(,Digit9
+),Digit0

src/pikvm_lib/pikvm.py

@@ -6,12 +6,7 @@
 import logging
 
 
-class PiKVM:
-    from ._atx import get_atx_state, set_atx_power, click_atx_button
-    from ._msd import (get_msd_state, set_msd_parameters, connect_msd, disconnect_msd, reset_msd,
-                       upload_msd_remote, upload_msd_image, remove_msd_image)
-    from ._gpio import pulse_gpio_channel, switch_gpio_channel, get_gpio_state
-
+class _BuildPiKVM:
     def __init__(self, hostname, username, password, secret=None, schema="https://", cert_trusted=False):
         """
         Initializes the PiKVM object.
@@ -24,23 +19,15 @@ def __init__(self, hostname, username, password, secret=None, schema="https://",
         - schema: The protocol schema, either "http" or "https" (default is "https").
         - cert_trusted: Whether the SSL certificate is issue by a trusted authority or not (default is False).
         """
-        self.logger = logging.getLogger(__name__)
-        try:
-            self.schema = re.findall(r"(http|https)://", schema.lower())[0]
-        except IndexError:
-            self.logger.error("Schema must be http or https")
-            raise
-        self.certificate_trusted = cert_trusted
-        if not self.certificate_trusted:
-            requests.packages.urllib3.disable_warnings()
-
         self.hostname = hostname
-        self.base_url = f"{self.schema}://{self.hostname}/placeholder"
         self.username = username
         self.password = password
         self.secret = secret  # Can be found in /etc/kvmd/totp.secret
+        self.certificate_trusted = cert_trusted
+        self.logger = logging.getLogger(__name__)
+        self.schema = schema
         self.headers = self._set_headers()
-        self.systeminfo = self.get_system_info()
+        self.base_url = f"{self.schema}://{self.hostname}/placeholder"
 
     def _set_headers(self):
         """
@@ -60,6 +47,23 @@ def _set_headers(self):
         }
         return headers
 
+
+class PiKVM(_BuildPiKVM):
+    from ._atx import get_atx_state, set_atx_power, click_atx_button
+    from ._msd import (get_msd_state, set_msd_parameters, connect_msd, disconnect_msd, reset_msd,
+                       upload_msd_remote, upload_msd_image, remove_msd_image)
+    from ._gpio import pulse_gpio_channel, switch_gpio_channel, get_gpio_state
+    from ._streamer import get_streamer_state, get_streamer_snapshot
+
+    def __init__(self, hostname, username, password, secret=None, schema="https", cert_trusted=False):
+        super().__init__(hostname, username, password, secret, schema, cert_trusted)
+        if self.schema not in ["http", "https"]:
+            self.logger.error("Schema must be http or https")
+            raise
+        if not self.certificate_trusted:
+            requests.packages.urllib3.disable_warnings()
+        self.systeminfo = self.get_system_info()
+
     def _call(self, path, options=None):
         """
         Forms the complete URL for an API call.
@@ -71,7 +75,8 @@ def _call(self, path, options=None):
         Returns:
         - str: The complete URL.
         """
-        self.logger.debug(f"Parameters: schema: {self.schema}, hostname: {self.hostname}, path={path}, options={options}")
+        self.logger.debug(
+            f"Parameters: schema: {self.schema}, hostname: {self.hostname}, path={path}, options={options}")
         if not options:
             url = urljoin(self.base_url, path)
         else:
@@ -207,4 +212,3 @@ def get_prometheus_metrics(self, path="/api/export/prometheus/metrics"):
         - dict: Prometheus metrics information.
         """
         return self._get_infos(path)
-