Merge pull request #47 from CoMPaTech/improvements

Improvements

Author: CoMPaTech

.github/workflows/merge.yml

@@ -4,40 +4,44 @@ env:
   CACHE_VERSION: 1
   DEFAULT_PYTHON: "3.13"
 
-# Only run on merges
 on:
   pull_request:
     types: closed
     branches:
       - main
 
 jobs:
-  publishing:
-    name: Build and publish Python 🐍 distributions 📦 to PyPI
+  determine_version:
+    name: Determine Package Version
     runs-on: ubuntu-latest
-    environment: pypi
+    outputs:
+      package_version: ${{ steps.get_version.outputs.package_version }}
+      should_publish: ${{ steps.scheck_pypi.outputs.should_publish }}
     permissions:
-      contents: read     # Required by actions/checkout
-      id-token: write    # Needed for OIDC-based Trusted Publishing
-    # Only trigger on merges, not just closes
+      contents: read
     if: github.event.pull_request.merged == true
     steps:
       - name: Check out committed code
         uses: actions/checkout@v4
-      - name: Prepare uv
-        run: |
-          pip install uv
-          uv venv --seed venv
-          . venv/bin/activate
-          uv pip install toml
-      - name: Check for existing package on PyPI
-        id: check_package
+      - name: Get Package Version from pyproject.toml
+        id: get_version
         run: |
-          . venv/bin/activate
+          # Install toml to parse pyproject.toml
+          pip install toml
           PACKAGE_VERSION=$(python -c "import toml; print(toml.load('pyproject.toml')['project']['version'])")
           PACKAGE_NAME=$(python -c "import toml; print(toml.load('pyproject.toml')['project']['name'])")
 
-          echo "Checking for package: $PACKAGE_NAME==$PACKAGE_VERSION"
+          # Set outputs for the next jobs
+          echo "package_version=$PACKAGE_VERSION" >> "$GITHUB_OUTPUT"
+          echo "package_name=$PACKAGE_NAME" >> "$GITHUB_OUTPUT"
+          echo "Package name and version: $PACKAGE_NAME==$PACKAGE_VERSION"
+
+      - name: Check for existing package on PyPI
+        id: scheck_pypi
+        run: |
+          # Using the package name and version from the previous step
+          PACKAGE_VERSION=${{ steps.get_version.outputs.package_version }}
+          PACKAGE_NAME=${{ steps.get_version.outputs.package_name }}
 
           if curl -s "https://pypi.org/pypi/$PACKAGE_NAME/json" | jq -r '.releases | keys[]' | grep -q "^$PACKAGE_VERSION$"; then
             echo "Package version already exists. Skipping upload."
@@ -46,13 +50,68 @@ jobs:
             echo "Package version does not exist. Proceeding with upload."
             echo "should_publish=true" >> $GITHUB_OUTPUT
           fi
+
+  publishing:
+    name: Build and publish Python 🐍 distributions 📦 to PyPI
+    runs-on: ubuntu-latest
+    needs: determine_version
+    if: needs.determine_version.outputs.should_publish == 'true'
+    environment: pypi
+    permissions:
+      contents: read
+      id-token: write
+    steps:
+      - name: Check out committed code
+        uses: actions/checkout@v4
+      - name: Prepare uv
+        run: |
+          pip install uv
+          uv venv --seed venv
+          . venv/bin/activate
       - name: Build
-        if: steps.check_package.outputs.should_publish == 'true'
         run: |
           . venv/bin/activate
           uv build
       - name: Publish distribution 📦 to PyPI
-        if: steps.check_package.outputs.should_publish == 'true'
         run: |
           . venv/bin/activate
           uv publish
+
+  create_tag:
+    name: Create Git Tag for Release
+    runs-on: ubuntu-latest
+    needs: [determine_version, publishing]
+    if: always() && needs.publishing.result == 'success' && needs.determine_version.outputs.should_publish == 'true'
+    permissions:
+      contents: write
+    steps:
+      - name: Check out committed code
+        uses: actions/checkout@v4
+      - name: Create release tag
+        id: tag_release
+        uses: actions/github-script@v6
+        with:
+          script: |
+            const version = process.env.PACKAGE_VERSION;
+            const tagName = `v${version}`;
+
+            console.log(`Attempting to create tag: ${tagName}`);
+
+            try {
+              await github.rest.git.createRef({
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                ref: `refs/tags/${tagName}`,
+                sha: context.sha,
+              });
+              console.log(`Successfully created tag: ${tagName}`);
+            } catch (error) {
+              console.error(`Failed to create tag: ${error.message}`);
+              if (error.status === 422) {
+                console.log(`Tag ${tagName} already exists. Skipping creation.`);
+              } else {
+                throw error;
+              }
+            }
+        env:
+          PACKAGE_VERSION: ${{ needs.determine_version.outputs.package_version }}

.gitignore

@@ -11,3 +11,4 @@ tests/__pycache__
 .coverage
 tmp
 todo
+.DS_Store

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.2.5] - 2025-08-05
+
+### Added
+
+- Added booleans determining station/accesspoint and PTP/PTMP in derived subclass
+
 ## [0.2.4] - 2025-08-03
 
 ### Added

CONTRIBUTE.md

@@ -0,0 +1,37 @@
+# Contributing
+
+It would be very helpful if you would share your configuration data to this project. This way we can make sure that the data returned by your airOS devices is processed correctly.
+
+## Current fixtures
+
+We currently have data on
+
+- Nanostation 5AC (LOCO5AC) - PTP - both AP and Station output of `/status.cgi` present (by @CoMPaTech)
+
+## Secure your data
+
+The best way to share your data is to remove any data that is not necessary for processing. To ensure you don't share any data by accident please follow the following
+
+- Log in to your device
+- Note down the following options set
+  - Is it a station or access point
+  - Is it PTP (or PTMP)
+  - Channel width in Mhz
+- Manually update the url to `/status.cgi`, e.g. `https://192.168.1.10/status.cgi`
+- Store the output (for instance in an editor, even notepad would be fine) for processing
+
+**NOTE**: when redacting, redact in a meaningful way by changing parameters, don't put text in number fields or vice-versa!
+
+- First and foremost: find any `lat` and `lon` information and redact these but keep them as floats. (I.e. a value with decimals)!
+  - There are potentially multiple of these (so keep searching)
+  - If you are unsure, apply them as `(...)"lat":52.379894,"lon":4.901608,(...)` to point to Amsterdam
+- Redact your IP addresses, especially public IPs (if present),
+  - Search/replace your range, say your AP is at `192.168.1.10` then search for `192.168.1.` and replace with `127.0.0.`
+  - Set them to `127.0.0.xxx` leaving the actual last octet what it was, just so devices are still different from **and** coheren to each other.
+- Redact your SSID, just name it WirelessABC (as long as it's still coherent)
+- Make sure your `hostname`s don't disclose unwanted information
+- You may redact your MAC addresses (hwaddr, mac, etc) just make sure they are still valid (`00:11:22:33:44:55`) and coherent (so make sure local and remote(s) still differ)
+
+### Examples
+
+See `fixtures/userdata` for examples of shared information

airos/airos8.py

@@ -196,6 +196,30 @@ def derived_data(
         self, response: dict[str, Any] | None = None
     ) -> dict[str, Any] | None:
         """Add derived data to the device response."""
+        derived = {
+            "station": False,
+            "access_point": False,
+            "ptp": False,
+            "ptmp": False,
+        }
+
+        # Access Point / Station vs PTP/PtMP
+        wireless_mode = response.get("wireless", {}).get("mode", "")
+        match wireless_mode:
+            case "ap-ptmp":
+                derived["access_point"] = True
+                derived["ptmp"] = True
+            case "sta-ptmp":
+                derived["station"] = True
+                derived["ptmp"] = True
+            case "ap-ptp":
+                derived["access_point"] = True
+                derived["ptp"] = True
+            case "sta-ptp":
+                derived["station"] = True
+                derived["ptp"] = True
+
+        # INTERFACES
         addresses = {}
         interface_order = ["br0", "eth0", "ath0"]
 
@@ -209,19 +233,18 @@ def derived_data(
             if interface["enabled"]:  # Only consider if enabled
                 addresses[interface["ifname"]] = interface["hwaddr"]
 
+        # Fallback take fist alternate interface found
+        derived["mac"] = interfaces[0]["hwaddr"]
+        derived["mac_interface"] = interfaces[0]["ifname"]
+
         for interface in interface_order:
             if interface in addresses:
-                response["derived"] = {
-                    "mac": addresses[interface],
-                    "mac_interface": interface,
-                }
-                return response
+                derived["mac"] = addresses[interface]
+                derived["mac_interface"] = interface
+                break
+
+        response["derived"] = derived
 
-        # Fallback take fist alternate interface found
-        response["derived"] = {
-            "mac": interfaces[0]["hwaddr"],
-            "mac_interface": interfaces[0]["ifname"],
-        }
         return response
 
     async def status(self) -> AirOSData:

airos/data.py

@@ -44,8 +44,9 @@ class IeeeMode(Enum):
 class WirelessMode(Enum):
     """Enum definition."""
 
-    PTP_ACCESSPOINT = "ap-ptp"
     PTMP_ACCESSPOINT = "ap-ptmp"
+    PTMP_STATION = "sta-ptmp"
+    PTP_ACCESSPOINT = "ap-ptp"
     PTP_STATION = "sta-ptp"
     # More to be added when known
 
@@ -437,6 +438,14 @@ class Derived:
     mac: str  # Base device MAC address (i.e. eth0)
     mac_interface: str  # Interface derived from
 
+    # Split for WirelessMode
+    station: bool
+    access_point: bool
+
+    # Split for WirelessMode
+    ptp: bool
+    ptmp: bool
+
 
 @dataclass
 class AirOS8Data(DataClassDictMixin):

airos/discovery.py

@@ -175,14 +175,13 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
                     offset += 2
 
                     if tlv_length > (len(data) - offset):
-                        log = f"TLV type {tlv_type:#x} length {tlv_length} exceeds remaining data "
-                        _LOGGER.warning(log)
-                        log = f"({len(data) - offset} bytes left). Packet malformed. "
-                        _LOGGER.warning(log)
-                        log = f"Data from TLV start: {data[offset - 3 :].hex()}"
+                        log = (
+                            f"TLV type {tlv_type:#x} length {tlv_length} exceeds remaining data "
+                            f"({len(data) - offset} bytes left). Packet malformed. "
+                            f"Data from TLV start: {data[offset - 3 :].hex()}"
+                        )
                         _LOGGER.warning(log)
-                        log = f"Malformed packet: {log}"
-                        raise AirOSEndpointError(log)
+                        raise AirOSEndpointError(f"Malformed packet: {log}")
 
                     tlv_value: bytes = data[offset : offset + tlv_length]
 
@@ -195,6 +194,7 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
                         else:
                             log = f"Unexpected length for 0x02 TLV (MAC+IP). Expected 10, got {tlv_length}. Value: {tlv_value.hex()}"
                             _LOGGER.warning(log)
+                            raise AirOSEndpointError(f"Malformed packet: {log}")
 
                     elif tlv_type == 0x03:
                         parsed_info["firmware_version"] = tlv_value.decode(
@@ -213,6 +213,7 @@ def parse_airos_packet(self, data: bytes, host_ip: str) -> dict[str, Any] | None
                         else:
                             log = f"Unexpected length for Uptime (Type 0x0A): {tlv_length}. Value: {tlv_value.hex()}"
                             _LOGGER.warning(log)
+                            raise AirOSEndpointError(f"Malformed packet: {log}")
 
                     elif tlv_type == 0x0B:
                         parsed_info["hostname"] = tlv_value.decode(