Посещаемост на културен и спортен комплекс: броене на входовете без проследяване на хора

01Какво ще научиш

• Как се строи система за посещаемост по принципа на минималните данни: само бройки, без идентификатори.
• Как nvdsanalytics брои пресичане на линия на вход и на изход и как числата стигат до базата на 15 минути.
• Как се подреждат данните в TimescaleDB (hypertable, непрекъснат агрегат) и как се държи Grafana.
• Три полезни метрики: пикови часове, оценка на заетостта и приход на посетител (EUR).
• Как се прави честна прогноза за посещаемост на събитие и седмичен текст — с проверка на числата.

02Преди да започнеш

• Машина от класа GB10 с DGX OS и Docker; безплатен акаунт и API ключ в NGC за образите на NVIDIA.
• Камера над входа (по възможност отгоре, така че лицата да не са целта на кадъра) и RTSP или файл за проба.
• Решение за поверителност (виж кутията по-горе) — преди реална камера.
• Python 3 с psycopg2-binary, pandas, scikit-learn, httpx; Ollama с модел (както в урок 04-150).
• Данни за билети (CSV експорт от билетната система) и календар на събитията.

03Стъпки

1. Какво измерваме и защо

   Целта е да отговорим на четири въпроса, а не да „събираме всичко“:

   • Кога е най-натоварено? (пикови часове — за персонал и почистване)
   • Колко хора са вътре приблизително? (оценка на заетостта)
   • Колко носи едно събитие на посетител? (приход на посетител, EUR)
   • Колко ще дойдат на следващото? (прогноза с интервал)

   Всяка метрика ползва само бройки от входовете и данни за билетите. Нищо не изисква да знаем кой е влязъл.

2. Брояч на входовете с DeepStream

   Конвейерът: източник на видео → детектор на хора (nvinfer) → тракер (nvtracker) → nvdsanalytics (пресичане на линия) → бройки. По документацията nvdsanalytics ползва за всички изчисления долната средна точка на рамката на обекта и поддържа пресичане на линия (line crossing) с посока: задаваш посоката и линията, а модулът прибавя към кадъра общия и текущия брой пресичания. Правилата са в конфигурационен файл; координатите са за размера, зададен с config-width/config-height, и се мащабират, ако потокът е с друга резолюция.

   ini · config_nvdsanalytics.txtкопирай

   [property]
   enable=1
   config-width=1920
   config-height=1080
   osd-mode=1
   display-font-size=12
   
   # Линия за влизане: посока (x1d;y1d;x2d;y2d) и линия (x1c;y1c;x2c;y2c) в координати на config-width/height
   [line-crossing-stream-0]
   enable=1
   line-crossing-Entry=960;400;960;700;700;540;1220;540
   class-id=0
   extended=0
   mode=balanced
   
   # Линия за излизане: същата линия, обратна посока
   [line-crossing-stream-1]
   enable=1
   line-crossing-Exit=960;700;960;400;700;540;1220;540
   class-id=0
   extended=0
   mode=balanced

   Две линии (влизане и излизане) са една и съща линия с обратна посока. class-id=0 е за класа „човек“ при някои модели — провери номера на класа за твоя модел, ⚠️ не сме го пускали.

   💡

   Бележки за DGX Spark (от документацията на NVIDIA)

   DeepStream 9.1 за DGX Spark се дава с отделен контейнер: nvcr.io/nvidia/deepstream:9.1-triton-sbsa-dgx-spark (нужен е вход в nvcr.io с твоя NGC ключ). В бележките към изданието: VIC не се поддържа — задай на мозайката (nvmultistreamtiler) compute-hw=1; предупреждението „Detected NVIDIA GB10 GPU, which is not yet supported in this version of the container“ е безвредно. Привързванията pyds са оттеглени — за нови приложения ползвай pyservicemaker.

   Как числата излизат от конвейера: според документацията резултатът от пресичането на линия е в метаданните на кадъра (NvDsAnalyticsFrameMeta). От там взимаш прирастите за влизане и излизане и ги подаваш на брояча по-долу — как точно (с pyservicemaker или с проба в C/C++), виж документацията на DeepStream; ⚠️ тази връзка не сме я пускали. Брояч, който държи бройките в паметта и записва един ред на 15 минути:

   python · counter15.pyкопирай

   from datetime import datetime, timezone
   
   import psycopg2
   
   
   class Counter15:
       """Събира входове/изходи в паметта и записва ЕДИН ред на 15 минути."""
   
       def __init__(self, dsn: str, location_id: int):
           self.conn = psycopg2.connect(dsn)
           self.location_id = location_id
           self.bucket = None
           self.entries = 0
           self.exits = 0
   
       @staticmethod
       def _bucket(ts: datetime) -> datetime:
           return ts.replace(minute=ts.minute - ts.minute % 15, second=0, microsecond=0)
   
       def add(self, entries: int = 0, exits: int = 0, now: datetime | None = None) -> None:
           now = now or datetime.now(timezone.utc)
           b = self._bucket(now)
           if self.bucket is not None and b != self.bucket:
               self.flush()
           self.bucket = b
           self.entries += entries
           self.exits += exits
   
       def flush(self) -> None:
           if self.bucket is None:
               return
           with self.conn, self.conn.cursor() as cur:
               cur.execute(
                   "INSERT INTO visitor_counts (time, location_id, entries, exits) VALUES (%s, %s, %s, %s)",
                   (self.bucket, self.location_id, self.entries, self.exits))
           self.bucket, self.entries, self.exits = None, 0, 0

3. Базата: TimescaleDB и Grafana в Docker

   TimescaleDB е разширение на PostgreSQL за времеви редове. Образът timescale/timescaledb:2.30.2-pg18 има версии за arm64 (проверено в Docker Hub към 03.10.2026). Тайните са в .env, портовете са само към 127.0.0.1, а данните — в именувани томове. PGDATA е закрепен, както в урок 04-01 (Postgres 18 смени мястото на данните).

   yaml · compose.yamlкопирай

   name: visitor-analytics
   
   services:
     db:
       image: timescale/timescaledb:2.30.2-pg18
       restart: unless-stopped
       environment:
         POSTGRES_USER: ${DB_USER}
         POSTGRES_PASSWORD: ${DB_PASSWORD}
         POSTGRES_DB: ${DB_NAME}
         PGDATA: /var/lib/postgresql/data
       ports:
         - "127.0.0.1:5432:5432"
       volumes:
         - db_data:/var/lib/postgresql/data
   
     grafana:
       image: grafana/grafana-oss:13.0.2
       restart: unless-stopped
       depends_on:
         - db
       environment:
         GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD}
       ports:
         - "127.0.0.1:3000:3000"
       volumes:
         - grafana_data:/var/lib/grafana
   
   volumes:
     db_data:
     grafana_data:

   bashкопирай

   cat > .env <<EOF
   DB_USER=visitors
   DB_PASSWORD=$(openssl rand -hex 24)
   DB_NAME=visitors
   GRAFANA_PASSWORD=$(openssl rand -hex 16)
   EOF
   chmod 600 .env
   docker compose up -d

   ⚠️

   Лиценз

   TimescaleDB има две издания — Apache 2 (образите с -oss) и Community (под лиценза на Timescale); по документацията не се смесват. Ние ползваме Community (образа без -oss), като по наша информация непрекъснатите агрегати са в него, но ⚠️ не сме я сверявали с лицензния текст — прочети условията за твоята употреба.

4. Таблиците и непрекъснатият агрегат

   visitor_counts е hypertable — таблица, разделена автоматично по време. Към нея има непрекъснат агрегат visitor_hourly с почасови суми, обновяван по политика на всеки 15 минути. Събитията и билетите са малки, обикновени таблици. Билетите внасяш от CSV, например в psql: \copy ticket_sales FROM 'sales.csv' CSV HEADER.

   sql · psqlкопирай

   CREATE EXTENSION IF NOT EXISTS timescaledb;
   
   -- Броене: един ред на вход и на 15 минути. Никакви идентификатори на хора.
   CREATE TABLE visitor_counts (
       time        timestamptz NOT NULL,
       location_id integer     NOT NULL,        -- 1 = главен вход, 2 = зала А …
       entries     integer     NOT NULL DEFAULT 0 CHECK (entries >= 0),
       exits       integer     NOT NULL DEFAULT 0 CHECK (exits >= 0)
   ) WITH (timescaledb.hypertable);
   
   -- Почасово обобщение (непрекъснат агрегат)
   CREATE MATERIALIZED VIEW visitor_hourly
   WITH (timescaledb.continuous) AS
   SELECT time_bucket('1 hour', time) AS hour,
          location_id,
          sum(entries) AS entries,
          sum(exits)   AS exits
   FROM visitor_counts
   GROUP BY time_bucket('1 hour', time), location_id
   WITH NO DATA;
   
   SELECT add_continuous_aggregate_policy('visitor_hourly',
       start_offset      => INTERVAL '3 hours',
       end_offset        => INTERVAL '1 hour',
       schedule_interval => INTERVAL '15 minutes');
   
   -- Събития и билети — обикновени таблици (малко редове)
   CREATE TABLE events (
       id                  serial PRIMARY KEY,
       name                text NOT NULL,
       event_type          text NOT NULL,        -- concert | conference | sport | exhibition
       location_id         integer NOT NULL,
       start_time          timestamptz NOT NULL,
       end_time            timestamptz NOT NULL,
       expected_capacity   integer NOT NULL,
       marketing_spend_eur numeric(10,2) NOT NULL DEFAULT 0
   );
   
   CREATE TABLE ticket_sales (
       sale_time   timestamptz NOT NULL,
       event_id    integer NOT NULL REFERENCES events(id),
       ticket_type text NOT NULL,                -- standard | student | vip
       price_eur   numeric(10,2) NOT NULL,
       quantity    integer NOT NULL DEFAULT 1
   );

   Първоначалното попълване на агрегата за стари данни: CALL refresh_continuous_aggregate('visitor_hourly', NULL, now() - interval '1 hour'); (по документацията на TimescaleDB). Часовете в агрегата са в UTC; при показване превръщаме във Europe/Sofia.

5. Три метрики със SQL

   Внимание към заетостта: „влизания минус излизания“ е само оценка — малките грешки на камерата се трупат и броячът се отклонява. Затова я нулираме всеки ден и я калибрираме от време на време с ръчно броене. Приходът на посетител смята посетителите на едно събитие като влизанията на входа на съответната зала от час преди началото до края му.

   sql · psqlкопирай

   -- 1) Пикови часове — средно влизания на час, по ден от седмицата и час (местно време)
   SELECT extract(dow  FROM hour AT TIME ZONE 'Europe/Sofia')::int AS dow,
          extract(hour FROM hour AT TIME ZONE 'Europe/Sofia')::int AS hour_of_day,
          round(avg(entries)) AS avg_entries
   FROM visitor_hourly
   WHERE location_id = 1
     AND hour > now() - interval '90 days'
   GROUP BY 1, 2
   ORDER BY avg_entries DESC
   LIMIT 5;
   
   -- 2) Оценка на заетостта — натрупани влизания минус излизания, нулира се всеки ден
   SELECT time, location_id,
          sum(entries - exits) OVER (
              PARTITION BY location_id, (time AT TIME ZONE 'Europe/Sofia')::date
              ORDER BY time) AS occupancy_est
   FROM visitor_counts
   WHERE location_id = 1
   ORDER BY time
   LIMIT 3;
   
   -- 3) Посетители и приход (EUR) на събитие
   SELECT e.id, e.name, e.event_type, e.expected_capacity,
          coalesce(v.visitors, 0)                              AS visitors,
          coalesce(t.revenue_eur, 0)                           AS revenue_eur,
          round(coalesce(t.revenue_eur, 0) / nullif(v.visitors, 0), 2) AS eur_per_visitor
   FROM events e
   LEFT JOIN LATERAL (
       SELECT sum(c.entries) AS visitors
       FROM visitor_counts c
       WHERE c.location_id = e.location_id
         AND c.time >= e.start_time - interval '1 hour'
         AND c.time <  e.end_time
   ) v ON true
   LEFT JOIN LATERAL (
       SELECT sum(s.price_eur * s.quantity) AS revenue_eur
       FROM ticket_sales s WHERE s.event_id = e.id
   ) t ON true
   ORDER BY e.start_time DESC;

   Заявките за анализ пуснахме върху обикновен PostgreSQL 16 с измислени данни (почасовото обобщение там е обикновен изглед). Числата ти ще зависят от твоите данни.

6. Прогноза за посещаемост на събитие

   Случайна гора (RandomForestRegressor) на процесор. Защо не на видеокарта? Стотици събития се обучават за секунди на процесор; видеокартата не носи полза и добавя зависимости. Интервалът идва от разсейването на предсказанията на дърветата (10-и до 90-и процентил) — това е ориентир, не гаранция. Под около 20 събития кодът не прави оценка на точността, защото тя би била безсмислена.

   python · forecast.pyкопирай

   import numpy as np
   import pandas as pd
   from sklearn.ensemble import RandomForestRegressor
   from sklearn.model_selection import cross_val_score
   
   NUM = ["dow", "month", "expected_capacity", "marketing_spend_eur"]
   
   
   def matrix(df: pd.DataFrame, columns=None) -> pd.DataFrame:
       x = pd.get_dummies(df[["event_type"] + NUM], columns=["event_type"], dtype=float)
       return x if columns is None else x.reindex(columns=columns, fill_value=0.0)
   
   
   def train(history: pd.DataFrame):
       """history: по един ред на събитие; колони event_type + NUM + visitors."""
       x, y = matrix(history), history["visitors"].astype(float)
       model = RandomForestRegressor(n_estimators=300, min_samples_leaf=2, random_state=42)
       if len(history) >= 20:                       # под ~20 събития оценката е безсмислена
           r2 = cross_val_score(model, x, y, cv=5, scoring="r2").mean()
           print(f"R2 (5-кратна кръстосана проверка): {r2:.2f}")
       else:
           print("Малко данни: само ориентир, без оценка на точността.")
       model.fit(x, y)
       return model, list(x.columns)
   
   
   def predict(model, columns, event: dict) -> dict:
       x = matrix(pd.DataFrame([event]), columns).to_numpy()
       per_tree = np.array([t.predict(x)[0] for t in model.estimators_])
       cap = event["expected_capacity"]
       mid = float(per_tree.mean())
       return {"expected": round(mid), "low": round(float(np.percentile(per_tree, 10))),
               "high": round(float(np.percentile(per_tree, 90))),
               "fill_pct": round(100 * mid / cap, 1)}
   
   
   if __name__ == "__main__":
       # Измислени данни за проба — в твоята система тук идва историята на събитията от базата.
       rng = np.random.default_rng(0)
       n = 60
       history = pd.DataFrame({
           "event_type": rng.choice(["concert", "conference", "sport", "exhibition"], n),
           "dow": rng.integers(0, 7, n),
           "month": rng.integers(1, 13, n),
           "expected_capacity": rng.choice([300, 500, 800, 1200], n),
           "marketing_spend_eur": rng.integers(0, 3000, n),
       })
       base = history.expected_capacity * (0.45 + 0.0001 * history.marketing_spend_eur) + (history.event_type == "concert") * 100
       history["visitors"] = (base + rng.normal(0, 40, n)).clip(0)
   
       model, cols = train(history)
       print(predict(model, cols, {"event_type": "conference", "dow": 2, "month": 9,
                                   "expected_capacity": 800, "marketing_spend_eur": 1000}))

   При нас (измислени данни, случайни, но с фиксирано зърно) изходът беше: R2 (5-кратна кръстосана проверка): 0.93 и {'expected': 473, 'low': 422, 'high': 570, 'fill_pct': 59.1}. Високото R² е, защото сме направили данните лесни; в живота очаквай много по-ниско и не вярвай на прогноза, която не си сравнил със случилото се. Числата може да се различават малко между версиите на пакетите.

7. Седмичен текст — чернова за човек

   Седмичната справка идва от SQL (влизания, предходна седмица, събития и приход в EUR); процентът на промяната се смята в кода, а не от модела. Локалният модел пише чернова; втората стойност, която функцията връща, са числа в текста, които ги няма в данните — човекът ги проверява първо.

   python · weekly.pyкопирай

   import json
   from datetime import date, datetime, time, timedelta
   from zoneinfo import ZoneInfo
   
   import psycopg2
   
   TZ = ZoneInfo("Europe/Sofia")
   
   WEEKLY_SQL = """
   SELECT json_build_object(
     'entries_week',      (SELECT coalesce(sum(entries), 0) FROM visitor_counts
                           WHERE location_id = 1 AND time >= %(s0)s AND time < %(s1)s),
     'entries_prev_week', (SELECT coalesce(sum(entries), 0) FROM visitor_counts
                           WHERE location_id = 1 AND time >= %(p0)s AND time < %(s0)s),
     'events', (SELECT coalesce(json_agg(x ORDER BY x.visitors DESC), '[]'::json) FROM (
         SELECT e.name, e.event_type, e.expected_capacity,
                coalesce(v.visitors, 0)    AS visitors,
                coalesce(t.revenue_eur, 0) AS revenue_eur
         FROM events e
         LEFT JOIN LATERAL (SELECT sum(c.entries) AS visitors FROM visitor_counts c
                            WHERE c.location_id = e.location_id
                              AND c.time >= e.start_time - interval '1 hour'
                              AND c.time <  e.end_time) v ON true
         LEFT JOIN LATERAL (SELECT sum(s.price_eur * s.quantity) AS revenue_eur
                            FROM ticket_sales s WHERE s.event_id = e.id) t ON true
         WHERE e.start_time >= %(s0)s AND e.start_time < %(s1)s) x)
   );
   """
   
   
   def weekly_stats(dsn: str, monday: date) -> dict:
       s0 = datetime.combine(monday, time.min, TZ)
       p = {"s0": s0, "s1": s0 + timedelta(days=7), "p0": s0 - timedelta(days=7)}
       with psycopg2.connect(dsn) as conn, conn.cursor() as cur:
           cur.execute(WEEKLY_SQL, p)
           stats = cur.fetchone()[0]
       prev = stats["entries_prev_week"]
       stats["change_pct"] = round(100 * (stats["entries_week"] - prev) / prev, 1) if prev else None
       return stats

   python · weekly_text.pyкопирай

   import json
   import os
   import re
   
   import httpx
   
   OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434")
   MODEL = os.environ.get("REPORT_MODEL", "llama3.3:70b")
   SCHEMA = {"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]}
   
   
   def weekly_text(stats: dict) -> tuple:
       prompt = (
           "Напиши кратък управленски текст (150-200 думи) за седмичната посещаемост. "
           "Ползвай САМО числата от данните; не измисляй числа, имена и причини. "
           "Включи тенденцията спрямо предходната седмица, най-посещаваното събитие и една препоръка. "
           f"ДАННИ (JSON):\n{json.dumps(stats, ensure_ascii=False, default=str)}"
       )
       r = httpx.post(f"{OLLAMA_URL}/api/chat", timeout=300, json={
           "model": MODEL, "stream": False, "format": SCHEMA,
           "messages": [{"role": "user", "content": prompt}],
           "options": {"temperature": 0.3, "num_predict": 700},
       })
       r.raise_for_status()
       text = json.loads(r.json()["message"]["content"])["text"]
       allowed = {a.replace(",", ".") for a in re.findall(r"\d+(?:[.,]\d+)?", json.dumps(stats, default=str))}
       found = {n.replace(",", ".") for n in re.findall(r"\d+(?:[.,]\d+)?", text)}
       return text, sorted(found - allowed)           # втората част: числа за проверка от човек

   Както в урок 04-150: само обобщени числа към модела, схема в format, ниска температура и човек, който чете, преди да изпрати. Заявката към Ollama не е пускана от нас.

8. Табло в Grafana

   В Grafana (http://localhost:3000 на машината; отдалечен достъп — през SSH тунел, както в урок 04-01) добави източник от тип PostgreSQL: хост db:5432, базата и потребителя от .env. По документацията на Grafana, при инсталирано разширение TimescaleDB тя се разпознава сама при запис и ползва time_bucket в макроса $__timeGroup. Пет панела:

   Панел	Тип	Заявка
   Посетители днес	Stat	SELECT sum(entries) FROM visitor_counts WHERE location_id = 1 AND time >= date_trunc('day', now() AT TIME ZONE 'Europe/Sofia') AT TIME ZONE 'Europe/Sofia'
   Влизания на час	Time series	SELECT hour AS time, entries FROM visitor_hourly WHERE $__timeFilter(hour) AND location_id = 1 ORDER BY 1
   Пикови часове	Heatmap / table	(заявка 1 от стъпка 6)
   Приход на посетител (EUR) по тип събитие	Bar chart	(заявка 3 от стъпка 6, осреднена по event_type)
   Запълненост на събитията	Gauge	SELECT 100.0 * avg(v.visitors / e.expected_capacity) FROM events e JOIN LATERAL (SELECT sum(entries) AS visitors FROM visitor_counts c WHERE c.location_id = e.location_id AND c.time >= e.start_time - interval '1 hour' AND c.time < e.end_time) v ON true WHERE e.start_time > now() - interval '90 days'

   ⚠️ Таблото не е пускано от нас. При първо влизане потребителят е по подразбиране admin, а паролата е тази от GRAFANA_PASSWORD.

04Проверка

• Има решение за поверителност (основание, табели, срок), преди реална камера.
• От системата излизат само бройки на 15 минути; номерата на тракера не се записват.
• TimescaleDB и Grafana се вдигат; visitor_counts е hypertable, а visitor_hourly се обновява.
• Трите заявки връщат смислени числа върху твои или пробни данни.
• Заетостта се нулира всеки ден и е калибрирана с поне едно ръчно броене.
• Прогнозата показва интервал; седмичният текст се чете от човек, преди да бъде изпратен.

Тест

05Какво следва

06Източници

1. DeepStream: Gst-nvdsanalytics — пресичане на линия, конфигурация, нужда от тракер.
2. DeepStream 9.1: бележки към изданието · контейнери 🔒 локално — DGX Spark, compute-hw=1, оттеглен pyds.
3. EDPB: насоки 3/2019 за обработване на лични данни чрез видеоустройства · Регламент (ЕС) 2016/679 (GDPR).
4. TimescaleDB: hypertable · непрекъснат агрегат · политика за обновяване.
5. Docker Hub: timescale/timescaledb · grafana/grafana-oss — тагове и архитектури (arm64).
6. Grafana: източник PostgreSQL 🔒 локално — TimescaleDB, макроси.
7. scikit-learn: RandomForestRegressor · Ollama: /api/chat 🔒 локално.