Package 'cs9'

Description

Diagnostic function to check CS9 environment configuration. This function validates required environment variables and database connectivity, providing detailed feedback for troubleshooting configuration issues.

Usage

check_environment_setup(verbose = TRUE, use_startup_message = FALSE)

Arguments

Details

CS9 requires specific environment variables for database connectivity and configuration. This function checks for:

• Required environment variables (CS9_DBCONFIG_ACCESS, CS9_DBCONFIG_DRIVER, etc.)

• Database configuration availability

• Database table initialization status

When CS9 is installed from CRAN without database configuration, the package loads with limited functionality. Use this function to diagnose what needs to be configured for full database-driven surveillance functionality.

Value

A list containing environment setup status and recommendations:

• status: "ok", "warning", or "error"

• issues: Character vector of identified problems

• recommendations: Character vector of suggested fixes

See Also

The installation vignette: vignette("installation", package = "cs9")

Examples

# Check environment setup with verbose output
check_environment_setup()

# Check silently and examine results
result <- check_environment_setup(verbose = FALSE)
if(result$status != "ok") {
  cat("Issues found:", result$issues, "\n")
  cat("Recommendations:", result$recommendations, "\n")
}

Description

Flags/values to be used in the 'dashboards' scene

Usage

config

Format

An object of class environment of length 7.

Description

Creates a directory and all necessary parent directories if they don't already exist.

Usage

create_folder_if_doesnt_exist(path)

Arguments

Value

Character string containing the created directory path

Examples

## Not run: 
# Create a new directory
create_folder_if_doesnt_exist("/tmp/my_analysis/results")

## End(Not run)

Description

Copies results from a dated folder to a "latest" folder, providing easy access to the most recent analysis results.

Usage

create_latest_folder(results_folder_name, date)

Arguments

Details

This function copies all contents from output/results_folder_name/date to output/results_folder_name/latest, overwriting existing files.

Value

No return value. This function is called for its side effect of copying files from the dated folder to the latest folder.

Examples

## Not run: 
# Copy today's results to latest folder
create_latest_folder("covid_reports", "2024-01-15")

## End(Not run)

Description

Retrieves data hash configuration entries from the database table for tracking data changes across surveillance task plans and elements.

Usage

get_config_data_hash_for_each_plan(
  task = NULL,
  index_plan = NULL,
  element_tag = NULL
)

Arguments

Value

A data.table containing the filtered hash configuration entries with columns: task, index_plan, element_tag, date, datetime, element_hash, all_hash

Examples

## Not run: 
# Get all hash data for a specific task
get_config_data_hash_for_each_plan(task = "covid_analysis")

# Get hash data for specific task and plan
get_config_data_hash_for_each_plan(task = "covid_analysis", index_plan = 1)

## End(Not run)

Description

Retrieves configuration log entries from the config_log table with optional filtering by surveillance system identifier, task name, and date range.

Usage

get_config_log(ss = NULL, task = NULL, start_date = NULL, end_date = NULL)

Arguments

Details

The function retrieves entries from the config_log table in the current configuration. If date filters are provided, they are applied to the timestamp field of the log entries.

Value

A data.table containing the filtered log entries.

Examples

## Not run: 
# Get all log entries
get_config_log()

# Get logs for a specific surveillance system
get_config_log(ss = "weather")

# Get logs for a specific task and date range
get_config_log(task = "data_import", start_date = "2024-01-01", end_date = "2024-12-31")

## End(Not run)

Description

Retrieves the last updated timestamps for database tables from the configuration tracking system.

Usage

get_config_tables_last_updated(table_name = NULL)

Arguments

Value

A data.table containing last updated information with columns: table_name, last_updated_datetime, and other tracking metadata

Examples

## Not run: 
# Get last updated info for all tables
get_config_tables_last_updated()

# Get info for a specific table
get_config_tables_last_updated(table_name = "anon_covid_cases")

## End(Not run)

Description

Retrieves runtime statistics and performance metrics for surveillance tasks from the configuration database.

Usage

get_config_tasks_stats(task = NULL, last_run = FALSE)

Arguments

Value

A data.table containing task statistics with columns including: task, datetime, runtime_seconds, memory_usage, status, and other metrics

Examples

## Not run: 
# Get all task statistics
get_config_tasks_stats()

# Get statistics for a specific task
get_config_tasks_stats(task = "covid_analysis")

# Get only the last run for a task
get_config_tasks_stats(task = "covid_analysis", last_run = TRUE)

## End(Not run)

Description

Applies a set of standard include/exclude filters to a database table or data frame using the conventional surveillance field names. Each NULL argument is a no-op, so callers only need to supply the dimensions they actually want to restrict.

Usage

mandatory_db_filter(
  .data,
  granularity_time = NULL,
  granularity_time_not = NULL,
  granularity_geo = NULL,
  granularity_geo_not = NULL,
  country_iso3 = NULL,
  location_code = NULL,
  age = NULL,
  age_not = NULL,
  sex = NULL,
  sex_not = NULL
)

Arguments

Value

The filtered object in the same class as .data.

Examples

## Not run: 
# Inside a data_selector function, filter a remote table before collecting:
d <- schema$anon_covid_cases$tbl() |>
  mandatory_db_filter(
    granularity_time = "isoweek",
    granularity_geo  = "county",
    age_not          = "total"
  ) |>
  dplyr::collect()

## End(Not run)

Description

Constructs the appropriate folder path for surveillance system results, with automatic switching between production and interactive modes.

Usage

path(
  ...,
  create_dir = FALSE,
  trailing_slash = FALSE,
  auto = cs9::config$is_auto
)

Arguments

Value

Character string containing the constructed file path

Examples

## Not run: 
# Get basic output path
path("reports", "daily")

# Create directory if it doesn't exist
path("reports", "daily", create_dir = TRUE)

# Get path with trailing slash
path("reports", "daily", trailing_slash = TRUE)

## End(Not run)

Description

Positron-friendly counterpart to run_task_sequentially_as_rstudio_job_using_load_all().

Usage

run_task_sequentially_as_callr_bg_using_load_all(
  task_name,
  ss_prefix = "global$ss"
)

Arguments

Value

Invisibly returns the TaskJob R6 object (already started).

Examples

## Not run: 
job <- run_task_sequentially_as_callr_bg_using_load_all("my_task")
job$wait()

## End(Not run)

Description

Executes a surveillance task as an RStudio job using devtools::load_all() for package development. This function creates a temporary R script that loads the package and runs the specified task sequentially (cores = 1).

Usage

run_task_sequentially_as_rstudio_job_using_load_all(
  task_name,
  ss_prefix = "global$ss"
)

Arguments

Details

This function is primarily used during package development to test tasks interactively. It creates a temporary R script that:

• Loads the package using devtools::load_all()

• Sets the task to run with single core (cores = 1)

• Executes the task via the surveillance system

Value

No return value. This function is called for its side effect of launching an RStudio job that executes the surveillance task.

Examples

## Not run: 
# Run a task as RStudio job during development
run_task_sequentially_as_rstudio_job_using_load_all("covid_analysis")

# Use custom surveillance system prefix
run_task_sequentially_as_rstudio_job_using_load_all(
  "covid_analysis", 
  ss_prefix = "my_ss"
)

## End(Not run)

Description

Core R6 class for creating and managing disease surveillance systems. This class orchestrates database tables, tasks, and analyses for real-time epidemiological monitoring and public health surveillance.

Details

SurveillanceSystem_v9 provides infrastructure for:

• Database table management with automated logging

• Task scheduling and parallel execution

• Data validation and schema enforcement

• Configuration and performance monitoring

The surveillance system follows a structured approach:

1. Define database tables with add_table()

2. Configure surveillance tasks with add_task()

3. Execute tasks with run_task() or external schedulers

Public fields

Methods

Public methods

• SurveillanceSystem_v9$new()

• SurveillanceSystem_v9$add_table()

• SurveillanceSystem_v9$add_partitionedtable()

• SurveillanceSystem_v9$add_task()

• SurveillanceSystem_v9$get_task()

• SurveillanceSystem_v9$run_task()

• SurveillanceSystem_v9$shortcut_get_tables()

• SurveillanceSystem_v9$shortcut_get_argset()

• SurveillanceSystem_v9$shortcut_get_data()

• SurveillanceSystem_v9$shortcut_get_plans_argsets_as_dt()

• SurveillanceSystem_v9$shortcut_get_num_analyses()

• SurveillanceSystem_v9$clone()

Method new()

Usage

SurveillanceSystem_v9$new(
  name = "unspecified",
  implementation_version = "unspecified"
)

Arguments

Method add_table()

Add a table

Usage

SurveillanceSystem_v9$add_table(
  name_access,
  name_grouping = NULL,
  name_variant = NULL,
  field_types,
  keys,
  indexes = NULL,
  validator_field_types = csdb::validator_field_types_blank,
  validator_field_contents = csdb::validator_field_contents_blank
)

Arguments

Returns

No return value. This method is called for its side effect of adding a table to the surveillance system.

Examples

\dontrun{
global$ss$add_table(
  name_access = c("anon"),
  name_grouping = "example_weather",
  name_variant = "data",
  field_types =  c(
    "granularity_time" = "TEXT",
    "granularity_geo" = "TEXT",
    "country_iso3" = "TEXT",
    "location_code" = "TEXT",
    "border" = "INTEGER",
    "age" = "TEXT",
    "sex" = "TEXT",

    "isoyear" = "INTEGER",
    "isoweek" = "INTEGER",
    "isoyearweek" = "TEXT",
    "season" = "TEXT",
    "seasonweek" = "DOUBLE",

    "calyear" = "INTEGER",
    "calmonth" = "INTEGER",
    "calyearmonth" = "TEXT",

    "date" = "DATE",

    "temp_max" = "DOUBLE",
    "temp_min" = "DOUBLE",
    "precip" = "DOUBLE"
  ),
  keys = c(
    "granularity_time",
    "location_code",
    "date",
    "age",
    "sex"
  ),
  validator_field_types = csdb::validator_field_types_csfmt_rts_data_v1,
  validator_field_contents = csdb::validator_field_contents_csfmt_rts_data_v1
)
}

Method add_partitionedtable()

Add a partitioned table to the surveillance system

Usage

SurveillanceSystem_v9$add_partitionedtable(
  name_access,
  name_grouping = NULL,
  name_variant = NULL,
  name_partitions = "default",
  column_name_partition = "partition",
  value_generator_partition = NULL,
  field_types,
  keys,
  indexes = NULL,
  validator_field_types = csdb::validator_field_types_blank,
  validator_field_contents = csdb::validator_field_contents_blank
)

Arguments

Returns

No return value. This method is called for its side effect of adding a partitioned table to the surveillance system.

Method add_task()

Add a surveillance task to the system

Usage

SurveillanceSystem_v9$add_task(
  name_grouping = NULL,
  name_action = NULL,
  name_variant = NULL,
  cores = 1,
  permission = NULL,
  plan_analysis_fn_name = NULL,
  for_each_plan = NULL,
  for_each_analysis = NULL,
  universal_argset = NULL,
  upsert_at_end_of_each_plan = FALSE,
  insert_at_end_of_each_plan = FALSE,
  action_fn_name,
  data_selector_fn_name = NULL,
  tables = NULL
)

Arguments

Returns

No return value. This method is called for its side effect of adding a task to the surveillance system.

Method get_task()

Get a surveillance task by name

Usage

SurveillanceSystem_v9$get_task(task_name)

Arguments

Returns

A Task R6 object representing the surveillance task

Method run_task()

Execute a surveillance task by name

Usage

SurveillanceSystem_v9$run_task(task_name)

Arguments

Returns

No return value. This method is called for its side effect of executing the task.

Method shortcut_get_tables()

Get database tables associated with a task

Usage

SurveillanceSystem_v9$shortcut_get_tables(task_name)

Arguments

Returns

A named list of database table objects used by the task

Method shortcut_get_argset()

Get argument set for a specific plan and analysis

Usage

SurveillanceSystem_v9$shortcut_get_argset(
  task_name,
  index_plan = 1,
  index_analysis = 1
)

Arguments

Returns

A named list containing the argument set for the specified plan and analysis

Method shortcut_get_data()

Get data for a specific plan

Usage

SurveillanceSystem_v9$shortcut_get_data(task_name, index_plan = 1)

Arguments

Returns

A named list containing the data extracted for the specified plan

Method shortcut_get_plans_argsets_as_dt()

Get plans and argsets as a data.table

Usage

SurveillanceSystem_v9$shortcut_get_plans_argsets_as_dt(task_name)

Arguments

Returns

A data.table containing plan and analysis information with columns including index_plan and index_analysis

Method shortcut_get_num_analyses()

Get the total number of analyses for a task

Usage

SurveillanceSystem_v9$shortcut_get_num_analyses(task_name)

Arguments

Returns

Integer value representing the total number of analyses across all plans for the task

Method clone()

The objects of this class are cloneable with this method.

Usage

SurveillanceSystem_v9$clone(deep = FALSE)

Arguments

Examples

## Not run: 
# Create surveillance system
ss <- SurveillanceSystem_v9$new(
  name = "covid_surveillance",
  implementation_version = "1.0"
)

# Add database table
ss$add_table(
  name_access = "anon",
  name_grouping = "covid",
  name_variant = "cases",
  field_types = c("date" = "DATE", "cases" = "INTEGER"),
  keys = c("date")
)

# Add surveillance task
ss$add_task(
  name_grouping = "covid",
  name_action = "import",
  name_variant = "daily_data",
  action_fn_name = "import_covid_data",
  data_selector_fn_name = "select_covid_sources"
)

# Run task
ss$run_task("covid_import_daily_data")

## End(Not run)


## ------------------------------------------------
## Method `SurveillanceSystem_v9$add_table`
## ------------------------------------------------

## Not run: 
global$ss$add_table(
  name_access = c("anon"),
  name_grouping = "example_weather",
  name_variant = "data",
  field_types =  c(
    "granularity_time" = "TEXT",
    "granularity_geo" = "TEXT",
    "country_iso3" = "TEXT",
    "location_code" = "TEXT",
    "border" = "INTEGER",
    "age" = "TEXT",
    "sex" = "TEXT",

    "isoyear" = "INTEGER",
    "isoweek" = "INTEGER",
    "isoyearweek" = "TEXT",
    "season" = "TEXT",
    "seasonweek" = "DOUBLE",

    "calyear" = "INTEGER",
    "calmonth" = "INTEGER",
    "calyearmonth" = "TEXT",

    "date" = "DATE",

    "temp_max" = "DOUBLE",
    "temp_min" = "DOUBLE",
    "precip" = "DOUBLE"
  ),
  keys = c(
    "granularity_time",
    "location_code",
    "date",
    "age",
    "sex"
  ),
  validator_field_types = csdb::validator_field_types_csfmt_rts_data_v1,
  validator_field_contents = csdb::validator_field_contents_csfmt_rts_data_v1
)

## End(Not run)

Description

Run a cs9 task in a background R process with live output streaming back to the current R console via later::later() polling. Equivalent in spirit to run_task_sequentially_as_rstudio_job_using_load_all() but does not depend on RStudio's job API (which Positron does not implement).

Methods: $start() spawn background process and begin polling $wait(t) block this session, draining the event loop, until done $is_alive() TRUE while the background process is running $status() summary $tail(n) last n lines of the log file (snapshot, not live) $kill() terminate the background process

Public fields

Methods

Public methods

• TaskJob$new()

• TaskJob$start()

• TaskJob$is_alive()

• TaskJob$status()

• TaskJob$wait()

• TaskJob$kill()

• TaskJob$tail()

• TaskJob$clone()

TaskJob$new()

Create a new TaskJob.

Usage

TaskJob$new(
  task_name,
  ss_prefix = "global$ss",
  log_dir = tempdir(check = TRUE)
)

Arguments

Returns

A new TaskJob object. Call $start() to launch the background process.

TaskJob$start()

Spawn the background process and begin polling its output.

Usage

TaskJob$start()

Returns

Invisibly returns the TaskJob object.

TaskJob$is_alive()

Check whether the background process is still running.

Usage

TaskJob$is_alive()

Returns

Logical. TRUE while the background process is alive.

TaskJob$status()

Print a brief status summary (task name, start time, alive status, log path).

Usage

TaskJob$status()

Returns

Invisibly returns the TaskJob object.

TaskJob$wait()

Block the current session until the background task finishes or the timeout expires, draining the later event loop while waiting.

Usage

TaskJob$wait(timeout_s = 600)

Arguments

Returns

Invisibly returns the TaskJob object.

TaskJob$kill()

Terminate the background process.

Usage

TaskJob$kill()

Returns

Invisibly returns the TaskJob object.

TaskJob$tail()

Print the last n lines of the task log file (snapshot, not live).

Usage

TaskJob$tail(n = 20)

Arguments

Returns

Invisibly returns the TaskJob object.

TaskJob$clone()

The objects of this class are cloneable with this method.

Usage

TaskJob$clone(deep = FALSE)

Arguments

Examples

## Not run: 
job <- TaskJob$new("my_task")
job$start()
job$status()
job$wait(timeout_s = 120)
job$tail(n = 30)

## End(Not run)

Description

Logs configuration updates with relevant metadata such as timestamp, session state, task name, and a custom message. The function inserts the log entry into the config_log table in the current configuration.

Usage

update_config_log(ss = "unspecified", task = "unspecified", ...)

Arguments

Details

The function records the type of interaction (automatic or interactive), session state, task description, and a user-provided message in the configuration log. It throws an error if the message argument is NULL.

Value

No return value; this function is called for its side effect of inserting a log entry into the config_log table.

Examples

## Not run: 
update_config_log(ss = "weather", task = "data_import", message = "Imported dataset successfully.")

## End(Not run)