linkos 数据同步及处理

app/Console/Commands/DeviceLogDailyReportCommand.php

@@ -0,0 +1,148 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Enums\DeviceType;
+use App\Models\Device;
+use App\Models\MeteorologicalDailyReport;
+use App\Models\MeteorologicalMonitoringDailyLog;
+use App\Models\MeteorologicalMonitoringLog;
+use App\Models\MeteorologicalReport;
+use App\Models\SoilDailyReport;
+use App\Models\SoilMonitoringDailyLog;
+use App\Models\SoilMonitoringLog;
+use App\Models\SoilReport;
+use App\Models\WaterQualityDailyReport;
+use App\Models\WaterQualityMonitoringDailyLog;
+use App\Models\WaterQualityMonitoringLog;
+use App\Models\WaterQualityReport;
+use App\Services\DeviceLogService;
+use Illuminate\Console\Command;
+
+class DeviceLogDailyReportCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'device-log:daily-report
+                            {factory}
+                            {--sleep=300 : 监控报告生产后的休眠时间（秒）}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = '按设备厂商生成监控报告';
+
+    /**
+     * @var \App\Services\DeviceLogService
+     */
+    protected $deviceLogService;
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(DeviceLogService $deviceLogService)
+    {
+        $this->deviceLogService = $deviceLogService;
+
+        $factory = $this->argument('factory');
+
+        $sleep = (int) value(fn ($sleep) => is_numeric($sleep) ? $sleep : 300, $this->option('sleep'));
+
+        while (true) {
+            /** @var \Illuminate\Database\Eloquent\Collection */
+            $devices = Device::with(['supplier'])->supplierBy($factory)->get();
+
+            foreach ($devices as $device) {
+                switch ($device->supplier?->key) {
+                    case 'linkos':
+                        $this->createReportToLinkosDevice($device);
+                        break;
+                }
+            }
+
+            sleep($sleep);
+        };
+    }
+
+    /**
+     * 创建 linkos 设备报告
+     */
+    protected function createReportToLinkosDevice(Device $device): void
+    {
+        [$lastReportedAt, $latestReportedAt] = value(function (Device $device) {
+            $lastReportedAt = null;
+
+            $latestReportedAt = null;
+
+            switch ($device->type) {
+                case DeviceType::WaterQuality:
+                    $lastReportedAt = WaterQualityMonitoringDailyLog::where('device_id', $device->id)
+                        ->latest('monitored_at')
+                        ->value('monitored_at');
+
+                    $lastReportedAt ??= WaterQualityMonitoringLog::where('device_id', $device->id)
+                        ->oldest('monitored_at')
+                        ->value('monitored_at');
+
+                    if ($lastReportedAt) {
+                        $latestReportedAt = WaterQualityMonitoringLog::where('device_id', $device->id)
+                            ->latest('monitored_at')
+                            ->value('monitored_at');
+                    }
+                    break;
+
+                case DeviceType::Meteorological:
+                    $lastReportedAt = MeteorologicalMonitoringDailyLog::where('device_id', $device->id)
+                        ->latest('monitored_at')
+                        ->value('monitored_at');
+
+                    $lastReportedAt ??= MeteorologicalMonitoringLog::where('device_id', $device->id)
+                        ->oldest('monitored_at')
+                        ->value('monitored_at');
+
+                    if ($lastReportedAt) {
+                        $latestReportedAt = MeteorologicalMonitoringLog::where('device_id', $device->id)
+                            ->latest('monitored_at')
+                            ->value('monitored_at');
+                    }
+                    break;
+
+                case DeviceType::Soil:
+                    $lastReportedAt = SoilMonitoringDailyLog::where('device_id', $device->id)
+                        ->latest('monitored_at')
+                        ->value('monitored_at');
+
+                    $lastReportedAt ??= SoilMonitoringLog::where('device_id', $device->id)
+                        ->oldest('monitored_at')
+                        ->value('monitored_at');
+
+                    if ($lastReportedAt) {
+                        $latestReportedAt = SoilMonitoringLog::where('device_id', $device->id)
+                            ->latest('monitored_at')
+                            ->value('monitored_at');
+                    }
+                    break;
+            }
+
+            return [$lastReportedAt, $latestReportedAt];
+        }, $device);
+
+        if ($lastReportedAt === null || $latestReportedAt === null) {
+            return;
+        }
+
+        /** @var \Carbon\Carbon */
+        $startAt = $lastReportedAt->copy()->startOfDay();
+
+        do {
+            $this->deviceLogService->createDailyReportToLinkosDevice($device, $startAt->copy());
+
+            $startAt->addDay();
+        } while ($latestReportedAt->gte($startAt));
+    }
+}

app/Console/Commands/DeviceLogReportCommand.php

@@ -0,0 +1,93 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Enums\DeviceType;
+use App\Models\Device;
+use App\Models\LinkosDeviceLog;
+use App\Models\MeteorologicalMonitoringLog;
+use App\Models\SoilMonitoringLog;
+use App\Models\WaterQualityMonitoringLog;
+use App\Services\DeviceLogService;
+use Illuminate\Console\Command;
+
+class DeviceLogReportCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'device-log:report
+                            {factory}
+                            {--sleep=300 : 监控报告生产后的休眠时间（秒）}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = '按设备厂商生成监控报告';
+
+    /**
+     * @var \App\Services\DeviceLogService
+     */
+    protected $deviceLogService;
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(DeviceLogService $deviceLogService)
+    {
+        $this->deviceLogService = $deviceLogService;
+
+        $factory = $this->argument('factory');
+
+        $sleep = (int) value(fn ($sleep) => is_numeric($sleep) ? $sleep : 300, $this->option('sleep'));
+
+        while (true) {
+            /** @var \Illuminate\Database\Eloquent\Collection */
+            $devices = Device::with(['supplier'])->supplierBy($factory)->get();
+
+            foreach ($devices as $device) {
+                switch ($device->supplier?->key) {
+                    case 'linkos':
+                        $this->createReportToLinkosDevice($device);
+                        break;
+                }
+            }
+
+            sleep($sleep);
+        };
+    }
+
+    /**
+     * 创建 linkos 设备报告
+     */
+    protected function createReportToLinkosDevice(Device $device): void
+    {
+        $lastReportedAt = match ($device->type) {
+            DeviceType::Soil => SoilMonitoringLog::where('device_id', $device->id)->latest('monitored_at')->value('monitored_at'),
+            DeviceType::WaterQuality => WaterQualityMonitoringLog::where('device_id', $device->id)->latest('monitored_at')->value('monitored_at'),
+            DeviceType::Meteorological => MeteorologicalMonitoringLog::where('device_id', $device->id)->latest('monitored_at')->value('monitored_at'),
+            default => null,
+        };
+
+        if (is_null($lastReportedAt ??= LinkosDeviceLog::where('device_id', $device->sn)->oldest('reported_at')->value('reported_at'))) {
+            return;
+        }
+
+        if (is_null($latestReportedAt = LinkosDeviceLog::where('device_id', $device->sn)->latest('reported_at')->value('reported_at'))) {
+            return;
+        }
+
+        /** @var \Carbon\Carbon */
+        $startAt = $lastReportedAt->copy()->startOfHour();
+
+        do {
+            $this->deviceLogService->createReportToLinkosDevice($device, $startAt->copy());
+
+            $startAt->addHour();
+        } while ($latestReportedAt->gte($startAt));
+    }
+}

app/Console/Commands/DeviceLogSyncCommand.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Enums\DeviceType;
+use App\Models\Device;
+use App\Services\DeviceLogService;
+use Illuminate\Console\Command;
+use Throwable;
+
+class DeviceLogSyncCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'device-log:sync
+                            {factory}
+                            {--sleep=60 : 设备数据同步完成后的休眠时间（秒）}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = '按设备厂商同步数据';
+
+    /**
+     * @var \App\Services\DeviceLogService
+     */
+    protected $deviceLogService;
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(DeviceLogService $deviceLogService)
+    {
+        $this->deviceLogService = $deviceLogService;
+
+        $factory = $this->argument('factory');
+
+        $sleep = (int) value(fn ($sleep) => is_numeric($sleep) ? $sleep : 60, $this->option('sleep'));
+
+        while (true) {
+            $this->runSync($factory);
+
+            sleep($sleep);
+        };
+    }
+
+    /**
+     * 执行同步
+     */
+    protected function runSync(string $factory): void
+    {
+        $end = now();
+        $start = $end->copy()->subHours(1);
+
+        $this->info('------------------------------------------');
+        $this->info('开始时间: '. $start);
+        $this->info('结束时间: '. $end);
+
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $devices = Device::with(['supplier'])->supplierBy($factory)->get();
+
+        /** @var \App\Models\Device */
+        foreach ($devices as $device) {
+            $this->info('==========================================');
+            $this->info('设备编号: ' . $device->sn);
+            $this->info('设备名称: ' . $device->name);
+            $this->info('设备类型: ' . match ($device->type) {
+                DeviceType::Soil => '土壤设备',
+                DeviceType::WaterQuality => '水质设备',
+                DeviceType::Meteorological => '气象设备',
+                default => '其它',
+            });
+
+            try {
+                $this->deviceLogService->sync($device, $start, $end);
+
+                $this->info('同步成功！');
+            } catch (Throwable $e) {
+                report($e);
+
+                $this->error('同步失败: '. $e->getMessage());
+            }
+
+            $this->info('==========================================');
+        }
+
+        $this->info('------------------------------------------');
+        $this->newLine();
+    }
+}

app/Http/Controllers/Callback/LinkosController.php

@@ -45,6 +45,8 @@ class LinkosController extends Controller
      */
     protected function handleDeviceDataNotify(array $data)
     {
+        return;
+
         if (! is_array($deviceData = $data['data'] ?? [])) {
             $deviceData = [];
         }

app/Iot/Linkos/HttpClient.php

@@ -0,0 +1,160 @@
+<?php
+
+namespace App\Iot\Linkos;
+
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\Http;
+use RuntimeException;
+
+class HttpClient
+{
+    public const ENDPOINT_URL = 'http://service.easylinkin.com';
+
+    public function __construct(
+        protected readonly string $apiKey,
+        protected readonly string $apiSecret,
+    ) {
+    }
+
+    /**
+     * 获取设备的历史数据
+     *
+     * @param string $deviceId
+     * @param \Illuminate\Support\Carbon $start
+     * @param \Illuminate\Support\Carbon $end
+     * @param int $page
+     * @param int $perPage
+     * @return array
+     */
+    public function deviceFlowList(string $deviceId, Carbon $start, Carbon $end, int $page = 1, int $perPage = 50): array
+    {
+        $result = $this->post('/api/deviceFlow/v1/list', [
+            'device_id' => $deviceId,
+            'start_time' => $start->unix() * 1000,
+            'end_time' => $end->unix() * 1000,
+            'pageable' => [
+                'page' => $page - 1,
+                'size' => $perPage,
+            ],
+        ]);
+
+        if (data_get($result, 'success') !== true) {
+            throw new RuntimeException(data_get($result, 'msg', '出错啦！'));
+        }
+
+        return $result['data'];
+    }
+
+    /**
+     * 设备数据下行
+     *
+     * @param string $deviceId
+     * @param string $service
+     * @param array $data
+     * @param boolean $confirm
+     * @param boolean $clear
+     * @param boolean $schedule
+     * @return array
+     */
+    public function deviceDataDownlink(string $deviceId, string $service, array $data = [], bool $confirm = true, bool $clear = true, bool $schedule = false): array
+    {
+        return $this->post('/api/down', [
+            'device_id' => $deviceId,
+            'service_id' => $service,
+            'parameter' => $data,
+            'clear' => (int) $clear,
+            'schedule' => (int) $schedule,
+            'confirm' => (int) $confirm,
+        ]);
+    }
+
+    /**
+     * 获取设备最新属性数据
+     */
+    public function getDeviceStatus(string $deviceId, array $props): array
+    {
+        $result = $this->get('/api/deviceStatus/v1/getDeviceStatus', [
+            'deviceCode' => $deviceId,
+            'prop' => implode(",", $props),
+        ]);
+
+        if (data_get($result, 'success') !== true) {
+            throw new RuntimeException(data_get($result, 'msg', '出错啦！'));
+        }
+
+        return $result['data'];
+    }
+
+    public function get(string $url, array $query = []): array
+    {
+        return $this->request('GET', $url, [
+            'query' => $query,
+        ]);
+    }
+
+    /**
+     * @param string $url
+     * @param array $data
+     * @return array
+     */
+    public function post(string $url, array $data = []): array
+    {
+        return $this->request('POST', $url, [
+            'json' => $data,
+        ]);
+    }
+
+    /**
+     * @param string $method
+     * @param string $url
+     * @param array $options
+     * @return array
+     *
+     * @throws \Illuminate\Http\Client\RequestException
+     * @throws \RuntimeException
+     */
+    public function request(string $method, string $url, array $options = []): array
+    {
+        $nonce = $this->nonce();
+
+        $timestamp = now()->getTimestampMs();
+
+        /** @var \Illuminate\Http\Client\Response */
+        $response = Http::withHeaders([
+            'Content-Type' => 'application/json',
+            'api-key' => $this->apiKey,
+            'Nonce' => $nonce,
+            'Timestamp' => $timestamp,
+            'Signature' => $this->sign(compact('nonce', 'timestamp')),
+        ])->baseUrl(self::ENDPOINT_URL)->send($method, $url, $options);
+
+        return $response->throw()->json();
+    }
+
+    /**
+     * @param array $data
+     * @return string
+     */
+    protected function sign(array $data): string
+    {
+        return sha1(
+            sprintf(
+                '%s%s%s',
+                $data['nonce'] ?? '',
+                $data['timestamp'] ?? '',
+                $this->apiSecret
+            )
+        );
+    }
+
+    protected function nonce(): string
+    {
+        $nonce = '';
+
+        for ($i = 0; $i < 8; $i++) {
+            $nonce .= mt_rand(0, 9);
+        }
+
+        return $nonce;
+    }
+}

app/Models/Device.php

@@ -2,13 +2,14 @@
 
 namespace App\Models;
 
-use App\Enums\DeviceType;
 use App\Enums\DeviceStatus;
-use EloquentFilter\Filterable;
+use App\Enums\DeviceType;
-use Illuminate\Database\Eloquent\Model;
 use Dcat\Admin\Traits\HasDateTimeFormatter;
-use Illuminate\Database\Eloquent\Relations\BelongsTo;
+use EloquentFilter\Filterable;
+use Illuminate\Contracts\Database\Eloquent\Builder;
 use Illuminate\Database\Eloquent\Factories\HasFactory;
+use Illuminate\Database\Eloquent\Model;
+use Illuminate\Database\Eloquent\Relations\BelongsTo;
 use Peidikeji\Keywords\Models\Keywords;
 
 class Device extends Model
@@ -41,6 +42,11 @@ class Device extends Model
         'supplier',
     ];
 
+    public function scopeSupplierBy(Builder $query, string $supplier): void
+    {
+        $query->whereHas('supplier', fn ($query) => $query->where('supplier_key', $supplier));
+    }
+
     public function base()
     {
         return $this->belongsTo(AgriculturalBase::class, 'agricultural_base_id');
@@ -60,4 +66,14 @@ class Device extends Model
     {
         return $this->belongsTo(Keywords::class, 'supplier_key', 'key');
     }
+
+    public function isTypeSoil(): bool
+    {
+        return $this->type === DeviceType::Soil;
+    }
+
+    public function isTypeMeteorological(): bool
+    {
+        return $this->type === DeviceType::Meteorological;
+    }
 }

app/Models/MeteorologicalMonitoringDailyLog.php

@@ -10,6 +10,15 @@ class MeteorologicalMonitoringDailyLog extends Model
 {
     use HasFactory;
 
+    const WIND_DIRECTION_NORTH = 0;
+    const WIND_DIRECTION_NORTHEAST = 1;
+    const WIND_DIRECTION_EAST = 2;
+    const WIND_DIRECTION_SOUTHEAST = 3;
+    const WIND_DIRECTION_SOUTH = 4;
+    const WIND_DIRECTION_SOUTHWEST = 5;
+    const WIND_DIRECTION_WEST = 6;
+    const WIND_DIRECTION_NORTHWEST = 7;
+
     protected $casts = [
         'wind_direction' => WindDirection::class,
         'monitored_at' => 'date',

app/Providers/AppServiceProvider.php

@@ -6,6 +6,7 @@ use App\Services\LinkosService;
 use Illuminate\Support\Facades\Schema;
 use Illuminate\Support\ServiceProvider;
 use Illuminate\Database\Eloquent\Relations\Relation;
+use App\Iot\Linkos\HttpClient as LinkosHttpClient;
 
 class AppServiceProvider extends ServiceProvider
 {
@@ -22,6 +23,8 @@ class AppServiceProvider extends ServiceProvider
 
             return new LinkosService($config['key'] ?? '', $config['secret'] ?? '');
         });
+
+        $this->registerLinkos();
     }
 
     /**
@@ -48,4 +51,14 @@ class AppServiceProvider extends ServiceProvider
         // ]);
 
     }
+
+    protected function registerLinkos(): void
+    {
+        $this->app->singleton(LinkosHttpClient::class, function ($app) {
+            return new LinkosHttpClient(
+                (string) $app['config']->get('services.linkos.key'),
+                (string) $app['config']->get('services.linkos.secret')
+            );
+        });
+    }
 }

app/Services/DeviceLogService.php

@@ -0,0 +1,590 @@
+<?php
+
+namespace App\Services;
+
+use App\Enums\DeviceType;
+use App\Iot\Linkos\HttpClient as LinkosHttpClient;
+use App\Models\Device;
+use App\Models\LinkosDeviceLog;
+use App\Models\MeteorologicalMonitoringDailyLog;
+use App\Models\MeteorologicalMonitoringLog;
+use App\Models\SoilMonitoringDailyLog;
+use App\Models\SoilMonitoringLog;
+use App\Models\WaterQualityMonitoringDailyLog;
+use App\Models\WaterQualityMonitoringLog;
+use Illuminate\Support\Arr;
+use Illuminate\Support\Carbon;
+
+class DeviceLogService
+{
+    /**
+     * 同步设备历史流水
+     */
+    public function sync(Device $device, Carbon $start, Carbon $end): void
+    {
+        switch ($device->supplier?->key) {
+            case 'linkos':
+                $this->syncLinkosDeviceLogs($device, $start, $end);
+                break;
+            case 'biang':
+                // $this->syncLinkosDeviceLogs($device, $start, $end);
+                break;
+        }
+    }
+
+    /**
+     * 同步 Linkos 设备历史流水
+     */
+    protected function syncLinkosDeviceLogs(Device $device, Carbon $start, Carbon $end): void
+    {
+        /** @var \App\Iot\Linkos\HttpClient */
+        $httpClient = app(LinkosHttpClient::class);
+
+        $page = 1;
+
+        $perPage = 50;
+
+        do {
+            $data = $httpClient->deviceFlowList(
+                $device->sn, $start, $end, $page, $perPage
+            );
+
+            $countResults = count($data['content']);
+
+            if ($countResults === 0) {
+                break;
+            }
+
+            foreach ($data['content'] as $item) {
+                if (! isset($item['data'])) {
+                    continue;
+                }
+
+                // 如果多合一气象监测器包含土壤监控时，需过滤掉气象监控的数据
+                if ($device->isTypeSoil() && Arr::hasAny($item['data'], [
+                    'current_rainfall',
+                    'day_rainfall',
+                    'accumulate_rainfall',
+                    'moment_rainfall',
+                    'pm10_concentration',
+                    'pm25_concentration',
+                    'box_illumination',
+                    'box_pressure',
+                    'box_carbon',
+                    'box_temperature',
+                    'box_humidity',
+                    'box_noise',
+                    'wind_degree',
+                    'wind_direction',
+                    'wind_power',
+                    'wind_speed',
+                ])) {
+                    continue;
+                }
+
+                LinkosDeviceLog::firstOrCreate([
+                    'device_id' => $device->sn,
+                    'reported_at' => $item['createTime'],
+                ], [
+                    'device_unit' => $device->model,
+                    'data' => empty($item['data']) ? (new \stdClass) : $item['data'],
+                ]);
+            }
+
+            unset($data);
+
+            $page++;
+        } while ($countResults === $perPage);
+    }
+
+    /**
+     * 创建 linkos 设备报告
+     */
+    public function createReportToLinkosDevice(Device $device, Carbon $time): void
+    {
+        switch ($device->type) {
+            case DeviceType::Soil:
+                $this->createReportToLinkosSoilDevice($device, $time);
+                break;
+
+            case DeviceType::Meteorological:
+                $this->createReportToLinkosMeteorologicalDevice($device, $time);
+                break;
+
+            case DeviceType::WaterQuality:
+                $this->createReportToLinkosWaterQualityDevice($device, $time);
+                break;
+        }
+    }
+
+    /**
+     * 创建 linkos 土壤设备报告
+     */
+    protected function createReportToLinkosSoilDevice(Device $device, Carbon $time): void
+    {
+        $reportedAt = $time->copy()->startOfHour();
+
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $logs = LinkosDeviceLog::where('device_id', $device->sn)
+            ->whereBetween('reported_at', [$reportedAt, $reportedAt->copy()->endOfHour()])
+            ->oldest('reported_at')
+            ->get();
+
+        if ($logs->isEmpty()) {
+            return;
+        }
+
+        $attributes = $logs->reduce(function (array $attributes, LinkosDeviceLog $log) {
+            if (is_array($data = $log->data)) {
+                foreach ($data as $k => $v) {
+                    $attribute = match ($k) {
+                        'nitrogen_content' => 'n',
+                        'potassium_content' => 'k',
+                        'phosphorus_content' => 'p',
+                        'electroconductibility' => 'conductivity',
+                        'temperature' => 'temperature',
+                        'moisture_content' => 'moisture',
+                        'conductivity' => 'conductivity',
+                        'soil_humidity' => 'humidity',
+                        'soil_temperature' => 'temperature',
+                        default => null,
+                    };
+
+                    if ($attribute) {
+                        $attributes[$attribute] = $v;
+                    }
+                }
+            }
+
+            return $attributes;
+        }, []);
+
+        $soilReport = SoilMonitoringLog::where([
+            'device_id' => $device->id,
+            'monitored_at' => $reportedAt,
+        ])->first();
+
+        if ($soilReport === null) {
+            $lastSoilReport = SoilMonitoringLog::where([
+                'device_id' => $device->id,
+                'monitored_at' => $reportedAt->copy()->subHour(),
+            ])->first();
+
+            $soilReport = $lastSoilReport?->replicate() ?: new SoilMonitoringLog();
+
+            $soilReport->fill([
+                'device_id' => $device->id,
+                'monitored_at' => $reportedAt,
+                'agricultural_base_id' => $device->agricultural_base_id,
+            ]);
+        }
+
+        $soilReport->fill($attributes)->save();
+    }
+
+    /**
+     * 创建 linkos 气象设备报告
+     */
+    protected function createReportToLinkosMeteorologicalDevice(Device $device, Carbon $time): void
+    {
+        $reportedAt = $time->copy()->startOfHour();
+
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $logs = LinkosDeviceLog::where('device_id', $device->sn)
+            ->whereBetween('reported_at', [$reportedAt, $reportedAt->copy()->endOfHour()])
+            ->oldest('reported_at')
+            ->get();
+
+        if ($logs->isEmpty()) {
+            return;
+        }
+
+        $attributes = $logs->reduce(function (array $attributes, LinkosDeviceLog $log) {
+            if (is_array($data = $log->data)) {
+                foreach ($data as $k => $v) {
+                    $attribute = match ($k) {
+                        // 'day_rainfall' => 'yesterday_rainfall',
+                        'current_rainfall' => 'current_rainfall',
+                        'accumulate_rainfall' => 'accumulated_rainfall',
+                        'moment_rainfall' => 'moment_rainfall',
+                        'pm10_concentration' => 'pm10',
+                        'pm25_concentration' => 'pm25',
+                        'box_illumination' => 'illumination',
+                        'box_pressure' => 'air_pressure',
+                        'box_carbon' => 'co2',
+                        'box_temperature' => 'air_temperature',
+                        'box_humidity' => 'air_humidity',
+                        'box_noise' => 'noise',
+                        'wind_degree' => 'wind_degree',
+                        'wind_direction' => 'wind_direction',
+                        'wind_power' => 'wind_power',
+                        'wind_speed' => 'wind_speed',
+                        default => null,
+                    };
+
+                    if ($attribute) {
+                        $attributes[$attribute] = $v;
+                    }
+                }
+            }
+
+            return $attributes;
+        }, []);
+
+        $meteorologicalReport = MeteorologicalMonitoringLog::where([
+            'device_id' => $device->id,
+            'monitored_at' => $reportedAt,
+        ])->first();
+
+        if ($meteorologicalReport === null) {
+            $lastMeteorologicalReport = MeteorologicalMonitoringLog::where([
+                'device_id' => $device->id,
+                'monitored_at' => $reportedAt->copy()->subHour(),
+            ])->first();
+
+            $meteorologicalReport = $lastMeteorologicalReport?->replicate() ?: new MeteorologicalMonitoringLog();
+
+            $meteorologicalReport->fill([
+                'device_id' => $device->id,
+                'monitored_at' => $reportedAt,
+                'agricultural_base_id' => $device->agricultural_base_id,
+            ]);
+        }
+
+        $meteorologicalReport->fill($attributes)->save();
+    }
+
+    /**
+     * 创建 linkos 水质设备报告
+     */
+    protected function createReportToLinkosWaterQualityDevice(Device $device, Carbon $time): void
+    {
+        $reportedAt = $time->copy()->startOfHour();
+
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $logs = LinkosDeviceLog::where('device_id', $device->sn)
+            ->whereBetween('reported_at', [$reportedAt, $reportedAt->copy()->endOfHour()])
+            ->oldest('reported_at')
+            ->get();
+
+        if ($logs->isEmpty()) {
+            return;
+        }
+
+        $attributes = $logs->reduce(function (array $attributes, LinkosDeviceLog $log) {
+            if (is_array($data = $log->data)) {
+                foreach ($data as $k => $v) {
+                    $attribute = match ($k) {
+                        'chlorine' => 'chlorine',
+                        'conductivity' => 'conductivity',
+                        'oxygen' => 'oxygen',
+                        'ph' => 'ph',
+                        'temp' => 'temperature',
+                        'turbidity' => 'turbidity',
+                        default => null,
+                    };
+
+                    if ($attribute) {
+                        $attributes[$attribute] = $v;
+                    }
+                }
+            }
+
+            return $attributes;
+        }, []);
+
+        $waterQualityReport = WaterQualityMonitoringLog::where([
+            'device_id' => $device->id,
+            'monitored_at' => $reportedAt,
+        ])->first();
+
+        if ($waterQualityReport === null) {
+            $lastWaterQualityReport = WaterQualityMonitoringLog::where([
+                'device_id' => $device->id,
+                'monitored_at' => $reportedAt->copy()->subHour(),
+            ])->first();
+
+            $waterQualityReport = $lastWaterQualityReport?->replicate() ?: new WaterQualityMonitoringLog();
+
+            $waterQualityReport->fill([
+                'device_id' => $device->id,
+                'monitored_at' => $reportedAt,
+                'agricultural_base_id' => $device->agricultural_base_id,
+            ]);
+        }
+
+        $waterQualityReport->fill($attributes)->save();
+    }
+
+    /**
+     * 创建 linkos 设备每日报告
+     */
+    public function createDailyReportToLinkosDevice(Device $device, Carbon $time): void
+    {
+        switch ($device->type) {
+            case DeviceType::Meteorological:
+                $this->createDailyReportToLinkosMeteorologicalDevice($device, $time);
+                break;
+
+            case DeviceType::WaterQuality:
+                $this->createDailyReportToLinkosWaterQualityDevice($device, $time);
+                break;
+
+            case DeviceType::Soil:
+                $this->createDailyReportToLinkosSoilDevice($device, $time);
+                break;
+        }
+    }
+
+    /**
+     * 创建 linkos 土壤设备每日报告
+     */
+    protected function createDailyReportToLinkosSoilDevice(Device $device, Carbon $date): void
+    {
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $soilReports = SoilMonitoringLog::where('device_id', $device->id)
+            ->whereDate('monitored_at', $date)
+            ->oldest('monitored_at')
+            ->get();
+
+        if ($soilReports->isEmpty()) {
+            return;
+        }
+
+        $attributes = value(function ($soilReports) {
+            $data = [
+                'n' => ['sum' => 0, 'count' => 0],
+                'p' => ['sum' => 0, 'count' => 0],
+                'k' => ['sum' => 0, 'count' => 0],
+                'conductivity' => ['sum' => 0, 'count' => 0],
+                'temperature' => ['sum' => 0, 'count' => 0],
+                'humidity' => ['sum' => 0, 'count' => 0],
+                'moisture' => ['sum' => 0, 'count' => 0],
+            ];
+
+            foreach ($soilReports as $soilReport) {
+                foreach ($data as $k => $item) {
+                    if (is_null($v = $soilReport->{$k})) {
+                        continue;
+                    }
+
+                    $item['sum'] = bcadd($item['sum'], $v, 2);
+                    $item['count']++;
+
+                    $data[$k] = $item;
+                }
+            }
+
+            $attributes = [];
+
+            foreach ($data as $key => $item) {
+                $attributes[$key] = $item['count'] > 0 ? round(bcdiv($item['sum'], $item['count'], 2), 2) : null;
+            }
+
+            return $attributes;
+        }, $soilReports);
+
+        /** @var \App\Models\SoilDailyReport */
+        $soilDailyReport = SoilMonitoringDailyLog::firstOrNew([
+            'device_id' => $device->id,
+            'monitored_at' => $date->format('Y-m-d'),
+        ], [
+            'agricultural_base_id' => $device->agricultural_base_id,
+        ]);
+
+        $soilDailyReport->fill($attributes)->save();
+    }
+
+    /**
+     * 创建 linkos 气象设备每日报告
+     */
+    protected function createDailyReportToLinkosMeteorologicalDevice(Device $device, Carbon $date): void
+    {
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $meteorologicalReports = MeteorologicalMonitoringLog::where('device_id', $device->id)
+            ->whereDate('monitored_at', $date)
+            ->oldest('monitored_at')
+            ->get();
+
+        if ($meteorologicalReports->isEmpty()) {
+            return;
+        }
+
+        $attributes = value(function ($meteorologicalReports) {
+            $data = [
+                'today_rainfall' => ['sum' => 0, 'count' => 0],
+                // 'yesterday_rainfall' => ['sum' => 0, 'count' => 0],
+                'accumulated_rainfall' => ['sum' => 0, 'count' => 0],
+                'moment_rainfall' => ['sum' => 0, 'count' => 0],
+                'pm10' => ['sum' => 0, 'count' => 0],
+                'pm25' => ['sum' => 0, 'count' => 0],
+                'illumination' => ['sum' => 0, 'count' => 0],
+                'air_pressure' => ['sum' => 0, 'count' => 0],
+                'co2' => ['sum' => 0, 'count' => 0],
+                'air_temperature' => ['sum' => 0, 'count' => 0],
+                'air_humidity' => ['sum' => 0, 'count' => 0],
+                'noise' => ['sum' => 0, 'count' => 0],
+                'wind_samples' => [],
+            ];
+
+            foreach ($meteorologicalReports as $meteorologicalReport) {
+                foreach ($data as $k => $item) {
+                    if ($k === 'wind_samples') {
+                        if (is_null($meteorologicalReport->wind_degree) || is_null($meteorologicalReport->wind_speed)) {
+                            continue;
+                        }
+
+                        $item[] = [
+                            'wind_degree' => $meteorologicalReport->wind_degree,     // 风向度数
+                            'wind_speed' => $meteorologicalReport->wind_speed,       // 风速
+                        ];
+                    } elseif (! is_null($v = $meteorologicalReport->{$k})) {
+                        $item['sum'] = bcadd($item['sum'], $v, 2);
+                        $item['count']++;
+                    }
+
+                    $data[$k] = $item;
+                }
+            }
+
+            $attributes = [];
+
+            foreach ($data as $key => $item) {
+                switch ($key) {
+                    case 'wind_samples':
+                        $attributes['wind_degree'] = value(function (array $windSamples) {
+                            if (empty($windSamples)) {
+                                return null;
+                            }
+
+                            $x = 0;
+                            $y = 0;
+
+                            foreach ($windSamples as $sample) {
+                                if ($sample['wind_degree'] == 0 && $sample['wind_speed'] == 0) {
+                                    continue;
+                                }
+
+                                // 角度转弧度
+                                $radian = deg2rad($sample['wind_degree']);
+
+                                // $x += $sample['wind_speed'] * sin($radian);
+                                // $y += $sample['wind_speed'] * cos($radian);
+                                $x += sin($radian);
+                                $y += cos($radian);
+                            }
+
+                            $degree = round(rad2deg(atan2($y, $x)));
+
+                            if (($x > 0 || $x < 0) && $y < 0) {
+                                $degree += 180;
+                            } elseif ($x < 0 && $y > 0) {
+                                $degree += 360;
+                            }
+
+                            return $degree;
+                        }, $item);
+
+                        $attributes['wind_direction'] = value(function ($windDegree) {
+                            if (is_null($windDegree)) {
+                                return null;
+                            }
+
+                            if ($windDegree >= 22.5 && $windDegree < 67.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_NORTHEAST;
+                            } elseif ($windDegree >= 67.5 && $windDegree < 112.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_EAST;
+                            } elseif ($windDegree >= 112.5 && $windDegree < 157.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_SOUTHEAST;
+                            } elseif ($windDegree >= 157.5 && $windDegree < 202.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_SOUTH;
+                            } elseif ($windDegree >= 202.5 && $windDegree < 247.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_SOUTHWEST;
+                            } elseif ($windDegree >= 247.5 && $windDegree < 292.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_WEST;
+                            } elseif ($windDegree >= 292.5 && $windDegree < 337.5) {
+                                return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_NORTHWEST;
+                            }
+
+                            return MeteorologicalMonitoringDailyLog::WIND_DIRECTION_NORTH;
+                        }, $attributes['wind_degree']);
+                        break;
+
+                    default:
+                        $attributes[$key] = $item['count'] > 0 ? round(bcdiv($item['sum'], $item['count'], 2), 2) : null;
+                        break;
+                }
+            }
+
+            return $attributes;
+        }, $meteorologicalReports);
+
+        /** @var \App\Models\MeteorologicalMonitoringDailyLog */
+        $meteorologicalDailyReport = MeteorologicalMonitoringDailyLog::firstOrNew([
+            'device_id' => $device->id,
+            'monitored_at' => $date,
+        ], [
+            'agricultural_base_id' => $device->agricultural_base_id,
+        ]);
+
+        $meteorologicalDailyReport->fill($attributes)->save();
+    }
+
+    /**
+     * 创建 linkos 水质设备每日报告
+     */
+    protected function createDailyReportToLinkosWaterQualityDevice(Device $device, Carbon $date): void
+    {
+        /** @var \Illuminate\Database\Eloquent\Collection */
+        $waterQualityReports = WaterQualityMonitoringLog::where('device_id', $device->id)
+            ->whereDate('monitored_at', $date)
+            ->oldest('monitored_at')
+            ->get();
+
+        if ($waterQualityReports->isEmpty()) {
+            return;
+        }
+
+        $attributes = value(function ($waterQualityReports) {
+            $data = [
+                'chlorine' => ['sum' => 0, 'count' => 0],
+                'conductivity' => ['sum' => 0, 'count' => 0],
+                'oxygen' => ['sum' => 0, 'count' => 0],
+                'ph' => ['sum' => 0, 'count' => 0],
+                'temperature' => ['sum' => 0, 'count' => 0],
+                'turbidity' => ['sum' => 0, 'count' => 0],
+            ];
+
+            foreach ($waterQualityReports as $waterQualityReport) {
+                foreach ($data as $k => $item) {
+                    if (is_null($v = $waterQualityReport->{$k})) {
+                        continue;
+                    }
+
+                    $item['sum'] = bcadd($item['sum'], $v, 2);
+                    $item['count']++;
+
+                    $data[$k] = $item;
+                }
+            }
+
+            $attributes = [];
+
+            foreach ($data as $key => $item) {
+                $attributes[$key] = $item['count'] > 0 ? round(bcdiv($item['sum'], $item['count'], 2), 2) : null;
+            }
+
+            return $attributes;
+        }, $waterQualityReports);
+
+        /** @var \App\Models\WaterQualityMonitoringDailyLog */
+        $WaterQualityMonitoringDailyLog = WaterQualityMonitoringDailyLog::firstOrNew([
+            'device_id' => $device->id,
+            'monitored_at' => $date->format('Y-m-d'),
+        ], [
+            'agricultural_base_id' => $device->agricultural_base_id,
+        ]);
+
+        $WaterQualityMonitoringDailyLog->fill($attributes)->save();
+    }
+}