RFLink integration incomplet?

Entraide Home Assistant
1

Bonjour
quelqu’un sait il ou se trouve la method/fonction create_rflink_connection() qui n’est dans aucun des sources livrés de l’integration RFLink. pourtant l’appel est bien dans le fichier init.

merci

Ma configuration

System Health

version: core-2021.9.7
installation_type: Home Assistant Core
dev: false
hassio: false
docker: false
user: homeassistant
virtualenv: true
python_version: 3.8.0
os_name: Linux
os_version: 5.10.60-v7l+
arch: armv7l
timezone: Europe/Paris

logged_in: false
can_reach_cert_server: ok
can_reach_cloud_auth: ok
can_reach_cloud: ok

dashboards: 1
resources: 0
views: 2
mode: storage

2

Salut,

C’est pas une fonction mais un constructeur d’un module …
from rflink.protocol import create_rflink_connection
Les sources sont là

3

Merci pour ton aide si rapide Poulpy.
Je connais ces sources et je n’y trouve pas de class create_rflink_connection ni son constructeur.
comme je suis un C/C++ developpeur et non un Python dev J’ai peut etre raté qq chose.
Si tu peux me guider je suis preneur.
En fait je cherche a comprendre comment la ligne 274 du init recupere le protocol (qui semble etre une class) et principalement comment son encodés les messages a destination du RFLink target.

Merci de ton aide.

4

Les sources du package RFLINK sont là

Dans le protocol.py , tu trouvera ça

def create_rflink_connection(
    port: int,
    host: str,
    baud: int = 57600,
    keepalive: Optional[int] = None,
    protocol: Type[ProtocolBase] = RflinkProtocol,
    packet_callback: Optional[Callable[[PacketType], None]] = None,
    event_callback: Optional[Callable[[PacketType], None]] = None,
    disconnect_callback: Optional[Callable[[Optional[Exception]], None]] = None,
    ignore: Optional[Sequence[str]] = None,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> "Coroutine[Any, Any, Tuple[asyncio.BaseTransport, ProtocolBase]]":
    """Create Rflink manager class, returns transport coroutine."""
    ...


@overload
def create_rflink_connection(
    port: str,
    host: None = None,
    baud: int = 57600,
    keepalive: None = None,
    protocol: Type[ProtocolBase] = RflinkProtocol,
    packet_callback: Optional[Callable[[PacketType], None]] = None,
    event_callback: Optional[Callable[[PacketType], None]] = None,
    disconnect_callback: Optional[Callable[[Optional[Exception]], None]] = None,
    ignore: Optional[Sequence[str]] = None,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> "Coroutine[Any, Any, Tuple[asyncio.BaseTransport, ProtocolBase]]":
    """Create Rflink manager class, returns transport coroutine."""
    ...


def create_rflink_connection(
    port: Union[None, str, int] = None,
    host: Optional[str] = None,
    baud: int = 57600,
    keepalive: Optional[int] = None,
    protocol: Type[ProtocolBase] = RflinkProtocol,
    packet_callback: Optional[Callable[[PacketType], None]] = None,
    event_callback: Optional[Callable[[PacketType], None]] = None,
    disconnect_callback: Optional[Callable[[Optional[Exception]], None]] = None,
    ignore: Optional[Sequence[str]] = None,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> "Coroutine[Any, Any, Tuple[asyncio.BaseTransport, ProtocolBase]]":
    """Create Rflink manager class, returns transport coroutine."""
    if loop is None:
        loop = asyncio.get_event_loop()
    # use default protocol if not specified
    protocol_factory = partial(
        protocol,
        loop=loop,
        packet_callback=packet_callback,
        event_callback=event_callback,
        disconnect_callback=disconnect_callback,
        ignore=ignore if ignore else [],
        keepalive=keepalive,
    )

    # setup serial connection if no transport specified
    if host:
        conn = loop.create_connection(protocol_factory, host, cast(int, port))
    else:
        conn = create_serial_connection(loop, protocol_factory, port, baud)

    return conn  # type: ignore

Il faut continuer à remonter la pile d’appels, là tu as 3 primitives différentes en fonction des paramétres