[wip] utils

osfv_cli/Makefile

@@ -8,7 +8,7 @@ build:
 	poetry build
 
 install: build
-	pip install dist/$(PACKAGE_NAME)-$(VERSION)-py3-none-any.whl
+	pip install --force-reinstall dist/$(PACKAGE_NAME)-$(VERSION)-py3-none-any.whl
 	test -f $(SNIPEIT_CONFIG_FILE) || install -D -m 644 snipeit.yml $(SNIPEIT_CONFIG_FILE)
 	test -f $(ZABBIX_CONFIG_FILE) || install -D -m 644 zabbix.yml $(ZABBIX_CONFIG_FILE)
 

osfv_cli/pyproject.toml

@@ -3,14 +3,14 @@ requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.scripts]
-osfv_cli = "osfv_cli.osfv_cli:main"
+osfv_cli = "osfv.cli.cli:main"
 
 [tool.poetry]
-name = "osfv_cli"
-version = "0.4.2"
+name = "osfv"
+version = "0.4.4"
 description = "Open Source Firmware Validation Command Line Interface Tool"
 authors = ["Maciej Pijanowski <[email protected]>"]
-include = ["osfv_cli/models/*.yml"]
+include = ["src/models/*.yml"]
 
 [tool.poetry.dependencies]
 python = "^3.8"
@@ -21,3 +21,8 @@ requests = "^2.31.0"
 unidecode = "^1.3.6"
 importlib-resources = "^6.1.1"
 voluptuous = "^0.14.2"
+
+# # Include your packages here
+# osfv_cli = {path = "./src/osfv_cli"}
+# osfv_libs = {path = "./src/osfv_libs"}
+# osfv_rf = {path = "./src/osfv_rf"}

osfv_cli/src/osfv/__main__.py

@@ -0,0 +1,6 @@
+import sys
+
+if __name__ == "__main__":
+    from osfv.osfv_cli.osfv_cli import main
+
+    sys.exit(main())

osfv_cli/osfv_cli/osfv_cli.py → osfv_cli/src/osfv/cli/cli.py

@@ -7,11 +7,10 @@
 
 import pexpect
 import requests
-
-from .rte import RTE
-from .snipeit_api import SnipeIT
-from .sonoff_api import SonoffDevice
-from .zabbix import Zabbix
+from osfv.libs.rte import RTE
+from osfv.libs.snipeit_api import SnipeIT
+from osfv.libs.sonoff_api import SonoffDevice
+from osfv.libs.zabbix import Zabbix
 
 
 # Check out an asset
@@ -72,6 +71,7 @@ def list_my_assets(snipeit_api, args):
         for asset in my_assets:
             print_asset_details(asset)
 
+
 # List unused assets
 def list_unused_assets(snipeit_api, args):
     all_assets = snipeit_api.get_all_assets()
@@ -154,10 +154,10 @@ def print_asset_details_for_zabbix(asset):
 
 def relay_toggle(rte, args):
     state_str = rte.relay_get()
-    if state_str == "low":
-        new_state_str = "high"
+    if state_str == "off":
+        new_state_str = "on"
     else:
-        new_state_str = "low"
+        new_state_str = "off"
     rte.relay_set(new_state_str)
     state = rte.relay_get()
     print(f"Relay state toggled. New state: {state}")
@@ -465,7 +465,7 @@ def update_zabbix_assets(snipeit_api):
 def main():
     parser = argparse.ArgumentParser(description="Open Source Firmware Validation CLI")
     parser.add_argument(
-        "-v", "--version", action="version", version=metadata.version("osfv_cli")
+        "-v", "--version", action="version", version=metadata.version("osfv")
     )
 
     parser.add_argument(
@@ -621,8 +621,8 @@ def main():
     set_rel_parser.add_argument(
         "state",
         choices=[
-            "high",
-            "low",
+            "on",
+            "off",
         ],
         help="Relay state",
     )

osfv_cli/osfv_cli/rte.py → osfv_cli/src/osfv/libs/rte.py

@@ -4,11 +4,9 @@
 import paramiko
 import yaml
 from importlib_resources import files
+from osfv.libs.rtectrl_api import rtectrl
 from voluptuous import Any, Optional, Required, Schema
 
-from .rtectrl_api import rtectrl
-from .sonoff_api import SonoffDevice
-
 
 class RTE(rtectrl):
     GPIO_SPI_ON = 1
@@ -30,15 +28,14 @@ class RTE(rtectrl):
     PROGRAMMER_CH341A = "ch341a_spi"
     FLASHROM_CMD = "flashrom -p {programmer} {args}"
 
-    def __init__(self, rte_ip, dut_model, snipeit_api):
+    def __init__(self, rte_ip, dut_model, sonoff):
         self.rte_ip = rte_ip
         self.dut_model = dut_model
         self.dut_data = self.load_model_data()
-        self.snipeit_api = snipeit_api
-        self.sonoff, self.sonoff_ip = self.init_sonoff()
+        self.sonoff = sonoff
 
     def load_model_data(self):
-        file_path = os.path.join(files("osfv_cli"), "models", f"{self.dut_model}.yml")
+        file_path = os.path.join(files("osfv"), "models", f"{self.dut_model}.yml")
 
         # Check if the file exists
         if not os.path.isfile(file_path):
@@ -98,20 +95,6 @@ def load_model_data(self):
         # Return the loaded data
         return data
 
-    def init_sonoff(self):
-        sonoff_ip = ""
-        sonoff = None
-
-        if self.dut_data["pwr_ctrl"]["sonoff"] is True:
-            sonoff_ip = self.snipeit_api.get_sonoff_ip_by_rte_ip(self.rte_ip)
-            if not sonoff_ip:
-                raise SonoffNotFound(
-                    exit(f"Sonoff IP not found in SnipeIT for RTE: {self.rte_ip}")
-                )
-            sonoff = SonoffDevice(sonoff_ip)
-
-        return sonoff, sonoff_ip
-
     def power_on(self, sleep=1):
         self.gpio_set(self.GPIO_POWER, "low", sleep)
         time.sleep(sleep)
@@ -125,10 +108,21 @@ def reset(self, sleep=1):
         time.sleep(sleep)
 
     def relay_get(self):
-        return self.gpio_get(self.GPIO_RELAY)
-
-    def relay_set(self, state):
-        self.gpio_set(self.GPIO_RELAY, state)
+        gpio_state = self.gpio_get(self.GPIO_RELAY)
+        relay_state = None
+        if gpio_state == "high":
+            relay_state = "on"
+        if gpio_state == "low":
+            relay_state = "off"
+        return relay_state
+
+    def relay_set(self, relay_state):
+        gpio_state = None
+        if relay_state == "on":
+            gpio_state = "high"
+        if relay_state == "off":
+            gpio_state = "low"
+        self.gpio_set(self.GPIO_RELAY, gpio_state)
 
     def reset_cmos(self):
         self.gpio_set(self.GPIO_CMOS, "low")
@@ -163,9 +157,9 @@ def pwr_ctrl_on(self):
             if state != "ON":
                 raise Exception("Failed to power control ON")
         elif self.dut_data["pwr_ctrl"]["relay"] is True:
-            self.relay_set("high")
+            self.relay_set("on")
             state = self.relay_get()
-            if state != "high":
+            if state != "on":
                 raise Exception("Failed to power control ON")
         time.sleep(5)
 
@@ -176,9 +170,9 @@ def pwr_ctrl_off(self):
             if state != "OFF":
                 raise Exception("Failed to power control OFF")
         elif self.dut_data["pwr_ctrl"]["relay"] is True:
-            self.relay_set("low")
+            self.relay_set("off")
             state = self.relay_get()
-            if state != "low":
+            if state != "off":
                 raise Exception("Failed to power control OFF")
         time.sleep(2)
 
@@ -318,6 +312,3 @@ class UnsupportedDUTModel(Exception):
 class SPIWrongVoltage(Exception):
     pass
 
-
-class SonoffNotFound(Exception):
-    pass

osfv_cli/osfv_cli/snipeit_api.py → osfv_cli/src/osfv/libs/snipeit_api.py

@@ -117,6 +117,29 @@ def get_sonoff_ip_by_rte_ip(self, rte_ip):
         # No asset found with matching RTE IP
         return None
 
+    def get_pikvm_ip_by_rte_ip(self, rte_ip):
+        # Retrieve all assets
+        all_assets = self.get_all_assets()
+
+        # Search for asset with matching RTE IP
+        for asset in all_assets:
+            custom_fields = asset.get("custom_fields", {})
+            if custom_fields:
+                rte_ip_field = next(
+                    (
+                        field_data["value"]
+                        for field_name, field_data in custom_fields.items()
+                        if field_name == "RTE IP"
+                    ),
+                    None,
+                )
+                if rte_ip_field == rte_ip:
+                    if custom_fields["PiKVM IP"]:
+                        return custom_fields["PiKVM IP"]["value"]
+
+        # No asset found with matching PiKVM IP
+        return None
+
     # Check out an asset
     def check_out_asset(self, asset_id):
         status, asset_data = self.get_asset(asset_id)

osfv_cli/osfv_cli/sonoff_api.py → osfv_cli/src/osfv/libs/sonoff_api.py

@@ -21,14 +21,14 @@ def _post_request(self, endpoint):
         return response.status_code
 
     def turn_on(self):
-        endpoint = "/switch/sonoff_s20_relay/turn_on"
+        endpoint = "/cm?cmnd=Power%20On"
         return self._post_request(endpoint)
 
     def turn_off(self):
-        endpoint = "/switch/sonoff_s20_relay/turn_off"
+        endpoint = "/cm?cmnd=Power%20off"
         return self._post_request(endpoint)
 
     def get_state(self):
-        endpoint = "/switch/sonoff_s20_relay"
+        endpoint = "/cm?cmnd=Power"
         response = self._get_request(endpoint)
-        return response.get("state")
+        return response.get("POWER")

osfv_cli/src/osfv/libs/utils.py

@@ -0,0 +1,25 @@
+
+from osfv.libs.sonoff_api import SonoffDevice
+
+def init_sonoff(init_sonoff_ip, rte_ip, snipeit_api=None):
+    sonoff_ip = ""
+    sonoff = None
+    if not snipeit_api:
+        if not init_sonoff_ip:
+            raise TypeError(
+                f"Expected a value for 'sonoff_ip', but got None"
+            )
+        sonoff_ip = init_sonoff_ip
+    else:
+        sonoff_ip = snipeit_api.get_sonoff_ip_by_rte_ip(rte_ip)
+        if not sonoff_ip:
+            raise SonoffNotFound(
+                exit(
+                    f"Sonoff IP not found in SnipeIT for RTE: {rte_ip}")
+            )
+    sonoff = SonoffDevice(sonoff_ip)
+    return sonoff, sonoff_ip
+
+class SonoffNotFound(Exception):
+    pass
+