Skip to content

Faster, more robust processes: shared feed I/O, base classes, tests and bug fixes #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
TrystanLea wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Faster, more robust processes: shared feed I/O, base classes, tests and bug fixes #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ca9fbb2
Add test harness and block based FeedReader/FeedWriter
TrystanLea Oct 9, 2026
763816e
Add MapProcess and InPlaceProcess base classes
TrystanLea Oct 9, 2026
95b0d0a
Add CombineProcess base class for multi input and accumulating processes
TrystanLea Oct 9, 2026
2cf4f69
Rebuild ModelHelper on FeedReader/FeedWriter, fix simulator resume bugs
TrystanLea Oct 9, 2026
9a839e1
basic_formula: evaluate formulas a block at a time, fix quirks
TrystanLea Oct 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
852 changes: 689 additions & 163 deletions common.php
View file
Open in desktop

Large diffs are not rendered by default.

6 changes: 5 additions & 1 deletion postprocess_run.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,11 @@
if ($process = $postprocess->pop_process_queue()) {
print json_encode($process)."\n";

$result = $process_classes[$process->params->process]->process($process->params);
try {
$result = $process_classes[$process->params->process]->process($process->params);
} catch (Throwable $e) {
$result = array("success"=>false, "message"=>$e->getMessage());
}

if ($result['success']) {
print "Success: ".$result['message']."\n";
Expand Down
108 changes: 20 additions & 88 deletions processes/accumulator.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,11 @@
<?php

class PostProcess_accumulator extends PostProcess_common
class PostProcess_accumulator extends CombineProcess
{
protected $inputs = array("input");
private $total;
private $value;

public function description() {
return array(
"name"=>"Accumulator",
Expand All @@ -16,95 +20,23 @@ public function description() {
);
}

public function process($params)
protected function reset($previous)
{
$result = $this->validate($params);
if (!$result["success"]) return $result;

$input_meta = getmeta($this->dir,$params->input);
$output_meta = getmeta($this->dir,$params->output);

// Check that output feed is empty or has same start time and interval
if ($output_meta->npoints>0) {
if ($input_meta->start_time != $output_meta->start_time) {
return array("success"=>false, "message"=>"start time mismatch");
}
if ($input_meta->interval != $output_meta->interval) {
return array("success"=>false, "message"=>"interval mismatch");
}
} else {
// Copies over start_time to output meta file
createmeta($this->dir,$params->output,$input_meta);
}

// If recent mode, check that input feed has more points than output feed
if ($params->process_mode=='recent' && $output_meta->npoints >= $input_meta->npoints) {
return array("success"=>true, "message"=>"output feed already up to date");
}

if (!$if = @fopen($this->dir.$params->input.".dat", 'rb')) {
return array("success"=>false, "message"=>"could not open input feed");
}

if (!$of = @fopen($this->dir.$params->output.".dat", 'c+')) {
return array("success"=>false, "message"=>"could not open output feed");
}

$buffer = "";
$total = 0;
$value = 0;
$start_pos = 0;

if ($params->process_mode=='from' && $output_meta->npoints>0) {
$start_pos = floor(($params->process_start - $input_meta->start_time) / $input_meta->interval);
if ($start_pos<0) $start_pos = 0;
if ($start_pos>$input_meta->npoints) {
return array("success"=>false, "message"=>"start time is after end of input feed");
}
}

if ($params->process_mode=='recent' && $output_meta->npoints>0) {
$start_pos = $output_meta->npoints;
}
$this->total = is_nan($previous) ? 0 : $previous;
$this->value = 0;
}

if ($start_pos>0) {
fseek($if,$start_pos*4);
fseek($of,($start_pos-1)*4);
$tmp = unpack("f",fread($of,4));
$total = $tmp[1]*1;
}

$filtered_count = 0;

for ($n=$start_pos; $n<$input_meta->npoints; $n++) {
$tmp = unpack("f",fread($if,4));

if (!is_nan($tmp[1])) $value = 1*$tmp[1];

// filter spurious values +-1M
if ($value>=$params->min_value && $value<$params->max_value) {
$total += $value;
$buffer .= pack("f",$total);
} else {
$filtered_count++;
// Missing values hold the last value. Values outside of the limits are treated
// as spurious and are not added.
protected function row($value)
{
if (!is_nan($value)) {
if ($value < $this->params->min_value || $value >= $this->params->max_value) {
return $this->total;
}
$this->value = $value;
}

fwrite($of,$buffer);

if ($filtered_count>0) {
print "Filtered count: $filtered_count\n";
}
fclose($of);
fclose($if);

$time = $input_meta->start_time + ($input_meta->npoints * $input_meta->interval);

$byteswritten = strlen($buffer);
if ($byteswritten>0) {
updatetimevalue($params->output,$time,$total);
}
return array("success"=>true, "message"=>"bytes written: ".$byteswritten.", last time value: ".$time." ".$total);

$this->total += $this->value;
return $this->total;
}
}
}
111 changes: 9 additions & 102 deletions processes/addfeeds.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
<?php

class PostProcess_addfeeds extends PostProcess_common
class PostProcess_addfeeds extends CombineProcess
{
protected $inputs = array("feedA", "feedB");

public function description() {
return array(
"name"=>"Add feeds",
Expand All @@ -15,106 +17,11 @@ public function description() {
);
}

public function process($params)
// Sum of both feeds, or whichever feed has data
protected function row($a, $b)
{
$result = $this->validate($params);
if (!$result["success"]) return $result;

$feedA_meta = getmeta($this->dir,$params->feedA);
$feedB_meta = getmeta($this->dir,$params->feedB);

// Find longest interval
$interval = $feedA_meta->interval;
if ($feedB_meta->interval>$interval) $interval = $feedB_meta->interval;
// Find latest start time
$start_time = $feedA_meta->start_time;
if ($feedB_meta->start_time>$start_time) $start_time = $feedB_meta->start_time;
// Round start time down to nearest interval
$start_time = floor($start_time / $interval) * $interval;

// Create output feed meta file
$out_meta = new stdClass();
$out_meta->start_time = $start_time;
$out_meta->interval = $interval;
createmeta($this->dir,$params->output,$out_meta);
$out_meta = getmeta($this->dir,$params->output);

// Find end time of input feeds
$feedA_end_time = $feedA_meta->start_time + ($feedA_meta->interval * $feedA_meta->npoints);
$feedB_end_time = $feedB_meta->start_time + ($feedB_meta->interval * $feedB_meta->npoints);

// Start time of this process
if ($params->process_mode=='recent') {
$start_time = $out_meta->start_time + ($out_meta->npoints * $out_meta->interval);
} else if ($params->process_mode=='from') {
$start_time = $params->process_start;
if ($start_time<$out_meta->start_time) $start_time = $out_meta->start_time;
// Align to the output interval and stay within the output, no gap
$start_time = $out_meta->start_time + floor(($start_time - $out_meta->start_time) / $out_meta->interval) * $out_meta->interval;
$out_end_time = $out_meta->start_time + ($out_meta->npoints * $out_meta->interval);
if ($start_time>$out_end_time) $start_time = $out_end_time;
} else {
$start_time = $out_meta->start_time;
}

// End time of this process
$end_time = $feedA_end_time;
if ($feedB_end_time<$end_time) $end_time = $feedB_end_time;

if ($start_time>=$end_time) {
return array("success"=>true, "message"=>"no new data to process");
}

// Open input and output feeds
if (!$feedA_fh = @fopen($this->dir.$params->feedA.".dat", 'rb')) {
return array("success"=>false, "message"=>"could not open feedA feed");
}
if (!$feedB_fh = @fopen($this->dir.$params->feedB.".dat", 'rb')) {
return array("success"=>false, "message"=>"could not open feedB feed");
}
if (!$out_fh = @fopen($this->dir.$params->output.".dat", 'c+')) {
return array("success"=>false, "message"=>"could not open output feed");
}
// Write from start_time: reprocess and from modes overwrite existing data
fseek($out_fh, (($start_time - $out_meta->start_time) / $out_meta->interval) * 4);

$buffer = "";
for ($time=$start_time; $time<$end_time; $time+=$interval)
{
$pos_feedA = floor(($time - $feedA_meta->start_time) / $feedA_meta->interval);
$pos_feedB = floor(($time - $feedB_meta->start_time) / $feedB_meta->interval);

$valueA = NAN;
$valueB = NAN;

if ($pos_feedA>=0 && $pos_feedA<$feedA_meta->npoints) {
fseek($feedA_fh,$pos_feedA*4);
$tmp = unpack("f",fread($feedA_fh,4));
$valueA = $tmp[1];
}

if ($pos_feedB>=0 && $pos_feedB<$feedB_meta->npoints) {
fseek($feedB_fh,$pos_feedB*4);
$tmp = unpack("f",fread($feedB_fh,4));
$valueB = $tmp[1];
}

$outval = NAN;
if (!is_nan($valueA)) $outval = $valueA;
if (!is_nan($valueB)) $outval = $valueB;
if (!is_nan($valueA) && !is_nan($valueB)) $outval = $valueB + $valueA;
$buffer .= pack("f",$outval);
}

fwrite($out_fh,$buffer);
fclose($out_fh);
fclose($feedA_fh);
fclose($feedB_fh);

$byteswritten = strlen($buffer);
if ($byteswritten>0) {
updatetimevalue($params->output,$time,$outval);
}
return array("success"=>true, "message"=>"bytes written: ".$byteswritten.", last time value: ".$time." ".$outval);
if (is_nan($a)) return $b;
if (is_nan($b)) return $a;
return $a + $b;
}
}
}
81 changes: 4 additions & 77 deletions processes/allownegative.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
<?php

class PostProcess_allownegative extends PostProcess_common
class PostProcess_allownegative extends MapProcess
{
public function description() {
return array(
Expand All @@ -15,81 +15,8 @@ public function description() {
);
}

public function process($params)
protected function value($value)
{
$result = $this->validate($params);
if (!$result["success"]) return $result;

$input_meta = getmeta($this->dir,$params->input);
$output_meta = getmeta($this->dir,$params->output);

// Check that output feed is empty or has same start time and interval
if ($output_meta->npoints>0) {
if ($input_meta->start_time != $output_meta->start_time) {
return array("success"=>false, "message"=>"start time mismatch");
}
if ($input_meta->interval != $output_meta->interval) {
return array("success"=>false, "message"=>"interval mismatch");
}
} else {
// Copies over start_time to output meta file
createmeta($this->dir,$params->output,$input_meta);
}

// If recent mode, check that input feed has more points than output feed
if ($params->process_mode=='recent' && $output_meta->npoints >= $input_meta->npoints) {
return array("success"=>true, "message"=>"output feed already up to date");
}

if (!$if = @fopen($this->dir.$params->input.".dat", 'rb')) {
return array("success"=>false, "message"=>"could not open input feed");
}

if (!$of = @fopen($this->dir.$params->output.".dat", 'c+')) {
return array("success"=>false, "message"=>"could not open output feed");
}

$buffer = "";
$start_pos = 0;

if ($params->process_mode=='from' && $output_meta->npoints>0) {
$start_pos = floor(($params->process_start - $input_meta->start_time) / $input_meta->interval);
if ($start_pos<0) $start_pos = 0;
if ($start_pos>$input_meta->npoints) {
return array("success"=>false, "message"=>"start time is after end of input feed");
}
}

if ($params->process_mode=='recent' && $output_meta->npoints>0) {
$start_pos = $output_meta->npoints;
}

if ($start_pos>0) {
fseek($if,$start_pos*4);
fseek($of,$start_pos*4);
}

for ($n=$start_pos; $n<$input_meta->npoints; $n++) {
$tmp = unpack("f",fread($if,4));

$value = NAN;
if (!is_nan($tmp[1])) {
$value = 1*$tmp[1];
if ($value>0) $value = 0;
}
$buffer .= pack("f",$value);
}

fwrite($of,$buffer);
fclose($of);
fclose($if);

$time = $input_meta->start_time + ($input_meta->npoints * $input_meta->interval);

$byteswritten = strlen($buffer);
if ($byteswritten>0) {
updatetimevalue($params->output,$time,$value);
}
return array("success"=>true, "message"=>"bytes written: ".$byteswritten.", last time value: ".$time." ".$value);
return $value > 0 ? 0 : $value;
}
}
}
Loading