use opc_da_client::{
BrowseNamespace, BrowseNodeFilter, BrowseNodeKind as ExtBrowseNodeKind, BrowseNodeToken,
BrowsePageRequest, BrowsePageToken, BrowseSessionToken,
InventoryControl as ExtInventoryControl, InventoryEvent as ExtInventoryEvent, InventoryOptions,
InventoryStream as ExtInventoryStream, OpcDaClient, OpcProvider, OpcValue as ExtOpcValue,
TagValue as ExtTagValue, WriteResult as ExtWriteResult,
};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use crate::opc::{
BrowseCapabilities, BrowseNode, BrowseNodeKind, BrowsePage, BrowseSource, InventoryCompleted,
InventoryControl, InventoryEntry, InventoryEvent, InventoryHandle, InventoryNodeKind,
InventoryProgress, InventoryStream, NamespaceOrganization, OpcClient, OpcValue, TagValue,
WriteResult,
};
#[derive(Default)]
pub struct OpcDaAdapter {
client: OpcDaClient,
sessions: Mutex<HashMap<String, (NamespaceOrganization, BrowseSource)>>,
}
#[async_trait::async_trait]
impl OpcClient for OpcDaAdapter {
async fn list_servers(&self, host: &str) -> anyhow::Result<Vec<String>> {
Ok(self.client.list_servers(host).await?)
}
async fn get_capabilities(&self, server: &str) -> anyhow::Result<BrowseCapabilities> {
let capabilities = self.client.browse_capabilities(server).await?;
let (organization, source) = map_capabilities(&capabilities);
Ok(BrowseCapabilities {
organization,
source,
supports_browse_sessions: true,
supports_search: true,
max_page_size: capabilities.max_page_size,
})
}
async fn open_browse_session(&self, server: &str) -> anyhow::Result<String> {
let capabilities = self.client.browse_capabilities(server).await?;
let (organization, source) = map_capabilities(&capabilities);
let session = self.client.open_browse_session(server).await?.to_string();
self.sessions
.lock()
.map_err(|_| anyhow::anyhow!("browse session lock poisoned"))?
.insert(session.clone(), (organization, source));
Ok(session)
}
async fn browse_page(
&self,
session_id: &str,
parent_node_key: Option<&str>,
page_token: Option<&str>,
page_size: u32,
_refresh: bool,
) -> anyhow::Result<BrowsePage> {
let session = parse_token::<BrowseSessionToken>(session_id, "browse session")?;
let parent = parent_node_key
.map(|value| parse_token::<BrowseNodeToken>(value, "browse node"))
.transpose()?;
let continuation = page_token
.map(|value| parse_token::<BrowsePageToken>(value, "browse page"))
.transpose()?;
let page = self
.client
.browse_page(
&session,
BrowsePageRequest {
parent,
filter: BrowseNodeFilter::All,
max_elements: page_size,
continuation,
},
)
.await?;
let (organization, source) = self
.sessions
.lock()
.map_err(|_| anyhow::anyhow!("browse session lock poisoned"))?
.get(session_id)
.copied()
.ok_or_else(|| anyhow::anyhow!("unknown browse session"))?;
Ok(BrowsePage {
nodes: page
.nodes
.into_iter()
.map(|node| BrowseNode {
node_key: node.token.to_string(),
display_name: node.name,
kind: map_node_kind(node.kind),
item_id: node.item_id,
})
.collect(),
next_page_token: page.continuation.map(|token| token.to_string()),
complete: page.continuation.is_none(),
organization,
source,
warning: None,
})
}
async fn close_browse_session(&self, session_id: &str) -> anyhow::Result<()> {
let session = parse_token::<BrowseSessionToken>(session_id, "browse session")?;
let result = self.client.close_browse_session(&session).await;
self.sessions
.lock()
.map_err(|_| anyhow::anyhow!("browse session lock poisoned"))?
.remove(session_id);
Ok(result?)
}
async fn start_inventory(
&self,
server: &str,
batch_size: u32,
) -> anyhow::Result<InventoryHandle> {
let stream = self
.client
.start_inventory(
server,
InventoryOptions {
batch_size,
max_entries: None,
},
)
.await?;
let control = stream.control();
Ok(InventoryHandle {
stream: Box::new(AdapterInventoryStream { inner: stream }),
control: Arc::new(AdapterInventoryControl { inner: control }),
})
}
async fn read_tag_values(
&self,
server: &str,
tag_ids: Vec<String>,
) -> anyhow::Result<Vec<TagValue>> {
let values = self.client.read_tag_values(server, tag_ids).await?;
Ok(values.into_iter().map(map_tag_value).collect())
}
async fn write_tag_value(
&self,
server: &str,
tag_id: &str,
value: OpcValue,
) -> anyhow::Result<WriteResult> {
let result = self
.client
.write_tag_value(server, tag_id, map_value(value))
.await?;
Ok(map_write_result(result))
}
}
struct AdapterInventoryControl {
inner: ExtInventoryControl,
}
impl InventoryControl for AdapterInventoryControl {
fn pause(&self) {
self.inner.pause();
}
fn resume(&self) {
self.inner.resume();
}
fn cancel(&self) {
self.inner.cancel();
}
fn is_cancelled(&self) -> bool {
self.inner.is_cancelled()
}
}
struct AdapterInventoryStream {
inner: ExtInventoryStream,
}
#[async_trait::async_trait]
impl InventoryStream for AdapterInventoryStream {
async fn next(&mut self) -> Option<anyhow::Result<InventoryEvent>> {
self.inner
.message()
.await
.map(|result| result.map(map_inventory_event).map_err(anyhow::Error::from))
}
}
fn map_inventory_event(event: ExtInventoryEvent) -> InventoryEvent {
match event {
ExtInventoryEvent::Entry(entry) => InventoryEvent::Entry(InventoryEntry {
display_name: entry.display_name,
item_id: entry.item_id,
kind: match entry.kind {
ExtBrowseNodeKind::Item => InventoryNodeKind::Item,
ExtBrowseNodeKind::BranchAndItem => InventoryNodeKind::BranchAndItem,
ExtBrowseNodeKind::Branch => InventoryNodeKind::Item,
},
breadcrumbs: entry.breadcrumbs,
}),
ExtInventoryEvent::Progress(progress) => InventoryEvent::Progress(InventoryProgress {
branches_visited: progress.branches_visited,
entries_seen: progress.entries_seen,
unique_items: progress.unique_items,
active_time_ms: progress.active_time_ms,
paused_time_ms: progress.paused_time_ms,
items_per_second: progress.items_per_second,
estimated_remaining_ms: progress.estimated_remaining_ms,
}),
ExtInventoryEvent::Completed(completed) => {
let (organization, source) = map_capabilities(&completed.capabilities);
InventoryEvent::Completed(InventoryCompleted {
complete: completed.complete,
cancelled: completed.cancelled,
truncated: completed.truncated,
warning: completed.warning,
organization,
source,
})
}
}
}
fn parse_token<T>(value: &str, label: &str) -> anyhow::Result<T>
where
T: std::str::FromStr,
T::Err: std::fmt::Display,
{
value
.parse()
.map_err(|error| anyhow::anyhow!("invalid {label} token: {error}"))
}
fn map_capabilities(
capabilities: &opc_da_client::BrowseCapabilities,
) -> (NamespaceOrganization, BrowseSource) {
let organization = match capabilities.namespace {
BrowseNamespace::Flat => NamespaceOrganization::Flat,
BrowseNamespace::Hierarchical => NamespaceOrganization::Hierarchical,
BrowseNamespace::Unknown => NamespaceOrganization::Unspecified,
};
let source = match capabilities.namespace {
BrowseNamespace::Flat => BrowseSource::Flat,
BrowseNamespace::Hierarchical | BrowseNamespace::Unknown => {
if capabilities.supports_da3 {
BrowseSource::Da3
} else if capabilities.supports_da2 {
BrowseSource::Da2
} else {
BrowseSource::Derived
}
}
};
(organization, source)
}
fn map_node_kind(kind: ExtBrowseNodeKind) -> BrowseNodeKind {
match kind {
ExtBrowseNodeKind::Branch => BrowseNodeKind::Branch,
ExtBrowseNodeKind::Item => BrowseNodeKind::Item,
ExtBrowseNodeKind::BranchAndItem => BrowseNodeKind::BranchAndItem,
}
}
fn map_tag_value(value: ExtTagValue) -> TagValue {
TagValue {
tag_id: value.tag_id,
value: value.value,
quality: value.quality,
timestamp: value.timestamp,
}
}
fn map_write_result(result: ExtWriteResult) -> WriteResult {
WriteResult {
tag_id: result.tag_id,
success: result.success,
error: result.error,
}
}
fn map_value(value: OpcValue) -> ExtOpcValue {
match value {
OpcValue::String(value) => ExtOpcValue::String(value),
OpcValue::Int(value) => ExtOpcValue::Int(value),
OpcValue::Float(value) => ExtOpcValue::Float(value),
OpcValue::Bool(value) => ExtOpcValue::Bool(value),
}
}