Merge pull request #50 from 7Last/develop

Develop

.github/workflows/pre_merge.yaml

@@ -2,29 +2,39 @@ name: Pre merge checks
 
 on:
   pull_request:
-    branches: [main]
-    types: [opened, reopened, review_requested, edited, synchronize]
+    branches: [main,develop]
+    types: [opened, reopened, edited, synchronize]
 
   workflow_dispatch:
 
 jobs:
   python_lints:
     runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./simulator
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
 
       - name: Install python
-        uses: actions/setup-python@v3
+        uses: actions/setup-python@v5
         with:
           python-version: 3.11
 
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install -r requirements.txt
+          pip install -r ./simulator/requirements.txt
 
       - name: Run ruff lint
-        run: ruff check --output-format=github
+        run: ruff check --fix --output-format=github ./simulator
+
+      - name: Run unit tests with coverage
+        run: |
+          python3 -m pip install coverage
+          coverage run --branch --source="./simulator/src" -m unittest discover -s ./simulator/test -v
+          coverage xml
+
+      - name: Get Cover
+        uses: orgoro/coverage@v3.1
+        with:
+          coverageFile: ./coverage.xml
+          token: ${{ secrets.TOKEN }}
+          thresholdAll: 0.8

.gitignore

@@ -2,5 +2,8 @@
 /.idea/
 ./logs/
 /.venv_samples/
-/volumes/
-**/__pycache__/**
+/redpanda/volumes/
+**/__pycache__/**
+.coverage
+.DS_Store
+/volumes/

README.md

@@ -1,23 +1,35 @@
 # Running locally
+
 This document provides instructions on how to run the project locally.
 
 Two profiles are provided in the `docker-compose.yaml` file:
+
 - `local`: for local test and debug purposes. Runs the Kafka stack in `docker` containers,
-but does not build the `simulator` image, which is expected to be run manually;
+  but does not build the `simulator` image, which is expected to be run manually;
 - `release`: builds also the `simulator` image, which is then run in the same network as the Kafka stack.
 
 ## Running with `local` profile
+
 Running the Kafka stack:
+
 ```bash
 docker-compose --profile local up -d
 ```
 
 Give all permissions to the `./volumes/` folder that has been created:
+
 ```bash
 chmod -R 777 ./volumes
 ```
 
+then **run all containers that are turned off** and then run:
+
+```bash
+curl "localhost:8083/connectors/" -H 'Content-Type: application/json' -d @./redpanda/connectors/configs/clickhouse.json
+```
+
 Running the simulator:
+
 - Create a virtual environment:
 
 ```bash
@@ -49,6 +61,13 @@ python3 ./simulator/main.py
 ```
 
 ## Running with `release` profile
+
 ```bash
 docker-compose --profile release up -d
 ```
+
+## Running with `test` profile (integration test)
+
+```bash
+docker-compose -f docker-compose-test.yaml --profile test up -d --build
+```

clickhouse/init/10_charging_efficiency.sql

@@ -0,0 +1,10 @@
+CREATE TABLE sensors.charging_efficiency
+(
+    sensor_uuid      UUID,
+    sensor_names     Array(String),
+    group_name       String,
+    timestamp        DateTime64,
+    efficiency_rate  Float64,
+    utilization_rate Float64
+) ENGINE = MergeTree()
+      ORDER BY (sensor_uuid, timestamp);

clickhouse/init/1_temperature.sql

@@ -1,78 +1,71 @@
-CREATE TABLE sensors.temperature_kafka
-(
-    data String
-) ENGINE = Kafka('redpanda:9092', 'temperature', 'ch_group_1', 'JSONAsString');
-
-CREATE TABLE sensors.temperatures
+CREATE TABLE sensors.temperature
 (
     sensor_uuid UUID,
     sensor_name String,
+    group_name  Nullable(String) default null,
     timestamp   DateTime64,
     value       Float32,
     latitude    Float64,
     longitude   Float64
 ) ENGINE = MergeTree()
       ORDER BY (sensor_uuid, timestamp);
 
-CREATE MATERIALIZED VIEW sensors.temperature_topic_mv TO sensors.temperatures as
-SELECT JSONExtractString(data, 'sensor_name')                             AS sensor_name,
-       toUUID(JSONExtractString(data, 'sensor_uuid'))                     AS sensor_uuid,
-       parseDateTime64BestEffort(JSONExtractString(data, 'timestamp')) AS timestamp,
-       JSONExtractFloat(data, 'value')                                    AS value,
-       JSONExtractFloat(data, 'latitude')                                 AS latitude,
-       JSONExtractFloat(data, 'longitude')                                AS longitude
-FROM sensors.temperature_kafka;
-
 -- 5m averages
-CREATE TABLE sensors.temperatures_5m
+CREATE TABLE sensors.temperature_5m
 (
     sensor_name         String,
+    group_name Nullable(String)       default null,
     date                DateTime64,
     avg_temperature     Float32,
     insertion_timestamp DateTime64(6) default now64()
 ) ENGINE = MergeTree()
       ORDER BY (sensor_name, date);
 
-CREATE MATERIALIZED VIEW sensors.temperatures_5m_mv
-            TO sensors.temperatures_5m AS
+CREATE MATERIALIZED VIEW sensors.temperature_5m_mv
+    TO sensors.temperature_5m AS
 SELECT sensor_name,
+       group_name,
        toStartOfFiveMinutes(timestamp) AS date,
        avg(value)                      AS avg_temperature
-FROM sensors.temperatures
-GROUP BY sensor_name, date;
+FROM sensors.temperature
+GROUP BY sensor_name, group_name, date;
 
--- Weekly temperatures
-CREATE TABLE sensors.temperatures_weekly
+-- Weekly temperature
+CREATE TABLE sensors.temperature_weekly
 (
     sensor_name         String,
+    group_name Nullable(String)       default null,
     date                DateTime64,
     avg_temperature     Float32,
     insertion_timestamp DateTime64(6) default now64()
 ) ENGINE = MergeTree()
       ORDER BY (sensor_name, date);
 
-CREATE MATERIALIZED VIEW sensors.temperatures_weekly_mv
-    TO sensors.temperatures_weekly AS
+CREATE MATERIALIZED VIEW sensors.temperature_weekly_mv
+    TO sensors.temperature_weekly AS
 SELECT sensor_name,
+       group_name,
        toStartOfWeek(timestamp) AS date,
        avg(value)               AS avg_temperature
-FROM sensors.temperatures
-GROUP BY sensor_name, date;
+FROM sensors.temperature
+GROUP BY sensor_name, group_name, date;
 
--- Daily temperatures
-CREATE TABLE sensors.temperatures_daily
+-- Daily temperature
+CREATE TABLE sensors.temperature_daily
 (
     sensor_name         String,
+    group_name Nullable(String)       default null,
     date                Date,
     avg_temperature     Float32,
     insertion_timestamp DateTime64(6) default now64()
 ) ENGINE = MergeTree()
       ORDER BY (sensor_name, date);
 
-CREATE MATERIALIZED VIEW sensors.temperatures_daily_mv
-    TO sensors.temperatures_daily AS
+CREATE MATERIALIZED VIEW sensors.temperature_daily_mv
+    TO sensors.temperature_daily AS
 SELECT sensor_name,
+       group_name,
        toStartOfDay(timestamp) AS date,
        avg(value)              AS avg_temperature
-FROM sensors.temperatures
-GROUP BY sensor_name, date;
+FROM sensors.temperature
+GROUP BY sensor_name, group_name, date;

clickhouse/init/2_traffic.sql

@@ -1,12 +1,8 @@
-CREATE TABLE sensors.traffic_kafka
-(
-    data String
-) ENGINE = Kafka('redpanda:9092', 'traffic', 'ch_group_1', 'JSONAsString');
-
 CREATE TABLE sensors.traffic
 (
     sensor_uuid UUID,
     sensor_name String,
+    group_name  Nullable(String) default null,
     timestamp   DateTime64,
     latitude    Float64,
     longitude   Float64,
@@ -15,16 +11,6 @@ CREATE TABLE sensors.traffic
 ) ENGINE = MergeTree()
       ORDER BY (sensor_uuid, timestamp);
 
-CREATE MATERIALIZED VIEW sensors.traffic_topic_mv TO sensors.traffic as
-SELECT JSONExtractString(data, 'sensor_name')                AS sensor_name,
-       toUUID(JSONExtractString(data, 'sensor_uuid'))        AS sensor_uuid,
-       parseDateTime64BestEffort(JSONExtractString(data, 'timestamp')) AS timestamp,
-       JSONExtractFloat(data, 'vehicles')                    AS vehicles,
-       JSONExtractFloat(data, 'avg_speed')                   AS avg_speed,
-       JSONExtractFloat(data, 'latitude')                    AS latitude,
-       JSONExtractFloat(data, 'longitude')                   AS longitude
-FROM sensors.traffic_kafka;
-
 -- 1h traffic
 CREATE TABLE sensors.traffic_1h
 (

clickhouse/init/3_recycling_point.sql

@@ -1,41 +1,11 @@
-CREATE TABLE sensors.recycling_point_kafka
-(
-    data String
-) ENGINE = Kafka('redpanda:9092', 'recycling_point', 'ch_group_1', 'JSONAsString');
-
 CREATE TABLE sensors.recycling_point
 (
     sensor_uuid   UUID,
     sensor_name   String,
+    group_name    Nullable(String) default null,
     timestamp     DateTime64,
     latitude      Float64,
     longitude     Float64,
-    filling_value Float32
+    filling       Float32
 ) ENGINE = MergeTree()
       ORDER BY (sensor_uuid, timestamp);
-
-CREATE MATERIALIZED VIEW sensors.recycling_point_topic_mv TO sensors.recycling_point as
-SELECT JSONExtractString(data, 'sensor_name')                          AS sensor_name,
-       toUUID(JSONExtractString(data, 'sensor_uuid'))                  AS sensor_uuid,
-       parseDateTime64BestEffort(JSONExtractString(data, 'timestamp')) AS timestamp,
-       JSONExtractFloat(data, 'filling_value')                         AS filling_value,
-       JSONExtractFloat(data, 'latitude')                              AS latitude,
-       JSONExtractFloat(data, 'longitude')                             AS longitude
-FROM sensors.recycling_point_kafka;
-
-CREATE TABLE sensors.recycling_point_5m
-(
-    sensor_name         String,
-    date                DateTime64,
-    avg_filling_value   Float32,
-    insertion_timestamp DateTime64(6) default now64()
-) ENGINE = MergeTree()
-      ORDER BY (sensor_name, date);
-
-CREATE MATERIALIZED VIEW sensors.recycling_point_5m_mv
-    TO sensors.recycling_point_5m AS
-SELECT sensor_name,
-       toStartOfFiveMinutes(timestamp) AS date,
-       avg(filling_value)              AS avg_filling_value
-from sensors.recycling_point
-GROUP BY sensor_name, date;

clickhouse/init/4_humidity.sql

@@ -0,0 +1,65 @@
+CREATE TABLE sensors.humidity
+(
+    sensor_uuid UUID,
+    sensor_name String,
+    group_name  Nullable(String) default null,
+    timestamp   DateTime64,
+    value       Float32,
+    latitude    Float64,
+    longitude   Float64
+) ENGINE = MergeTree()
+      ORDER BY (sensor_uuid, timestamp);
+
+-- 5m averages
+CREATE TABLE sensors.humidity_5m
+(
+    sensor_name         String,
+    date                DateTime64,
+    avg_humidity        Float32,
+    insertion_timestamp DateTime64(6) default now64()
+) ENGINE = MergeTree()
+      ORDER BY (sensor_name, date);
+
+CREATE MATERIALIZED VIEW sensors.humidity_5m_mv
+            TO sensors.humidity_5m AS
+SELECT sensor_name,
+       toStartOfFiveMinutes(timestamp) AS date,
+       avg(value)                      AS avg_humidity
+FROM sensors.humidity
+GROUP BY sensor_name, date;
+
+-- Weekly humidity
+CREATE TABLE sensors.humidity_weekly
+(
+    sensor_name         String,
+    date                DateTime64,
+    avg_humidity        Float32,
+    insertion_timestamp DateTime64(6) default now64()
+) ENGINE = MergeTree()
+      ORDER BY (sensor_name, date);
+
+CREATE MATERIALIZED VIEW sensors.humidity_weekly_mv
+    TO sensors.humidity_weekly AS
+SELECT sensor_name,
+       toStartOfWeek(timestamp) AS date,
+       avg(value)               AS avg_humidity
+FROM sensors.humidity
+GROUP BY sensor_name, date;
+
+-- Daily humidity
+CREATE TABLE sensors.humidity_daily
+(
+    sensor_name         String,
+    date                Date,
+    avg_humidity        Float32,
+    insertion_timestamp DateTime64(6) default now64()
+) ENGINE = MergeTree()
+      ORDER BY (sensor_name, date);
+
+CREATE MATERIALIZED VIEW sensors.humidity_daily_mv
+    TO sensors.humidity_daily AS
+SELECT sensor_name,
+       toStartOfDay(timestamp) AS date,
+       avg(value)              AS avg_humidity
+FROM sensors.humidity
+GROUP BY sensor_name, date;

clickhouse/init/5_air_quality.sql

@@ -0,0 +1,15 @@
+CREATE TABLE sensors.air_quality
+(
+    sensor_uuid UUID,
+    sensor_name String,
+    group_name  Nullable(String) default null,
+    timestamp   DateTime64,
+    latitude    Float64,
+    longitude   Float64,
+    pm25        Float32,
+    pm10        Float32,
+    no2         Float32,
+    o3          Float32,
+    so2         Float32
+) ENGINE = MergeTree()
+      ORDER BY (sensor_uuid, timestamp);