Writing A Driver
Everything you need to know about writing drivers for PlaceOS
There are three main uses of drivers:
- Streaming IO (TCP, SSH, UDP, Multicast etc.)
- HTTP Client
- Logic
From a driver structure standpoint there is no difference between these types.
- The same driver works over a TCP, UDP or SSH transport
- All drivers support HTTP methods (except logic modules)
- for example a websocket driver or tcp driver will also be provided a default HTTP client at the base URI of the websocket and IP address of the tcp driver.
- this default client URL can be overwritten, for example where the HTTP port is different to the websocket port
transport.http_uri_override = URI.new
- All drivers have access to logic helpers when associated with a System
The queue is a list of potentially asynchronous tasks that should be performed in a sequence.
- Each task has a priority (defaults to
50) - higher priority tasks run first - Tasks have names - if there's a name conflict, the newer task overwrites the older one
- Tasks have a timeout (defaults to
5.seconds) - Tasks a set amount of re-tries (defaults to
3before failing)
Tasks have a callback which can run the task
# => you can set queue defaults globally
# set a delay between the current task completing and the next task
queue.delay = 1.second
queue.retries = 5
queue(priority: 20, timeout: 1.second) do |task|
# perform action here
# signal result
task.success("optional success value")
task.abort("optional failure message")
task.retry
# Give me more time to complete the task
task.reset_timers
end
In most cases you won't need to use the queue explicitly, but it's good to understand that how it functions.
The transport loaded is defined by settings in the database.
def on_load
transport.tokenizer = Tokenizer.new("\r\n")
end
Here are the ways to use streaming IO methods:
- 1.Send and receive
def perform_action
# You call send with some data.
# you can also optionally pass some queue options to the function
send("message data", priority: 30, name: "generic-message")
end
# A common received function for handling responses
def received(data, task)
# data is always `Bytes`
# task is always `PlaceOS::Driver::Task?` (i.e. could be nil if no active task)
# convert data into the appropriate format
data = String.new(data)
# decide if the request was a success or not
# you can pass any value that is JSON serialisable to success
# (if it can't be serialised then nil is sent)
task.try &.success(data)
end
- 1.Send and callback
def perform_action
request = "build request"
send(request, priority: 30, name: "generic-message") do |data, task|
data = String.new(data)
# process response here (might need to know the request context)
task.try &.success(data)
end
end
- 1.Send immediately (no queuing)
def perform_action_now!
transport.send("no queue")
end
You can also add a pre-processor to data coming in. This can be useful if you want to strip away a protocol layer. For example, if you are using Telnet and want to remove the telnet signals leaving the raw data for tokenizing
def on_load
transport.pre_processor do |bytes|
# you must return some byte data or nil if no processing is required
# tokenisation occurs on the data returned here
bytes[1..-2]
end
end
def received(data, task)
# data coming in here is both pre_processed and tokenised
end
All drivers have built-in methods for performing HTTP requests.
- For streaming IO devices this defaults to
http://device.ip.address(httpsif the transport is using TLS / SSH) - All devices can provide a custom HTTP base URI
There are methods for all the typical HTTP verbs: get, post, put, patch, delete
def perform_action
basic_auth = "Basic #{Base64.strict_encode("#{@username}:#{@password}")}"
response = post("/v1/message/path", body: {
messages: numbers,
}.to_json, headers: {
"Authorization" => basic_auth,
"Content-Type" => "application/json",
"Accept" => "application/json",
}, params: {
"key" => "value"
})
raise "request failed with #{response.status_code}" unless (200...300).include?(data.status_code)
end
SSH connections will attempt to open a shell to the remote device. Sometimes you may be able to execute operations independently.
def perform_action
# if the application launched supports input you can use the bidirectional IO
# to communicate with the app
io = exec("command")
end
Logic drivers belong to a System and cannot be shared, which makes them different from other transports. All other drivers can appear in any number of systems.
You can access remote modules in the system via the
system helper# Get a system proxy
sys = system
sys.name #=> "Name of system"
sys.email #=> "[email protected]"
sys.capacity #=> 12
sys.bookable #=> true
sys.id #=> "sys-tem~id"
sys.modules #=> ["Array", "Of", "Unique", "Module", "Names", "In", "System"]
sys.count("Module") #=> 3
sys.implementing(PlaceOS::Driver::Interface::Powerable) #=> ["Camera", "Display"]
# Look at status on a remote module
system[:Display][:power] #=> true
# Access a different module index
system[:Display_2][:power]
system.get(:Display, 2)[:power]
# Access all modules of a type
system.all(:Display)
# Check if a module exists
system.exists?(:Display) #=> true
system.exists?(:Display_2) #=> false
You can bind to state in remote modules
bind Display_1, :power, :power_changed
private def power_changed(subscription, new_value)
logger.debug new_value
end
# you can also bind to internal state (available in all drivers)
bind :power, :power_changed
It's also possible to create shortcuts to other modules. This is powerful as these shortcuts are exposed as metadata. It allows Backoffice to perform system verification.
For example, consider the following video conference system:
# It requires at least one camera that can move and be turned on and off
accessor camera : Array(Camera), implementing: [Powerable, Moveable]
# Optional room blinds that can be opened and closed
accessor blinds : Array(Blind)?, implementing: [Switchable]
# A single display is required with an optional screen (maybe it's a projector)
accessor main_display : Display_1, implementing: Powerable
accessor screen : Screen?
Cross system communication is possible if you know the ID of the remote system.
# once you have reference to the remote system you can perform any
# actions that you might perform on the local system
sys = system("sys-12345")
sys.name #=> "Name of remote system"
sys[:Display_2][:power] #=> true
You can dynamically bind to state of interest in remote modules
# subscription is returned and provided with every status update in the callback
subscription = system.subscribe(:Display_1, :power) do |subscription, new_value|
# values are always raw JSON strings
JSON.parse(new_value)
end
# Local subscriptions
subscription = subscribe(:state) do |subscription, new_value|
# values are always raw JSON strings
JSON.parse(new_value)
end
# Clearing all subscriptions
subscriptions.clear
Like subscriptions, channels can be setup for broadcasting any data that might not need be exposed as state.
subscription = monitor(:channel_name) do |subscription, new_value|
# values are always raw JSON strings
JSON.parse(new_value)
end
# Publish something on the channel to all listeners
publish(:channel_name, "some event")
def connected
schedule.every(40.seconds) { poll_device }
schedule.in(200.milliseconds) { send_hello }
end
def disconnected
schedule.clear
end
Settings are stored as JSON and then extracted as required, serializing to the specified type. There are two types:
- Required settings - raise an error if the setting is unavailable
- Optional settings - return
nilif the setting is unavailable
All settings will raise an error if they exist but fail to serialize (due to incorrect formatting etc.)
# Required settings
def on_update
@display_id = setting(Int32, :display_id)
# Can extract deeply nested values
# i.e. {input: {list: ["HDMI", "VGA"] }}
@primary_input = setting(InputEnum, :input, :list, 0)
end
# Optional settings (you can optionally provide a default)
def on_update
@display_id = setting?(Int32, :display_id) || 1
@primary_input = setting?(InputEnum, :input, :list, 0) || InputEnum::HDMI
end
You can update the local settings of a module, persisting them to the database. Settings must be JSON serializable
define_setting(:my_setting_name, "some JSON serialisable data")
warnand above are written to diskdebugandinfoare only available when there is an open debugging session
logger.warn { "error unknown response" }
logger.debug { "function called with #{value}" }
The logging format has been pre-configured so all logging from PlaceOS is uniform and parsed as-is
Many components use metadata to simplify configuration.
generic_name=> the name that a system should use to access the moduledescriptive_name=> the manufacturers name for the devicedescription=> notes or any other descriptive information you wish to addtcp_port=> TCP port the TCP transport should connect toudp_port=> UDP port the UDP transport should connect touri_base=> The HTTP base for any HTTP requestsdefault_settings=> Default or example settings that for configuring a module
class MyDevice < PlaceOS::Driver
generic_name :Driver
descriptive_name "Driver model Test"
description "This is the driver used for testing"
tcp_port 22
default_settings({
name: "Room 123",
username: "steve",
password: "$encrypt",
complex: {
crazy_deep: 1223,
},
})
# ...
end
By default all public functions are exposed for execution. You can limit who is able to execute sensitive functions.
@[Security(Level::Administrator)]
def perform_task(name : String | Int32)
queue &.success("hello #{name}")
end
Use the
Security annotation to define the access level of the function. The options are:- Administrator
Level::Administrator - Support
Level::Support
Drivers can expose any methods that make sense for the device, service or logic they encapsulate. Across these there are often core sets of similar functionality. Interfaces provide a standard way of implementing and interacting with this.
Though optional, they're recommended as they make drivers more modular and less complex.
A full list of interfaces is available in the driver framework. This will expand over time to cover common, repeated patterns as they emerge.
Each interface is a module containing abstract methods, types and functionality built from these.
First include the module within the driver body.
include Interface::Powerable
You will then need to provide implementations of the abstract methods. The compiler will guide you in this.
Some interfaces will also provide default implementation for other methods. These may be overridden if the device or service provides a more efficient way to do the same thing. To keep compatibility, overridden methods must maintain feature and functional parity with the original.
You can use the
system.implementing method from any logic module. It returns a list of all drivers in the system which implement the Interface.The
accessor macro provides a way to declare a dependency on a sibling driver for a specific function.Where multiple functions are likely to raise similar errors, the errors can be handled generically using the
rescue_from helper.class MyDevice < PlaceOS::Driver
rescue_from JSON::ParseException do |error|
logger.warn(exception: error) { "error parsing JSON payload" }
{} of String => JSON::Any
end
# any external call to this function will result in the empty hash above
# being returned to the caller. Internally in the driver the error will
# be raised as normal.
def no_error_externally
JSON.parse %({invalid: 'json')
end
end
Alternatively this can be handled via an explicit function. Useful if it's desirable to use the same code in the received function.
class MyDevice < PlaceOS::Driver
rescue_from JSON::ParseException, :handle_parse_error
protected def handle_parse_error(error)
logger.warn(exception: error) { "error parsing JSON payload" }
{} of String => JSON::Any
end
# The above might be used as follows:
def no_error_externally
# externally returns {}
JSON.parse %({invalid: 'json')
end
# Keep error parsing DRY
def received(data, task)
result = JSON.parse(String.new data)
task.try &.success(result)
rescue error : JSON::ParseException
result = handle_parse_error(error)
task.try &.success(result)
end
end
Last modified 4mo ago