Websocket: Unterschied zwischen den Versionen

Aktuelle Version vom 20. Januar 2024, 12:33 Uhr

FHEM kann mit Websockets kommunizieren indem DevIo genutzt wird (Ankündigung des Supports in diesem Forenbeitrag). Bisher werden Websockets nur mit Perl-Befehlen angesprochen.

Schritte:

Setzen der Parameter:
- Setzen des Kommunikationsendpunktes: $hash->{DeviceName} = "wss:echo.websocket.org:443/pfad"; Wobei die Portnummer zwingend hinzugefügt werden muss, wenn ein Pfad spezifiziert wird.
- Optional: Setzen von speziellen Headerangaben: $hash->{header}{'Sec-WebSocket-Protocol'} = 'graphql-transport-ws';
Setzen einer CallBack Funktion:
- Wenn Daten von der Websocket empfangen werden, werden diese an eine Funktion übergeben. Diese macht man mit der directReadFn bekannt: $hash->{directReadFn}
Starten der Kommunikation: DevIo_OpenDev(...);
Ist die Verbindung aufgebaut, kann man Daten mit der Funktion DevIo_SimpleWrite(...) senden. Die Beispiele zu "Tibber" und "Owntone" senden jeweils Daten beim Aufbau der Verbindung als ASCII String.

Beispiele

Tibber Live-Messdaten auslesen

Screenshot vom Tibber Websocket und weitere Tibber-Devices

Folgendes Beispiel liest Strommesswerte des Anbieters Tibber aus. Mit

set Tibber.ws start

wird die Verbindung aufgebaut und mit

set Tibber.ws stop

wieder gestoppt. Siehe auch das zugehörige Forenthema ab diesem Beitrag.

defmod Tibber.ws dummy
attr Tibber.ws userattr websocketURL homeId token myId minInterval
attr Tibber.ws alias Tibber Websocket
attr Tibber.ws event-on-change-reading .*
attr Tibber.ws eventMap /cmd connect:start/cmd disconnect:stop/
attr Tibber.ws homeId 96a14971-525a-4420-aae9-e5aedaa129ff
attr Tibber.ws icon hue_filled_plug
attr Tibber.ws minInterval 30
attr Tibber.ws myId TorxgewindeID
attr Tibber.ws readingList cmd
attr Tibber.ws setList cmd
attr Tibber.ws stateFormat payload_data_liveMeasurement_accumulatedCost payload_data_liveMeasurement_currency (payload_data_liveMeasurement_power W, Import: payload_data_liveMeasurement_accumulatedConsumption kWh, Export: payload_data_liveMeasurement_accumulatedProduction kWh)
attr Tibber.ws token 5K4MVS-OjfWhK_4yrjOlFe1F6kJXPVf7eQYggo8ebAE
attr Tibber.ws userReadings connect:cmd:.connect {\
	my $hash = $defs{$name};;\
	my $devState = DevIo_IsOpen($hash);;\
	return "Device already open" if (defined($devState));;\
	\
	# establish connection to websocket\
	# format must also include portnumber if a path is to be specified\
	$hash->{DeviceName} = AttrVal($name, "websocketURL", "wss:echo.websocket.org:443");;\
	\
	# special headers needed for Tibber, see also Developer Tools in Browser\
	$hash->{header}{'Sec-WebSocket-Protocol'} = 'graphql-transport-ws';;\
	$hash->{header}{'Host'} = 'websocket-api.tibber.com';;\
	$hash->{header}{'Origin'} = 'https://developer.tibber.com';;\
	\
	# callback function when "select()" signals data for us\
	# websocket Ping/Pongs are treated in DevIo but still call this function\
	$hash->{directReadFn} = sub () {\
		my $hash = $defs{$name};;\
		\
		# we can read without closing the DevIo, because select() signalled data\
		my $buf = DevIo_SimpleRead($hash);;\
		\
		# if read fails, close device\
		if(!defined($buf)) {\
			DevIo_CloseDev($hash);;\
			$buf = "not_connected";;\
		}\
		\
		#Log(3, "$name:$reading: websocket data: >>>$buf<<<");;\
		\
		# only update our reading if buffer is not empty and if last update is older than minInterval\
		if ($buf ne "") {\
			my $websocketDataAge = ReadingsAge($name, "websocketData", 3600);;\
			my $minInterval = AttrVal($name, "minInterval", 0);;\
			my $isNext = ($buf =~ /.*id.*type.*next.*payload.*data.*liveMeasurement.*/s);;\
			\
			readingsBeginUpdate($hash);;\
			readingsBulkUpdate($hash, "websocketData", "$buf") if ($isNext && $websocketDataAge > $minInterval);;\
			readingsBulkUpdate($hash, "websocketData", "$buf") if (!$isNext);;\
			readingsEndUpdate($hash, 1);;\
		}\
	};;\
	\
	# open DevIo websocket\
	DevIo_OpenDev($hash, 0, undef, sub(){\
		my ($hash, $error) = @_;;\
		return "$error" if ($error);;\
		\
		my $token = AttrVal($name, "token", "???");;\
		\
		DevIo_SimpleWrite($hash, '{"type":"connection_init","payload":{"token":"'.$token.'"}}', 2);;\
	});;\
	readingsBulkUpdate($hash, "websocketData", "");;\
		\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
disconnect:cmd:.disconnect {\
	my $hash = $defs{$name};;\
	RemoveInternalTimer($hash);;\
	DevIo_SimpleRead($hash);;\
	DevIo_CloseDev($hash);;\
	\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onDisconnect {\
	my $myState = ReadingsVal($name, "state", "???");;\
	my $myData = ReadingsVal($name, "websocketData", "???");;\
	return if ($myState ne "disconnected" and $myData ne "not_connected");;\
	\
	## timer callback function, called after a few seconds to initiate a reconnect\
	my $timerFunction = sub() {\
		my ($arg) = @_;;\
		my $hash = $defs{$name};;\
		my $devState = DevIo_IsOpen($hash);;\
		\
		# only re-connect if device is not connected\
		readingsSingleUpdate($hash, "cmd", "connect", 1) if (!defined($devState));;\
	};;\
	RemoveInternalTimer($name.$reading.'Timer');;\
	\
	# wait a random time before reconnect (exponential backoff TBD):\
	my $rwait = int(rand(200)) + 30;;\
	InternalTimer(gettimeofday() + $rwait, $timerFunction, $name.$reading.'Timer');;\
	\
	#set cmd to a new value, informs user and allows to retrigger when timer expires\
	my $hash = $defs{$name};;\
	readingsBulkUpdate($hash, "cmd", "reconnect attempt in $rwait seconds");;\
	\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onTimeout:websocketData:.* {\
	#re-establish websocket connection if no data received in the past ten minutes\
	#but only if our reading "cmd" was not set to the value "disconnect"\
	\
	#timeout in seconds when the connection is considered dead\
	my $timeoutTime = 600;;\
	\
	# function to execute when timeout expired\
	# defining the function here in the userReading, allows us to insert variables directly\
	my $timerFunction = sub() {\
		my ($arg) = @_;;\
		my $hash = $defs{$name};;\
		my $rCmd = ReadingsVal($name, "cmd", "???");;\
		my $age  = ReadingsAge($name, "websocketData", 0);;\
		\
		Log(3, "$name: onTimeoutTimer triggered >>$arg<<");;\
		\
		#do not do anything further if disconnect is on purpose\
		if ( $rCmd eq "disconnect" ) {\
			Log(3, "$name: cmd was set to disconnect");;\
			return;;\
		}\
		\
		# for whatever reason, we triggered to soon (80%)\
		if ( $age < $timeoutTime*0.8 ) {\
			Log(3, "$name: websocketData is not outdated");;\
			return;;\
		}\
		\
		DevIo_CloseDev($hash);;\
		Log(3, "$name: onTimeoutTimer closed DevIo...");;\
		\
		readingsSingleUpdate($hash, "cmd", "connect", 1);;\
		Log(3, "$name: onTimeoutTimer set cmd to value 'connect'");;\
	};;\
	\
	#remove/cancel previous timers, because we got fresh data and countdown starts again\
	RemoveInternalTimer($name.$reading.'Timer');;\
	\
	#set timer to expire and execute function defined above, give special arg as identifier\
	InternalTimer(gettimeofday() + $timeoutTime, $timerFunction, $name.$reading.'Timer');;\
	\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onConnectionAck:websocketData:.*connection_ack.* {\
	#websocketData contains the string "connection_ack"\
	Log(3, "$name:$reading: got connection ack");;\
	\
	# do not proceed if connection is lost\
	my $hash = $defs{$name};;\
	my $devState = DevIo_IsOpen($hash);;\
	return "Device not open" if (!defined($devState));;\
	\
	readingsBulkUpdate($hash, "cmd", "got connection ack");;\
	\
	my $homeId = AttrVal($name, "homeId", "???");;\
	my $myId = AttrVal($name, "myId", "???");;\
	\
	# build the query, do it in pieces, the comma at the end caused perl errors\
	# so we put it together in this not very elegant way\
	my $json = '{ "id":"'. $myId .'", "type":"subscribe"'.", ";;\
	$json .= '"payload":{';;\
	$json .= '"variables":{}'.", ";;\
	$json .= '"extensions":{}'.", ";;\
	$json .= '"query":"subscription { liveMeasurement( homeId: \"'.$homeId.'\" ) ';;\
	#$json .= '{ timestamp power accumulatedConsumption accumulatedCost currency minPower averagePower maxPower signalStrength }}"';;\
	$json .= '{ timestamp power lastMeterConsumption accumulatedConsumption accumulatedProduction ';;\
	$json .= 'accumulatedProductionLastHour accumulatedCost accumulatedReward currency minPower averagePower maxPower ';;\
	$json .= 'powerProduction powerReactive powerProductionReactive minPowerProduction maxPowerProduction lastMeterProduction ';;\
	$json .= 'powerFactor voltagePhase1 voltagePhase2 voltagePhase3 signalStrength }}"';;\
	$json .= '}}';;\
	\
	#send the string via websocket as ASCII\
	Log(3, "$name:$reading: sending JSON: >>>$json<<<");;\
	DevIo_SimpleWrite($hash, $json, 2);;\
		\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onNextLiveMeasurement:websocketData:.*next.*payload.*data.*liveMeasurement.* {\
	#websocketData contains next-live-measurement-data\
	my $val = ReadingsVal($name, "websocketData", "{}");;\
	my %res = %{json2nameValue($val, undef, undef, "payload_data_liveMeasurement.*")};;\
	\
	my $ret = "got values for:\n";;\
	foreach my $k (sort keys %res) {\
		$ret .= "$k\n";;\
		readingsBulkUpdate($hash, makeReadingName($k), $res{$k});;\
	}\
	return $ret;;\
}
attr Tibber.ws webCmd start:stop
attr Tibber.ws websocketURL wss:websocket-api.tibber.com:443/v1-beta/gql/subscriptions

Owntone (ehemals ForkedDaapd)

Der Musikserver Owntone kann mit einer Websocket Informationen bereitstellen. Diese kann man ebenfalls mit einem einfachen Device auswerten (siehe auch Beitrag und das zugehörige Owntone-Device im Thema ):

defmod WS dummy
attr WS userattr websocketURL
attr WS alias Owntone Websocket
attr WS devStateIcon opened:general_ok@green:stop disconnected:rc_STOP@red:start
attr WS eventMap /wert connect:start/wert disconnect:stop/
attr WS icon hue_filled_plug
attr WS readingList wert
attr WS setList wert
attr WS userReadings connect:wert:.connect {\
	my $hash = $defs{$name};;\
	my $devState = DevIo_IsOpen($hash);;\
	return "Device already open" if (defined($devState));;\
	\
	$hash->{DeviceName} = AttrVal($name, "websocketURL", "ws:echo.websocket.org:443");;\
	\
	# special headers needed for Owntone\
	# https://owntone.github.io/owntone-server/json-api/#push-notifications\
	$hash->{header}{'Sec-WebSocket-Protocol'} = 'notify';;\
	$hash->{header}{'Host'} = 'localhost:3688';;\
	$hash->{header}{'Origin'} = 'http://localhost:3688';;\
	\
	# callback function when "select" signals data for us\
	# websocket Ping/Pongs are treated in DevIo but still call this function\
	$hash->{directReadFn} = sub () {\
		my $hash = $defs{$name};;\
		readingsBeginUpdate($hash);;\
		\
		# we can read without closing the DevIo, because select signalled data\
		my $buf = DevIo_SimpleRead($hash);;\
		\
		if(!defined($buf)) {\
			DevIo_CloseDev($hash);;\
			$buf = "not connected";;\
		}\
		\
		# only update our reading if buffer is not empty\
		readingsBulkUpdate($hash, "websocketData", "$buf") if ($buf ne "");;\
		readingsEndUpdate($hash, 1);;\
	};;\
	\
	# open DevIo websocket\
	DevIo_OpenDev($hash, 0, undef, sub(){\
		my ($hash, $error) = @_;;\
		return "$error" if ($error);;\
		\
		#immediately send Owntone what we would like to be notified for (here we selected everything)\
		DevIo_SimpleWrite($hash, '{"notify":["update","database","player","options","outputs","volume","queue","spotify","lastfm","pairing"]}', 2);;\
	});;\
	\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
disconnect:wert:.disconnect {\
	my $hash = $defs{$name};;\
	RemoveInternalTimer($hash);;\
	DevIo_SimpleRead($hash);;\
	DevIo_CloseDev($hash);;\
	\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onDisconnect {\
	my $myState = ReadingsVal($name, "state", "???");;\
	return if ($myState ne "disconnected");;\
	\
	# timer callback function, called after a few seconds to initiate a reconnect\
	my $timerFunction = sub() {\
		my ($arg) = @_;;\
		my $hash = $defs{$name};;\
		my $devState = DevIo_IsOpen($hash);;\
		readingsSingleUpdate($hash, "wert", "connect", 1) if (!defined($devState));;\
	};;\
	\
	RemoveInternalTimer($name.$reading.'Timer');;\
	InternalTimer(gettimeofday() + 10, $timerFunction, $name.$reading.'Timer');;\
	\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onPlayer:websocketData:.*player.* {\
	fhem("set Owntone.device reread");;\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onOutputs:websocketData:.*outputs.* {\
	fhem("get Owntone.device outputs");;\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onVolume:websocketData:.*volume.* {\
	fhem("get Owntone.device volume");;\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
},\
onQueue:websocketData:.*queue.* {\
	fhem("get Owntone.device queue");;\
	return POSIX::strftime("%H:%M:%S",localtime(time()));;\
}
attr WS websocketURL ws:192.168.123.123:3688