chore(release): v0.2.0 – CSV writer, partition_by, YAML config + autodiscovery

- add CSV writer with header and flush
- support --partition-by (device/type/none)
- add YAML config parsing (--config) with PyYAML
- implement config autodiscovery (config.sensors.yaml, dummysensors.yaml, config.yaml)
- bump version to 0.2.0

Author: SculptTechProject

.gitignore

@@ -3,6 +3,9 @@ __pycache__/
 *.py[codz]
 *$py.class
 
+# config.sensors.yaml
+config.sensors.yaml
+
 # Visual Studio code
 .vscode/
 

out/temp.jsonl

@@ -0,0 +1,5 @@
+{"ts_ms": 0, "device_id": "engine-A", "sensor_id": "temp-0", "type": "temp", "value": 24.643698021231547}
+{"ts_ms": 500, "device_id": "engine-A", "sensor_id": "temp-0", "type": "temp", "value": 23.70217328027316}
+{"ts_ms": 1000, "device_id": "engine-A", "sensor_id": "temp-0", "type": "temp", "value": 21.464851782876178}
+{"ts_ms": 1500, "device_id": "engine-A", "sensor_id": "temp-0", "type": "temp", "value": 26.817176898799882}
+{"ts_ms": 2000, "device_id": "engine-A", "sensor_id": "temp-0", "type": "temp", "value": 17.562920097753118}

out/vibration.csv

@@ -0,0 +1,6 @@
+ts_ms,device_id,sensor_id,type,value
+0,engine-A,vibration-0,vibration,0.09920709782168856
+500,engine-A,vibration-0,vibration,0.24727765238362118
+1000,engine-A,vibration-0,vibration,0.10491826983174454
+1500,engine-A,vibration-0,vibration,0.15830367289446529
+2000,engine-A,vibration-0,vibration,0.20296206525424756

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "dummysensors"
-version = "0.1.0"
+version = "0.2.0"
 description = "Dummy sensor data generator for testing IoT/ML pipelines"
 readme = "README.md"
 authors = [{ name = "Mateusz Dalke" }]

requirements.txt

@@ -2,3 +2,4 @@ pytest>=8.2
 ruff>=0.6
 build>=1.2
 hatchling>=1.25
+pyyaml>=6.0

src/dummysensors/cli.py

@@ -1,29 +1,23 @@
 from __future__ import annotations
-import argparse
-import sys
+import argparse 
 import time
-import json
 from .sensors import TemperatureSensor
 from .orchestrator import run_stream
-import os
+from .io import jsonl_writer, csv_writer
 
 def _stdout_writer():
-    def _w(sample: dict):
-        sys.stdout.write(json.dumps(sample) + "\n")
-    return _w
+    return jsonl_writer(None)
 
 def _jsonl_file_writer(path: str):
-    d = os.path.dirname(path)
-    if d:
-        os.makedirs(d, exist_ok=True)
-    f = open(path, "a", buffering=1, encoding="utf-8")
-    def _w(sample: dict):
-        f.write(json.dumps(sample) + "\n")
-    return _w
+    return jsonl_writer(path)
+
+def _csv_file_writer(path: str):
+    return csv_writer(path)
 
 def _make_writer_map(out_args: list[str] | None):
     """
-    out args format: ["temp=out/temp.jsonl", "vibration=out/vib.jsonl", "*=stdout"]
+    out args format: ["temp=out/temp.jsonl", "vibration=out/vib.csv", "*=stdout"]
+    wybór formatu po rozszerzeniu pliku: .jsonl -> JSONL, .csv -> CSV
     """
     mapping = {}
     if not out_args:
@@ -35,14 +29,16 @@ def _make_writer_map(out_args: list[str] | None):
         if dest == "stdout":
             mapping[key] = _stdout_writer()
         else:
-            mapping[key] = _jsonl_file_writer(dest)
+            if dest.lower().endswith(".csv"):
+                mapping[key] = _csv_file_writer(dest)
+            else:
+                mapping[key] = _jsonl_file_writer(dest)
     return mapping
 
 def main(argv=None):
     p = argparse.ArgumentParser(prog="dummy-sensors")
     sub = p.add_subparsers(dest="cmd", required=True)
 
-    # generate 
     gen = sub.add_parser("generate", help="single temperature stream to stdout/file")
     gen.add_argument("--rate", type=float, default=5.0)
     gen.add_argument("--duration", type=float, default=5.0)
@@ -52,13 +48,14 @@ def main(argv=None):
     gen.add_argument("--noise", type=float, default=0.5)
     gen.add_argument("--jsonl", type=str, default=None)
 
-    # run
-    run = sub.add_parser("run", help="multi-device/multi-sensor run via spec")
+    run = sub.add_parser("run", help="multi-device/multi-sensor run via spec or config")
     run.add_argument("--rate", type=float, default=5.0)
     run.add_argument("--duration", type=float, default=None)
     run.add_argument("--count", type=int, default=None)
-    run.add_argument("--spec", type=str, required=True, help='e.g. "device=A: temp*2,vibration*1; device=B: temp*3"')
+    run.add_argument("--spec", type=str, help='e.g. "device=A: temp*2,vibration*1"')
     run.add_argument("--out", action="append", help='mapping like "temp=out/temp.jsonl", "*=stdout"', default=None)
+    run.add_argument("--partition-by", choices=["none","type","device"], default="none")
+    run.add_argument("--config", type=str, help="YAML config path (overrides --spec/--out)")
 
     args = p.parse_args(argv)
 
@@ -77,8 +74,15 @@ def main(argv=None):
                 while time.perf_counter() < t_next:
                     time.sleep(min(0.002, t_next - time.perf_counter()))
     else:
-        writers = _make_writer_map(args.out)
-        run_stream(args.spec, rate_hz=args.rate, duration_s=args.duration, total_count=args.count, writer_for_type=writers)
+        # jeśli podano --config, jedziemy konfiguracją YAML (sekcja 4)
+        if args.config:
+            from .config import run_from_config
+            run_from_config(args.config)
+            return
 
-if __name__ == "__main__":
-    main()
+        # spec + out
+        writers = _make_writer_map(args.out)
+        if not args.spec:
+            p.error("Provide --spec or --config")
+        run_stream(args.spec, rate_hz=args.rate, duration_s=args.duration, total_count=args.count,
+                   writer_for_type=writers, partition_by=args.partition_by)

src/dummysensors/config.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+from typing import Callable, Dict
+from .orchestrator import run_stream
+from .io import jsonl_writer
+from .io import csv_writer
+from pathlib import Path
+
+DEFAULT_CONFIG_CANDIDATES = (
+    "config.sensors.yaml",
+    "config.sensors.yml",
+    "dummysensors.yaml",
+    "dummysensors.yml",
+)
+
+def find_config_path(start: str | None = None) -> str | None:
+    """Return first matching config file path from DEFAULT_CONFIG_CANDIDATES in `start` or CWD."""
+    base = Path(start) if start else Path.cwd()
+    for name in DEFAULT_CONFIG_CANDIDATES:
+        cand = base / name
+        if cand.exists():
+            return str(cand)
+    return None
+
+def _writer_for_decl(decl: dict):
+    fmt = str(decl.get("type", "jsonl")).lower()
+    path = decl.get("path")
+    if fmt == "csv":
+        if not path:
+            raise ValueError("CSV output requires 'path' in config.")
+        return csv_writer(str(path))
+    # default jsonl; if path is None -> stdout
+    return jsonl_writer(path)
+
+def run_from_config(path: str) -> None:
+    try:
+        import yaml  # type: ignore
+    except ImportError as e:
+        raise RuntimeError("Install PyYAML to use --config (pip install pyyaml)") from e
+
+    with open(path, "r", encoding="utf-8") as f:
+        cfg = yaml.safe_load(f) or {}
+
+    rate = float(cfg.get("rate", 5))
+    duration = cfg.get("duration")
+    count = cfg.get("count")
+
+    # outputs
+    writers: Dict[str, Callable[[dict], None]] = {}
+    for out in cfg.get("outputs", []):
+        k = out.get("for", "*")
+        writers[k] = _writer_for_decl(out)
+
+    # partitioning
+    partition_by = cfg.get("partition_by", "none")
+
+    # build spec_str z drzewa devices
+    # devices: [{id: "engine-A", sensors: [{kind:"temp", count:2}, ...]}, ...]
+    parts = []
+    for dev in cfg.get("devices", []):
+        sid = dev["id"]
+        items = []
+        for s in dev.get("sensors", []):
+            kind = s["kind"]
+            cnt = int(s.get("count", 1))
+            items.append(f"{kind}*{cnt}")
+        parts.append(f"device={sid}: " + ",".join(items))
+    spec_str = "; ".join(parts)
+
+    run_stream(
+        spec_str,
+        rate_hz=rate,
+        duration_s=duration,
+        total_count=count,
+        writer_for_type=writers,
+        partition_by=partition_by,
+    )

src/dummysensors/io.py

@@ -0,0 +1,35 @@
+import os
+import json
+import csv
+import sys
+from typing import Callable
+from typing import Iterable
+
+def _ensure_dir(path: str) -> None:
+    d = os.path.dirname(path)
+    if d:
+        os.makedirs(d, exist_ok=True)
+
+def jsonl_writer(path: str | None) -> Callable[[dict], None]:
+    if not path:
+        # stdout
+        return lambda rec: (sys.stdout.write(json.dumps(rec) + "\n"), None)[1]
+    _ensure_dir(path)
+    f = open(path, "a", encoding="utf-8", buffering=1)
+    return lambda rec: (f.write(json.dumps(rec) + "\n"), None)[1]
+
+def csv_writer(path, header=("ts_ms","device_id","sensor_id","type","value")):
+    _ensure_dir(path)
+    f = open(path, "w", newline="", encoding="utf-8", buffering=1)
+    w = csv.writer(f)
+    cols = list(header)
+    w.writerow(cols)
+    f.flush()
+    os.fsync(f.fileno())
+
+    def _write(rec: dict) -> None:
+        w.writerow([rec.get(c, "") for c in cols])
+        f.flush()
+        os.fsync(f.fileno())
+
+    return _write

src/dummysensors/orchestrator.py

@@ -12,11 +12,12 @@ def run_stream(
     duration_s: float | None,
     total_count: int | None,
     writer_for_type: dict[str, Callable[[dict], None]],
+    partition_by: str = "none",  # <--- NEW
 ):
     period = 1.0 / rate_hz if rate_hz > 0 else 0.0
     devices = parse_spec(spec_str)
 
-    instances: list[tuple[str, str, str, object]] = []  # (device_id, sensor_id, type, sensor)
+    instances: list[tuple[str, str, str, object]] = []
     for d in devices:
         idx = {}
         for s in d.sensors:
@@ -27,22 +28,38 @@ def run_stream(
                 instances.append((d.id, sid, s.kind, sensor))
                 idx[s.kind] += 1
 
-    # writer default
-    default_writer = writer_for_type.get("*", lambda x: sys.stdout.write(json.dumps(x)+"\n"))
+    default_writer = writer_for_type.get("*", lambda x: (sys.stdout.write(json.dumps(x)+"\n"), None)[1])
 
-    def get_writer(k: str):
-        return writer_for_type.get(k, default_writer)
+    # cache writerów per klucz (type/device/*)
+    cache: dict[str, Callable[[dict], None]] = {}
+
+    def _key(rec: dict) -> str:
+        if partition_by == "device":
+            return rec["device_id"]
+        if partition_by == "type":
+            return rec["type"]
+        return "*"  # none
+
+    def _writer_for(rec: dict) -> Callable[[dict], None]:
+        k = _key(rec)
+        if k in cache:
+            return cache[k]
+        # jeżeli jest mapowanie po konkretnym typie, użyj go
+        if k in writer_for_type:
+            cache[k] = writer_for_type[k]
+        else:
+            cache[k] = default_writer
+        return cache[k]
 
-    # main loop
     t0 = time.perf_counter()
     n = total_count if total_count is not None else int((duration_s or 0) * rate_hz)
     for i in range(n):
         t_s = time.perf_counter() - t0
         for dev, sid, kind, sensor in instances:
             val = sensor.read(t_s)  # type: ignore[attr-defined]
             rec = {"ts_ms": int(t_s*1000), "device_id": dev, "sensor_id": sid, "type": kind, "value": val}
-            get_writer(kind)(rec)
+            _writer_for(rec)(rec)
         if period > 0:
             t_next = t0 + (i + 1) * period
             while time.perf_counter() < t_next:
-                time.sleep(min(0.002, t_next - time.perf_counter()))
+                time.sleep(min(0.002, t_next - time.perf_counter()))

tests/test_config.py

@@ -0,0 +1,58 @@
+import json
+import csv
+import os
+
+from dummysensors.config import run_from_config
+
+def test_config_yaml_creates_outputs(tmp_path):
+    # prepare config
+    cfg_path = tmp_path / "config.example.yaml"
+    cfg_path.write_text(
+        """
+rate: 2
+count: 5
+partition_by: type
+
+outputs:
+  - type: jsonl
+    for: temp
+    path: temp.jsonl
+  - type: csv
+    for: vibration
+    path: vibration.csv
+
+devices:
+  - id: engine-A
+    sensors:
+      - kind: temp
+        count: 1
+      - kind: vibration
+        count: 1
+"""
+    )
+
+    # run
+    cwd = os.getcwd()
+    os.chdir(tmp_path)
+    try:
+        run_from_config(str(cfg_path))
+    finally:
+        os.chdir(cwd)
+
+    # check outputs
+    temp_file = tmp_path / "temp.jsonl"
+    vib_file = tmp_path / "vibration.csv"
+
+    assert temp_file.exists()
+    assert vib_file.exists()
+
+    # check JSONL
+    with open(temp_file, encoding="utf-8") as f:
+        lines = [json.loads(line) for line in f]
+    assert all("value" in rec for rec in lines)
+
+    # check CSV
+    with open(vib_file, newline="", encoding="utf-8") as f:
+        reader = list(csv.reader(f))
+    assert reader[0] == ["ts_ms","device_id","sensor_id","type","value"]
+    assert len(reader) > 1