Skip to content

Driver Development

Jump to bottom
Chris Nighswonger edited this page Mar 25, 2026 · 1 revision

Driver Development

This page covers writing, registering, and testing hardware drivers for Kanfei. For general contribution workflow see Development and Contributing. For the protocol roadmap and manufacturer templates see docs/DRIVER_ROADMAP.md in the repo.

StationDriver Interface

All drivers implement the StationDriver abstract base class in backend/app/protocol/base.py.

Abstract Methods

Method Signature Purpose
connect async connect() -> None Open connection, detect hardware, perform initial setup
disconnect async disconnect() -> None Cleanly shut down the connection
poll async poll() -> Optional[SensorSnapshot] Read current sensor values; return None on failure
detect_hardware async detect_hardware() -> HardwareInfo Identify connected station model and capabilities

Abstract Properties

Property Type Purpose
connected bool Whether the driver currently has an active connection
station_name str Human-readable name of the connected station model
capabilities set[str] Set of capability strings this driver supports

Optional Override

  • request_stop() -> None -- Signal the driver to abort any blocking I/O (default is no-op)

SensorSnapshot

SensorSnapshot is a frozen data class that represents a single reading. All values must be in SI units. The driver is responsible for converting from native hardware formats before returning.

Field Unit Type Notes
inside_temp °C Optional[float]
outside_temp °C Optional[float]
inside_humidity % (0-100) Optional[int]
outside_humidity % (0-100) Optional[int]
wind_speed m/s Optional[float]
wind_direction degrees (0-359) Optional[int]
wind_gust m/s Optional[float]
barometer hPa Optional[float] Sea-level corrected
rain_rate mm/hr Optional[float]
rain_daily mm Optional[float] Since midnight
rain_yearly mm Optional[float] Since Jan 1
solar_radiation W/m² Optional[int]
uv_index index Optional[float]
soil_temp °C Optional[float]
soil_moisture centibars Optional[int]
leaf_wetness 0-15 Optional[int]
et_daily mm Optional[float] Evapotranspiration
extra -- Optional[dict] Vendor-specific fields

Fields should be None when the sensor is absent or the reading is invalid.

Capabilities

Drivers declare supported features via the capabilities property. The UI and services check these before offering hardware-specific operations.

Constant String Description
CAP_ARCHIVE_SYNC archive_sync Retrieve historical records from station memory
CAP_CALIBRATION_RW calibration_rw Read/write sensor calibration offsets
CAP_CLOCK_SYNC clock_sync Set the station's internal clock
CAP_RAIN_RESET rain_reset Clear rain accumulators
CAP_HILOWS hilows Retrieve daily/monthly/yearly hi/low records

Drivers that don't support a feature simply omit it from the set. The UI adapts automatically -- unsupported actions are hidden rather than disabled.

HardwareInfo

Returned from detect_hardware():

@dataclass(frozen=True)
class HardwareInfo:
    name: str               # Human-readable model name
    model_code: int         # Numeric station type identifier
    capabilities: set[str]  # Feature flags

Driver Factory

The logger daemon instantiates drivers via _create_driver() in backend/logger_main.py, keyed by the station_driver_type config value:

Config value Driver class Connection
legacy LinkDriver RS-232 serial
vantage VantageDriver RS-232/USB serial
weatherlink_ip WeatherLinkIPDriver TCP (port 22222)
weatherlink_live WeatherLinkLiveDriver HTTP
ecowitt EcowittDriver TCP LAN API (port 45000)
tempest TempestDriver UDP broadcast (port 50222)
ambient AmbientDriver HTTP push (configurable port)

Current Drivers

Driver Manufacturer Protocol Async Model Capabilities
LinkDriver Davis (legacy) Serial run_in_executor + RLock archive_sync, calibration_rw, clock_sync, rain_reset, hilows
VantageDriver Davis Vantage Serial run_in_executor + RLock archive_sync, clock_sync, rain_reset + conditional calibration_rw, hilows
WeatherLinkIPDriver Davis TCP wrapper over LinkDriver Delegates to LinkDriver Delegates to LinkDriver
WeatherLinkLiveDriver Davis HTTP/JSON Fully async (none)
EcowittDriver Ecowitt/Fine Offset TCP binary LAN API Fully async (none)
TempestDriver WeatherFlow UDP broadcast Fully async (none)
AmbientDriver Ambient Weather HTTP push receiver Fully async (none)

Writing a New Driver

Step-by-step

  1. Create the driver module under backend/app/protocol/<manufacturer>/. At minimum you need a driver.py containing a class that inherits from StationDriver.

  2. Implement all abstract methods and properties -- connect, disconnect, poll, detect_hardware, connected, station_name, capabilities.

  3. Return SensorSnapshot from poll() with all values converted to SI units. Leave fields None for sensors the hardware doesn't have.

  4. Declare capabilities -- only include capabilities the hardware actually supports.

  5. Register with the factory -- add a case to _create_driver() in backend/logger_main.py mapping a config key to your driver class.

  6. Add test fixtures -- see Adding Driver Tests below.

  7. Follow the hardware test plan in docs/HARDWARE_TEST_PLAN.md for field validation.

Design guidelines

  • Prefer fully async using asyncio directly. Only use run_in_executor + threading locks if the underlying protocol library is synchronous (e.g. serial I/O).
  • Parse in a separate module -- put raw-data-to-SensorSnapshot conversion in a sensors.py (or similar) alongside the driver. This keeps the parsing logic independently testable without mocking hardware connections.
  • Handle staleness -- for passive drivers (HTTP push, UDP listener), track _last_data_time and return None from poll() if data is older than a staleness threshold.
  • Unit conversion belongs in the driver -- everything downstream assumes SI. Never pass raw hardware units past the driver boundary.

Adding Driver Tests

Driver tests live in tests/backend/test_drivers.py and are driven entirely by the JSON fixture file tests/backend/driver_fixtures.json. The test framework is parametrized -- pytest discovers every fixture in the JSON and runs it through the appropriate driver parser automatically.

How it works

  1. driver_fixtures.json contains named test cases, each specifying raw hardware input and expected SI output.
  2. test_drivers.py maps each fixture's "driver" field to a runner function that calls the driver's parser.
  3. test_driver_output_si compares every field in "expected" against the SensorSnapshot the parser returns, with a tolerance of ±0.15 for floats and exact match for integers.
  4. Two guard tests ensure correctness of the fixture file itself:
    • test_all_fixtures_have_runners -- every fixture must map to a known driver runner
    • test_snapshot_fields_are_valid -- every expected field must exist on SensorSnapshot

Adding a test for an existing driver

If a runner already exists for your driver type, you only need to add a JSON entry. No code changes required.

Add an entry to driver_fixtures.json:

"my_new_test_case": {
  "driver": "ecowitt",
  "description": "Short description of what this case covers",
  "input_format": "livedata_markers",
  "raw_markers": {
    "0x01": 225,
    "0x02": 318
  },
  "expected": {
    "inside_temp": 22.5,
    "outside_temp": 31.8
  }
}

The "driver" value must match a key in _DRIVER_RUNNERS. The raw input key name and structure vary by driver type:

Driver Input key Format
ecowitt raw_markers Dict of hex string keys to numeric values
tempest raw_obs Array of observation values
tempest_air raw_obs Array (legacy Air sensor)
tempest_sky raw_obs Array (legacy Sky sensor)
ambient params Dict of HTTP parameter strings
weatherlink_live raw_json WLL JSON response structure
vantage loop_data Dict of LOOP packet field names to raw values

Optional keys depending on driver:

Key Used by Purpose
rain_state tempest, tempest_sky, ambient Pre-set rain accumulations (rain_daily_mm, rain_yearly_mm, rain_rate_mm_hr)
loop2_data vantage LOOP2 packet fields (overrides LOOP values when present)
rain_click_inches vantage Rain collector click size (default 0.01)

Adding a test for a new driver

If your driver type doesn't have a runner yet, you need two things:

1. Add a runner function in test_drivers.py:

def _run_mydriver(fixture: dict) -> SensorSnapshot:
    """Feed raw data through my driver's parser."""
    from app.protocol.mydriver.sensors import parse_raw

    return parse_raw(fixture["raw_data"])

2. Register it in the _DRIVER_RUNNERS dict:

_DRIVER_RUNNERS = {
    # ... existing runners ...
    "mydriver": _run_mydriver,
}

3. Add fixtures to driver_fixtures.json as described above, using "driver": "mydriver".

Then run:

python station.py test

What to cover in fixtures

Aim for at least:

  • A basic reading with all primary sensor fields populated
  • A below-freezing / edge-case reading (negative temps, zero rain, calm winds)
  • Any format variations the driver supports (e.g. Vantage LOOP-only vs LOOP+LOOP2, Ambient Wunderground vs Ecowitt format)

Field Validation

After unit tests pass, new drivers should go through the hardware test plan (docs/HARDWARE_TEST_PLAN.md) which covers:

  • Setup and detection (wizard completion, /api/station connected, correct station name)
  • Live data validity (plausible temps, humidity, barometer, wind, rain)
  • 30-minute stability run (no disconnect loops, no timeout/CRC spam, WebSocket updates continue)
  • Station controls (reconnect, config read/write, clock sync -- only if capabilities declared)
  • Regression safety (history, forecast, astronomy pages still work)

Hardware validation issues are tracked on the kanfei issue tracker.

Related Pages

Clone this wiki locally