Skip to content

Ports

Hexagonal-architecture port definitions — the contracts that adapters must implement.

gas2mqtt.ports

Hardware adapter ports for gas2mqtt.

Defines Protocol classes for hardware interfaces, following the Ports & Adapters (Hexagonal Architecture) pattern. Production code depends only on these protocols — concrete adapters are injected at runtime by cosalette's adapter registry.

MagneticReading dataclass 

MagneticReading(
    bx: int, by: int, bz: int, temperature_raw: int
)

A single reading from the magnetometer.

Attributes:

Name Type Description
bx int

Magnetic field strength on X axis (raw sensor units).
by int

Magnetic field strength on Y axis (raw sensor units).
bz int

Magnetic field strength on Z axis (raw sensor units).
temperature_raw int

Raw temperature sensor value.

MagnetometerPort

Bases: Protocol

Port for reading a 3-axis magnetometer with temperature sensor.

Implementations must provide magnetic field readings (bx, by, bz) and raw temperature values. The QMC5883L adapter reads from I2C; the fake adapter returns configurable values for testing/dry-run.

read 

read() -> MagneticReading

Read magnetic field and temperature from the sensor.

Pre-condition

initialize() must have been called first.

Returns:

Type Description
MagneticReading

MagneticReading with bx, by, bz, and temperature_raw values.

Raises:

Type Description
IOError

If the I2C bus communication fails.
Source code in packages/src/gas2mqtt/ports.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
def read(self) -> MagneticReading:
    """Read magnetic field and temperature from the sensor.

    Pre-condition:
        ``initialize()`` must have been called first.

    Returns:
        MagneticReading with bx, by, bz, and temperature_raw values.

    Raises:
        IOError: If the I2C bus communication fails.
    """
    ...

initialize 

initialize() -> None

Initialize the sensor with configuration registers.

Must be called before the first read(). Called automatically by __aenter__ during adapter lifecycle entry.

Source code in packages/src/gas2mqtt/ports.py 
55
56
57
58
59
60
61
def initialize(self) -> None:
    """Initialize the sensor with configuration registers.

    Must be called before the first read(). Called automatically
    by ``__aenter__`` during adapter lifecycle entry.
    """
    ...

close 

close() -> None

Release hardware resources (close I2C bus).

Called automatically by __aexit__ during adapter lifecycle teardown.

Source code in packages/src/gas2mqtt/ports.py 
63
64
65
66
67
68
69
def close(self) -> None:
    """Release hardware resources (close I2C bus).

    Called automatically by ``__aexit__`` during adapter
    lifecycle teardown.
    """
    ...

__aenter__ async 

__aenter__() -> Self

Enter async context: initialize the sensor.

Enables cosalette 0.1.5 adapter lifecycle management via AsyncExitStack.

Source code in packages/src/gas2mqtt/ports.py 
71
72
73
74
75
76
77
async def __aenter__(self) -> Self:
    """Enter async context: initialize the sensor.

    Enables cosalette 0.1.5 adapter lifecycle management via
    ``AsyncExitStack``.
    """
    ...

__aexit__ async 

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit async context: release hardware resources.

Source code in packages/src/gas2mqtt/ports.py 
79
80
81
82
83
84
85
86
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit async context: release hardware resources."""
    ...