chore+feat(home-assistant): update to pyo3 0.27 and update extraction errors, switch out SmolStr for Arc<str>, tighten up light service calls and implement some for notify, start implementing units of measurement like for power
This commit is contained in:
@@ -2,7 +2,7 @@ use super::{
|
||||
event::{context::context::Context, specific::state_changed},
|
||||
home_assistant::HomeAssistant,
|
||||
};
|
||||
use crate::entity_id::EntityId;
|
||||
use crate::{entity_id::EntityId, home_assistant::GetStatesError, state_machine::GetStateError};
|
||||
use chrono::{DateTime, Utc};
|
||||
use emitter_and_signal::signal::Signal;
|
||||
use once_cell::sync::OnceCell;
|
||||
@@ -10,6 +10,7 @@ use pyo3::{
|
||||
prelude::*,
|
||||
types::{PyCFunction, PyDict, PyTuple},
|
||||
};
|
||||
use snafu::{ResultExt, Snafu};
|
||||
use std::{future::Future, sync::Arc};
|
||||
use tokio::{select, sync::mpsc};
|
||||
|
||||
@@ -24,48 +25,86 @@ pub struct StateObject<State, Attributes, ContextEvent> {
|
||||
pub context: Context<ContextEvent>,
|
||||
}
|
||||
|
||||
pub type ExtractStateObjectError<'a, 'py, State, Attributes, ContextEvent> =
|
||||
<StateObject<State, Attributes, ContextEvent> as FromPyObject<'a, 'py>>::Error;
|
||||
|
||||
#[derive(Debug, Snafu)]
|
||||
pub enum CreateSignalError {
|
||||
/// couldn't get the state machine from the Home Assistant object
|
||||
GetStatesError { source: GetStatesError },
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Snafu)]
|
||||
pub enum StateObjectSignalError<ExtractStateObjectError: 'static + snafu::Error> {
|
||||
/// couldn't get the state object directly from the state machine
|
||||
GetFromStateMachine {
|
||||
source: GetStateError<ExtractStateObjectError>,
|
||||
},
|
||||
|
||||
/// couldn't get the state object from the new state event
|
||||
GetFromNewStateEvent { source: Arc<PyErr> },
|
||||
}
|
||||
|
||||
impl<
|
||||
State: Send + Sync + 'static + for<'py> FromPyObject<'py>,
|
||||
Attributes: Send + Sync + 'static + for<'py> FromPyObject<'py>,
|
||||
ContextEvent: Send + Sync + 'static + for<'py> FromPyObject<'py>,
|
||||
State: Send + Sync + 'static + for<'a, 'py> FromPyObject<'a, 'py>,
|
||||
Attributes: Send + Sync + 'static + for<'a, 'py> FromPyObject<'a, 'py>,
|
||||
ContextEvent: Send + Sync + 'static + for<'a, 'py> FromPyObject<'a, 'py>,
|
||||
> StateObject<State, Attributes, ContextEvent>
|
||||
{
|
||||
pub fn store(
|
||||
py: Python<'_>,
|
||||
home_assistant: &HomeAssistant,
|
||||
pub fn signal<'a, 'py>(
|
||||
py: Python<'py>,
|
||||
home_assistant: &'py HomeAssistant,
|
||||
entity_id: EntityId,
|
||||
) -> PyResult<(
|
||||
Signal<Option<Arc<Self>>>,
|
||||
impl Future<Output = Result<(), emitter_and_signal::signal::JoinError>>,
|
||||
)> {
|
||||
let state_machine = home_assistant.states(py)?;
|
||||
let current = state_machine.get(py, entity_id.clone())?;
|
||||
) -> Result<
|
||||
(
|
||||
Signal<
|
||||
Option<
|
||||
Arc<
|
||||
Result<
|
||||
Self,
|
||||
StateObjectSignalError<<Self as FromPyObject<'a, 'py>>::Error>,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
impl Future<Output = Result<(), emitter_and_signal::signal::JoinError>>,
|
||||
),
|
||||
CreateSignalError,
|
||||
> {
|
||||
let state_machine = home_assistant.states(py).context(GetStatesSnafu)?;
|
||||
let current = state_machine
|
||||
.get(py, entity_id.clone())
|
||||
.context(GetFromStateMachineSnafu)
|
||||
.transpose();
|
||||
|
||||
let py_home_assistant = home_assistant.into_pyobject(py)?.unbind();
|
||||
let Ok(py_home_assistant) = home_assistant.into_pyobject(py);
|
||||
let py_home_assistant = py_home_assistant.unbind();
|
||||
|
||||
let (store, task) = Signal::new(current.map(Arc::new), |mut publisher_stream| async move {
|
||||
while let Some(publisher) = publisher_stream.wait().await {
|
||||
let (new_state_sender, mut new_state_receiver) = mpsc::channel(8);
|
||||
let (signal, task) = Signal::new(
|
||||
current.map(Arc::new),
|
||||
|mut publisher_stream| async move {
|
||||
while let Some(publisher) = publisher_stream.wait().await {
|
||||
let (new_state_sender, mut new_state_receiver) = mpsc::channel(8);
|
||||
|
||||
let untrack = Python::with_gil::<_, PyResult<_>>(|py| {
|
||||
static EVENT_MODULE: OnceCell<Py<PyModule>> = OnceCell::new();
|
||||
let untrack = Python::attach::<_, PyResult<_>>(|py| {
|
||||
static EVENT_MODULE: OnceCell<Py<PyModule>> = OnceCell::new();
|
||||
|
||||
let event_module = EVENT_MODULE
|
||||
.get_or_try_init(|| {
|
||||
Result::<_, PyErr>::Ok(
|
||||
py.import("homeassistant.helpers.event")?.unbind(),
|
||||
)
|
||||
})?
|
||||
.bind(py);
|
||||
let event_module = EVENT_MODULE
|
||||
.get_or_try_init(|| {
|
||||
Result::<_, PyErr>::Ok(
|
||||
py.import("homeassistant.helpers.event")?.unbind(),
|
||||
)
|
||||
})?
|
||||
.bind(py);
|
||||
|
||||
let untrack = {
|
||||
let callback =
|
||||
let untrack = {
|
||||
let callback =
|
||||
move |args: &Bound<'_, PyTuple>,
|
||||
_kwargs: Option<&Bound<'_, PyDict>>| {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("calling the closure");
|
||||
|
||||
if let Ok((event,)) = args.extract::<(
|
||||
let new_state_res = args.extract::<(
|
||||
state_changed::Event<
|
||||
State,
|
||||
Attributes,
|
||||
@@ -75,77 +114,65 @@ impl<
|
||||
ContextEvent,
|
||||
Py<PyAny>,
|
||||
>,
|
||||
)>() {
|
||||
let new_state = event.data.new_state;
|
||||
)>().map(|event| event.0.data.new_state).map_err(Arc::new).context(GetFromNewStateEventSnafu);
|
||||
|
||||
new_state_sender.try_send(new_state_res).unwrap();
|
||||
};
|
||||
let callback = PyCFunction::new_closure(py, None, None, callback)?;
|
||||
let args = (
|
||||
py_home_assistant.clone_ref(py),
|
||||
vec![entity_id.clone()],
|
||||
callback,
|
||||
);
|
||||
event_module.call_method1("async_track_state_change_event", args)?
|
||||
};
|
||||
|
||||
let untrack = untrack.unbind();
|
||||
|
||||
Ok(untrack)
|
||||
});
|
||||
|
||||
if let Ok(untrack) = untrack {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!(
|
||||
"untrack is ok, going to wait for the next relevant event..."
|
||||
);
|
||||
loop {
|
||||
select! {
|
||||
biased;
|
||||
_ = publisher.all_unsubscribed() => {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("calling untrack");
|
||||
let res = Python::attach(|py| untrack.call0(py));
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("sending a new state"); // TODO: remove
|
||||
new_state_sender.try_send(new_state).unwrap();
|
||||
tracing::debug!(?res);
|
||||
break;
|
||||
}
|
||||
};
|
||||
let callback = PyCFunction::new_closure(py, None, None, callback)?;
|
||||
let args = (
|
||||
py_home_assistant.clone_ref(py),
|
||||
vec![entity_id.clone()],
|
||||
callback,
|
||||
);
|
||||
event_module.call_method1("async_track_state_change_event", args)?
|
||||
};
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!(?untrack, "as any");
|
||||
|
||||
let is_callable = untrack.is_callable();
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!(?is_callable);
|
||||
|
||||
// let untrack = untrack.downcast_into::<PyFunction>()?;
|
||||
// tracing::debug!(?untrack, "as downcast");
|
||||
|
||||
let untrack = untrack.unbind();
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!(?untrack, "as unbound");
|
||||
|
||||
Ok(untrack)
|
||||
});
|
||||
|
||||
if let Ok(untrack) = untrack {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("untrack is ok, going to wait for the next relevant event...");
|
||||
loop {
|
||||
select! {
|
||||
biased;
|
||||
_ = publisher.all_unsubscribed() => {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("calling untrack");
|
||||
let res = Python::with_gil(|py| untrack.call0(py));
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!(?res);
|
||||
break;
|
||||
}
|
||||
new_state = new_state_receiver.recv() => {
|
||||
match new_state {
|
||||
Some(new_state) => {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("publishing new state");
|
||||
publisher.publish(new_state.map(Arc::new))
|
||||
},
|
||||
None => {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("channel dropped");
|
||||
break
|
||||
},
|
||||
new_state_res_option = new_state_receiver.recv() => {
|
||||
match new_state_res_option {
|
||||
Some(new_state_res) => {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("publishing new state");
|
||||
publisher.publish(new_state_res.transpose().map(Arc::new));
|
||||
},
|
||||
None => {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("channel dropped");
|
||||
break
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("untrack is err");
|
||||
}
|
||||
} else {
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("untrack is err");
|
||||
}
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
Ok((store, task))
|
||||
Ok((signal, task))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user