Drupal 11：使用 REST 服务控制 LED 灯
在我公司之前发布的关于 Pimoroni Plasma 2350 W 的文章之后，我们决定利用其 Wi-Fi 接口做些有趣的事情，将其连接到 Drupal 网站。在 Drupal 开发中，这样的应用场景十分常见，尤其是在 Drupal11 下，能为我们带来更多的可能性。
Pimoroni 的 Plasma 2350 W 固件附带了一个示例，该示例可连接到一个免费 API 来随机更新颜色。我们看到它运行后意识到，将其转换为从 Drupal REST 服务拉取数据应该不会太难。这也是 Drupal 模块开发中的一个典型应用。
在 Drupal 中创建 RESTful 服务非常容易，只需要一个类。我们需要做的就是创建一个表单来控制在 REST 服务中选择的颜色。
在本文中，我们将介绍创建一个包含 RESTful 接口的 Drupal 模块，然后把 Plasma 2350 W 连接到该模块以更新灯光颜色。
一、设置表单
为了能够设置 LED 灯的颜色，我们需要创建一个专门用于此目的的表单。
为了将颜色保存到系统中，我们将使用 state 服务，这是一个方便的键/值服务，可让我们将简单的值写入数据库。该服务是存储不属于 Drupal 配置系统的值的好方法。理想情况下，你希望这些值在不存在时能由系统轻松重新创建。因此，颜色设置非常适合使用 state 服务。
注入此服务后设置表单并不复杂，但我们还可以通过抽象 state 本身的获取和设置方法来简化表单集成。
以下是表单类的基本结构。
namespace Drupal\hashbangcode_plasma\Form;
use Drupal\Core\Form\FormBase;
use Drupal\Core\Form\FormStateInterface;
use Drupal\Core\State\StateInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;
class SetColourForm extends FormBase {
/**
* The state service.
*
* @var \Drupal\Core\State\StateInterface
*/
protected StateInterface $state;
public static function create(ContainerInterface $container) {
$instance = new static();
$instance->state = $container->get('state');
return $instance;
}
/**
* Get the colour from the state service.
*
* @return string
* The colour.
*/
public function getColourState() {
return $this->state->get('hashbangcode_plasma.colour', '#ffffff');
}
/**
* Set the colour in the state interface.
*
* @param string $colour
* The colour to set.
*/
public function setColourState($colour) {
$this->state->set('hashbangcode_plasma.colour', $colour);
}
public function buildForm(array $form, FormStateInterface $form_state) {
// Add form here.
}
public function submitForm(array &$form, FormStateInterface $form_state) {
// Add submit here.
}
}
这里需要注意的主要是，state 服务有一个 get() 方法，该方法允许我们设置一个默认值。在上述代码中，这个默认值被设置为纯白色（即 #ffffff）。
这个项目的表单只需要接受颜色并允许提交表单。有趣的是，Drupal 自带了一种 “color” 字段类型，它是 HTML 输入类型 “color” 的包装器。这意味着我们不需要包含任何额外的库来实现颜色选择功能，只需添加一个具有适当类型的字段即可。
public function buildForm(array $form, FormStateInterface $form_state) {
$form['colour'] = [
'#type' => 'color',
'#title' => $this->t('Colour'),
'#default_value' => $this->getColourState(),
'#required' => TRUE,
];
$form['submit'] = [
'#type' => 'submit',
'#value' => 'Set Colour',
];
return $form;
}
HTML 颜色字段类型是个有趣的点，虽然它被广泛支持，并且允许通过一个简单的 HTML 元素进行颜色选择，但我们在不同的设备和浏览器上至少看到了 3 种不同的输入界面。
表单的提交处理函数非常简单。我们只需要从表单状态中提取颜色值，使用一开始设置的 setColourState() 方法将其写入 state 服务，然后重新构建表单，以便页面重新加载后新值能显示在表单中。
public function submitForm(array &$form, FormStateInterface $form_state) {
$colour = $form_state->getValue('colour');
$this->setColourState($colour);
$form_state->setRebuild();
}
表单基本就是这样了，但我们可以通过 Drupal 的 flood 服务来防止滥用，从而进一步改进它。
防刷保护
我们不希望看到表单上线后，人们每隔几秒就更改一次颜色来滥用它。幸运的是，Drupal 有一个内置的名为 flood 的服务，它使我们能够通过跟踪给定时间段内的提交次数并将其与阈值进行比较，来防止表单被滥用。
要让这个功能生效，我们需要像添加 state 服务一样，将 flood 服务添加到表单中。
/**
* The flood service.
*
* @var \Drupal\Core\Flood\FloodInterface
*/
protected FloodInterface $flood;
public static function create(ContainerInterface $container) {
$instance = new static();
$instance->state = $container->get('state');
$instance->flood = $container->get('flood');
return $instance;
}
然后我们更新提交处理函数，这样当用户更改颜色时，我们就可以向 flood 服务记录他们的操作。在这种情况下，我们使用用户的 IP 地址来记录这个操作。
public function submitForm(array &$form, FormStateInterface $form_state) {
$colour = $form_state->getValue('colour');
$this->setColourState($colour);
$form_state->setRebuild();
// Register flood handler.
$floodIdentifier = $this->getRequest()->getClientIp();
$this->flood->register('hashbangcode_plasma.plasma_flood_protection', self::ABUSE_WINDOW, $floodIdentifier);
}
有了这些设置后，我们可以给表单添加一个验证处理函数，用于检查 flood 阈值的状态。如果用户的提交触发了 flood 阈值（这里设置为 10 分钟内 30 次），我们将拒绝该提交，并给用户一个错误消息。
public function validateForm(array &$form, FormStateInterface $form_state) {
parent::validateForm($form, $form_state);
$floodIdentifier = $this->getRequest()->getClientIp();
if (!$this->flood->isAllowed('hashbangcode_plasma.plasma_flood_protection', 30, 610, $floodIdentifier)) {
$form_state->setErrorByName('', $this->t('Too many uses of this form from your IP address. This address is temporarily banned. Try again later.'));
}
}
如果你想了解更多关于 Drupal 中 flood 系统的相关内容，我们之前写过相关文章，里面有关于这个系统的更多详细信息。
现在我们有了一个（受保护的）表单，接下来可以构建一个 REST 资源，以便将颜色以 JSON 提要的形式显示出来。
二、创建 REST 资源
Drupal 的 API 系统非常强大，它自带了用于格式和认证机制的插件。对于这个项目，我们只需要 JSON 格式化器和基本的 “cookie” 认证提供者，这样就可以让用户轻松匿名访问该资源。在 Drupal 开发中，合理利用这些插件能大大提高开发效率。
要定义一个 RESTful API 接口，我们首先需要创建一个插件类。这个类将设置端点并编写驱动接口所需的代码。此外，由于我们的颜色存储在 state 服务中，我们将使用依赖注入将该服务注入到类中。
/**
* Provides a resource for database watchdog log entries.
*/
#[RestResource(
id: "plasma",
label: new TranslatableMarkup("Plasma resource"),
uri_paths: [
"canonical" => "/plasma/[REDACTED]",
]
)]
class PlasmaResource extends ResourceBase {
}
我们想为这个资源定义一个 GET 方法，所以需要创建一个名为 get() 的方法，该方法将由 REST 服务自动触发。这个方法实际上很简单，我们将响应负载定义为一个数组，并向其中添加一个名为 “colour” 的字段。然后，我们使用 state 服务从表单设置的 state 中提取颜色，如果没有则默认为纯白色。
通常，当我们从 REST 资源返回响应时，会创建一个新的 Drupal\rest\ResourceResponse 对象，该对象会处理将负载转换为 JSON 字符串的所有上游格式化操作。但对于这个项目，这种响应类型不太合适，因为它会被 Drupal 缓存。由于 Plasma 2350 W 是向 REST 端点发起匿名请求（这样我们就不必处理设备上的认证问题），表单中所做的更新就无法被获取到。
经过一些试验，我们发现对于匿名用户使用的 RESTful 资源，关闭缓存的最有效方法是返回一个新的 Drupal\rest\ModifiedResourceResponse 对象。它的作用与 ResourceResponse 类似，但假定每个响应都是唯一的，因此可以防止响应被缓存。我们看到很多人建议将缓存最大期限设置为 0，但这对匿名流量不起作用。
以下是 get() 方法的内容。
public function get() {
$colour = [
'colour' => $this->state->get('hashbangcode_plasma.colour', '#ffffff'),
];
// Do not cache the response.
return new ModifiedResourceResponse($colour, 200);
}
REST 资源将以 200 状态码响应，并返回一个包含少量 JSON 数据的有效负载。
为了完整起见，以下是完整的 REST 插件类。
<?php
namespace Drupal\hashbangcode_plasma\Plugin\rest\resource;
use Drupal\Core\State\StateInterface;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\rest\Attribute\RestResource;
use Drupal\rest\ModifiedResourceResponse;
use Drupal\rest\Plugin\ResourceBase;
use Symfony\Component\DependencyInjection\ContainerInterface;
/**
* Provides a resource for database watchdog log entries.
*/
#[RestResource(
id: "plasma",
label: new TranslatableMarkup("Plasma resource"),
uri_paths: [
"canonical" => "/plasma/[REDACTED]",
]
)]
class PlasmaResource extends ResourceBase {
/**
* The state service.
*
* @var \Drupal\Core\State\StateInterface
*/
protected StateInterface $state;
public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) {
$instance = new static(
$configuration,
$plugin_id,
$plugin_definition,
$container->getParameter('serializer.formats'),
$container->get('logger.factory')->get('rest')
);
$instance->state = $container->get('state');
return $instance;
}
public function get() {
$colours = [
'colour' => $this->state->get('hashbangcode_plasma.colour', '#ffffff'),
];
// Do not cache the response.
return new ModifiedResourceResponse($colours, 200);
}
}
有了这些设置后，我们现在需要启用该资源，因为所有 REST 资源默认最初都是禁用的。
Drupal 没有内置的界面来启用或配置 REST 资源，但我们可以使用 REST UI 模块轻松启用 Plasma 资源，并设置它将使用的方法、格式和认证类型。安装 REST UI 模块后，我们可以启用 Plasma 资源，并允许它接受 GET 请求，现在看起来是这样的。
这将生成一个配置实体，我们可以将其导出并写入 Drupal 配置目录，然后进行上游部署。
langcode: en
status: true
dependencies:
module:
- hashbangcode_plasma
- serialization
- user
id: plasma
plugin_id: plasma
granularity: resource
configuration:
methods:
- GET
formats:
- json
authentication:
- cookie
当然，如果不设置资源的权限，我们就无法访问 RESTful 接口，所以最后一步是允许匿名用户访问 plasma 资源。我们在文章中对资源的实际路径进行了模糊处理，但一旦访问，我们会看到以下输出。
{"colour":"#ffffff"}
通过颜色表单更改这个颜色，然后刷新 REST 资源，我们会看到一个新的颜色。
{"colour":"#19bbbe"}
Drupal 方面的所有设置都已完成并且可以正常工作，现在让我们看看如何使用 MicroPython 从 API 拉取颜色并改变灯光颜色。
三、编写 MicroPython 代码
正如我们之前提到的，Pimoroni 的 Plasma 2350 W 固件附带了一个 MicroPython 脚本，该脚本可连接到一个包含颜色信息的免费 API。每隔 120 秒，MicroPython 会连接到这个 API，将找到的十六进制代码转换为 RGB 值，然后更新 LED 灯的颜色。
我们需要做的是更新使用的 URL，并缩短更新间隔时间。这只需在脚本中修改 URL 变量即可。
URL = "https://www.hashbangcode.com/plasma/[REDACTED]"
我们自定义的 JSON 提要使用的字段与原始提要不同，所以我们只需要更新从提要中提取该值的代码。
# extract hex colour from the data
hex = j["colour"]
以下是完整的脚本，它与原始脚本非常相似，但我们对其进行了一些简化。
import time
import urequests
from ezwifi import connect
from machine import Pin
import plasma
URL = "https://www.hashbangcode.com/plasma/[REDACTED]"
UPDATE_INTERVAL = 20 # refresh interval in secs
NUM_LEDS = 66
# wifi connection failed
def wifi_failed(message=""):
print(f"Wifi connection failed! {message}")
# Print out WiFi connection messages for debugging
def wifi_message(_wifi, message):
print(message)
def hex_to_rgb(hex):
# converts a hex colour code into RGB
h = hex.lstrip("#")
r, g, b = (int(h[i:i + 2], 16) for i in (0, 2, 4))
return r, g, b
# set up the Pico W's onboard LED
pico_led = Pin("LED", Pin.OUT)
# set up the WS2812 / NeoPixel™ LEDs
led_strip = plasma.WS2812(NUM_LEDS, color_order=plasma.COLOR_ORDER_BGR)
# start updating the LED strip
led_strip.start()
# set up wifi
try:
connect(failed=wifi_failed, info=wifi_message, warning=wifi_message, error=wifi_message)
except ValueError as e:
wifi_failed(e)
while True:
r = urequests.get(URL)
j = r.json()
r.close()
# flash the onboard LED after getting data
pico_led.value(True)
time.sleep(0.2)
pico_led.value(False)
# extract hex colour from the data
hex = j["colour"]
# and convert it to RGB
r, g, b = hex_to_rgb(hex)
# light up the LEDs
for i in range(NUM_LEDS):
led_strip.set_rgb
