feat: capture screenshot from mtda

Add feature to capture screenshots from HDMI streamer on MTDA.

Signed-off-by: Shivaschandra KL <shivaschandra.k-l@siemens.com>

Author: Shivaschandra KL

mtda-cli

@@ -17,6 +17,7 @@ import time
 import sys
 import socket
 from argparse import ArgumentParser, RawTextHelpFormatter
+from pathlib import Path
 
 # Local imports
 from mtda.main import MultiTenantDeviceAccess
@@ -88,6 +89,27 @@ class Application:
 
     def debug(self, level, msg):
         self.client().debug(level, msg)
+    
+    def capture_screen(self,args):
+        file_name = args.capture_file_name
+        file_path = Path(args.capture_file_name)
+        parent_dir = file_path.parent
+        valid_extensions = ('.png', '.jpg', '.jpeg')
+        if not file_name.lower().endswith(valid_extensions):
+            print(f"'{input_string}' is not a valid image path (must end in .png ,.jpg or .jpeg).")
+            return False
+        if parent_dir.exists() and parent_dir.is_dir():
+            print(f"Path leading to '{file_path.name}' is valid. Proceeding with capture...")
+        else:
+            print(f"Error: The directory '{parent_dir}' does not exist.")
+            return False
+        response = self.client().capture_screen()
+        if response is not None:
+            with open(file_name, "wb") as f:
+                f.write(response)
+            print(f"Captured image saved at {file_name}")
+        else:
+            print("Failed to capture image")
 
     def command_cmd(self, args):
         result = self.client().command(args.cmd_command)
@@ -612,6 +634,20 @@ class Application:
             "cmd_command", metavar="cmd", type=str, help="Command to send"
         )
         p.set_defaults(func=cmd)
+        
+        # subcommand: capture
+        cmd = self.capture_screen
+        p = subparsers.add_parser(
+            "capture",
+            help="This utility helps you to capture screenshot of current diplay on monitor.\n"
+        )
+        p.add_argument(
+            "capture_file_name",
+            metavar="capture_file_name",
+            type=str,
+            help="Filename to store the captured image, .png or .jpg"
+        )
+        p.set_defaults(func=cmd)
 
         # subcommand: console
         cmd = self.console_cmd

mtda/client.py

@@ -82,6 +82,9 @@ def wrapper(*args, **kwargs):
             return wrapper
         return attr
 
+    def capture_screen(self, filename):
+        return self._impl.capture_screen()
+
     def console_prefix_key(self):
         return self._agent.console_prefix_key()
 

mtda/main.py

@@ -181,6 +181,23 @@ def _composite_stop(self):
         self.mtda.debug(3, f"main._composite_stop(): {result}")
         return result
 
+    @Pyro4.expose
+    def capture_screen(self):
+        self.mtda.debug(3, f"main.capture_screen()")
+        result = None
+        if self.video is not None:
+            if self.video.variant == "mjpg_streamer":
+                url = self.video.url() +  "/?action=snapshot"
+                self.mtda.debug(3, f"main.capture_screen(): {url} ")
+                try:
+                    result = requests.get(url, timeout=5)
+                    result.raise_for_status()
+                    return result.content
+                except Exception as e:
+                    self.mtda.debug(3, f"main.captue screenshot(): {e} ")
+         self.mtda.debug(3, f"main.captue screenshot(): {result} ")
+         return result
+
     @Pyro4.expose
     def config_set_power_timeout(self, timeout, **kwargs):
         self.mtda.debug(3, "main.config_set_power_timeout()")
@@ -1560,6 +1577,8 @@ def video_url(self, host="", opts=None, **kwargs):
         self.session_ping(session)
         if self.video is not None:
             result = self.video.url(host, opts)
+            if self.video.variant == "mjpg_streamer":
+                result = result + "/?action=stream"
 
         self.mtda.debug(3, f"main.video_url(): {result}")
         return result

mtda/video/mjpg_streamer.py

@@ -154,7 +154,7 @@ def url(self, host="", opts=None):
             host = socket.getfqdn()
             self.mtda.debug(3, "video.mjpg_streamer."
                                f"url: using host='{str(host)}'")
-        result = f"http://{host}:{self.port}/?action=stream"
+        result = f"http://{host}:{self.port}"
 
         self.mtda.debug(3, f"video.mjpg_streamer.url(): {str(result)}")
         return result