Skip to content

Ship162

tangram_ship162

log module-attribute 

log = getLogger(__name__)

router module-attribute 

router = APIRouter(
    prefix="/ship162",
    tags=["ship162"],
    responses={404: {"description": "Not found"}},
)

plugin module-attribute 

plugin = Plugin(
    frontend_path="dist-frontend",
    routers=[router],
    config_class=ShipsConfig,
    frontend_config_class=ShipsFrontendConfig,
    into_frontend_config_function=into_frontend,
)

TrailColorOptions dataclass

Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
169
170
171
172
173
@dataclass(frozen=True)
class TrailColorOptions:
    by_attribute: Annotated[Literal["speed"], FrontendMutable()] = "speed"
    min: Annotated[float, FrontendMutable()] = 0.0
    max: Annotated[float, FrontendMutable()] = 14.0

by_attribute class-attribute instance-attribute 

by_attribute: Annotated[
    Literal["speed"], FrontendMutable()
] = "speed"

min class-attribute instance-attribute 

min: Annotated[float, FrontendMutable()] = 0.0

max class-attribute instance-attribute 

max: Annotated[float, FrontendMutable()] = 14.0

__init__ 

__init__(
    by_attribute: Annotated[
        Literal["speed"], FrontendMutable()
    ] = "speed",
    min: Annotated[float, FrontendMutable()] = 0.0,
    max: Annotated[float, FrontendMutable()] = 14.0,
) -> None

ShipsConfig dataclass

Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
@dataclass(frozen=True)
class ShipsConfig:
    ship162_channel: str = "ship162"
    history_table_name: str = "ship162"
    history_control_channel: str = "history:control"
    search_channel: str = "ship162:search"
    state_vector_expire: int = 600  # 10 minutes
    stream_interval_secs: float = 1.0
    log_level: str = "INFO"
    history_buffer_size: int = 100_000
    history_flush_interval_secs: int = 5
    history_optimize_interval_secs: int = 120
    history_optimize_target_file_size: int = 134217728
    history_vacuum_interval_secs: int = 120
    history_vacuum_retention_period_secs: int | None = 120
    history_checkpoint_interval: int = 10
    topbar_order: int = 100
    sidebar_order: int = 100
    trail_color: str | TrailColorOptions = TrailColorOptions()
    trail_alpha: float = 0.6

ship162_channel class-attribute instance-attribute 

ship162_channel: str = 'ship162'

history_table_name class-attribute instance-attribute 

history_table_name: str = 'ship162'

history_control_channel class-attribute instance-attribute 

history_control_channel: str = 'history:control'

search_channel class-attribute instance-attribute 

search_channel: str = 'ship162:search'

state_vector_expire class-attribute instance-attribute 

state_vector_expire: int = 600

stream_interval_secs class-attribute instance-attribute 

stream_interval_secs: float = 1.0

log_level class-attribute instance-attribute 

log_level: str = 'INFO'

history_buffer_size class-attribute instance-attribute 

history_buffer_size: int = 100000

history_flush_interval_secs class-attribute instance-attribute 

history_flush_interval_secs: int = 5

history_optimize_interval_secs class-attribute instance-attribute 

history_optimize_interval_secs: int = 120

history_optimize_target_file_size class-attribute instance-attribute 

history_optimize_target_file_size: int = 134217728

history_vacuum_interval_secs class-attribute instance-attribute 

history_vacuum_interval_secs: int = 120

history_vacuum_retention_period_secs class-attribute instance-attribute 

history_vacuum_retention_period_secs: int | None = 120

history_checkpoint_interval class-attribute instance-attribute 

history_checkpoint_interval: int = 10

topbar_order class-attribute instance-attribute 

topbar_order: int = 100

sidebar_order class-attribute instance-attribute 

sidebar_order: int = 100

trail_color class-attribute instance-attribute 

trail_color: str | TrailColorOptions = TrailColorOptions()

trail_alpha class-attribute instance-attribute 

trail_alpha: float = 0.6

__init__ 

__init__(
    ship162_channel: str = "ship162",
    history_table_name: str = "ship162",
    history_control_channel: str = "history:control",
    search_channel: str = "ship162:search",
    state_vector_expire: int = 600,
    stream_interval_secs: float = 1.0,
    log_level: str = "INFO",
    history_buffer_size: int = 100000,
    history_flush_interval_secs: int = 5,
    history_optimize_interval_secs: int = 120,
    history_optimize_target_file_size: int = 134217728,
    history_vacuum_interval_secs: int = 120,
    history_vacuum_retention_period_secs: int | None = 120,
    history_checkpoint_interval: int = 10,
    topbar_order: int = 100,
    sidebar_order: int = 100,
    trail_color: str
    | TrailColorOptions = TrailColorOptions(),
    trail_alpha: float = 0.6,
) -> None

ShipsFrontendConfig dataclass

Bases: HasTopbarUiConfig, HasSidebarUiConfig

Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
198
199
200
201
202
203
204
205
206
@dataclass(frozen=True)
class ShipsFrontendConfig(
    tangram_core.config.HasTopbarUiConfig, tangram_core.config.HasSidebarUiConfig
):
    search_channel: str
    topbar_order: Annotated[int, FrontendMutable()]
    sidebar_order: Annotated[int, FrontendMutable()]
    trail_color: Annotated[str, FrontendMutable(kind="color")] | TrailColorOptions
    trail_alpha: Annotated[float, Ge(0), Le(1), FrontendMutable()]

search_channel instance-attribute 

search_channel: str

topbar_order instance-attribute 

topbar_order: Annotated[int, FrontendMutable()]

sidebar_order instance-attribute 

sidebar_order: Annotated[int, FrontendMutable()]

trail_color instance-attribute 

trail_color: (
    Annotated[str, FrontendMutable(kind="color")]
    | TrailColorOptions
)

trail_alpha instance-attribute 

trail_alpha: Annotated[
    float, Ge(0), Le(1), FrontendMutable()
]

__init__ 

__init__(
    search_channel: str,
    topbar_order: Annotated[int, FrontendMutable()],
    sidebar_order: Annotated[int, FrontendMutable()],
    trail_color: Annotated[
        str, FrontendMutable(kind="color")
    ]
    | TrailColorOptions,
    trail_alpha: Annotated[
        float, Ge(0), Le(1), FrontendMutable()
    ],
) -> None

get_trajectory_data async 

get_trajectory_data(
    mmsi: int, backend_state: InjectBackendState
) -> Response

Get the full trajectory for a given ship MMSI.

Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@router.get("/data/{mmsi}")
async def get_trajectory_data(
    mmsi: int, backend_state: tangram_core.InjectBackendState
) -> Response:
    """Get the full trajectory for a given ship MMSI."""
    if not _HISTORY_AVAILABLE:
        raise HTTPException(
            status_code=501,
            detail="History feature is not installed.",
        )

    redis_key = "tangram:history:table_uri:ship162"
    table_uri_bytes = await backend_state.redis_client.get(redis_key)

    if not table_uri_bytes:
        raise HTTPException(status_code=404, detail="Table 'ship162' not found.")
    table_uri = table_uri_bytes.decode("utf-8")

    try:
        df = (
            pl.scan_delta(table_uri)
            .filter(pl.col("mmsi") == mmsi)
            .with_columns(pl.col("timestamp").dt.epoch(time_unit="s"))
            .sort("timestamp")
            .collect()
        )
        return Response(df.write_json(), media_type="application/json")
    except Exception as e:
        raise HTTPException(
            status_code=500, detail=f"Failed to query trajectory data: {e}"
        )

search_ships async 

search_ships(
    q: str, backend_state: InjectBackendState
) -> Response
Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@router.get("/search")
async def search_ships(
    q: str, backend_state: tangram_core.InjectBackendState
) -> Response:
    if not _HISTORY_AVAILABLE:
        return ORJSONResponse(content=[])

    redis_key = "tangram:history:table_uri:ship162"
    table_uri_bytes = await backend_state.redis_client.get(redis_key)
    if not table_uri_bytes:
        return ORJSONResponse(content=[])
    table_uri = table_uri_bytes.decode("utf-8")

    try:
        df = pl.scan_delta(table_uri)
        q_lower = q.lower()

        candidates = (
            df.filter(
                pl.col("ship_name").str.to_lowercase().str.contains(q_lower)
                | pl.col("mmsi").cast(pl.String).str.contains(q_lower)
                | pl.col("callsign").str.to_lowercase().str.contains(q_lower)
            )
            .select("mmsi")
            .unique()
            .head(20)
            .collect()
        )
        candidate_mmsis = candidates["mmsi"].to_list()

        if not candidate_mmsis:
            return ORJSONResponse(content=[])

        intervals = (
            df.filter(pl.col("mmsi").is_in(candidate_mmsis))
            .sort(["mmsi", "timestamp"])
            .with_columns(pl.col("ship_name").forward_fill().over("mmsi"))
            .with_columns(
                gap_minutes=(
                    pl.col("timestamp") - pl.col("timestamp").shift(1)
                ).dt.total_minutes()
            )
            .with_columns(
                new_interval=(
                    (pl.col("gap_minutes") >= 60)
                    | (pl.col("mmsi") != pl.col("mmsi").shift(1))
                ).fill_null(True)
            )
            .with_columns(interval_id=pl.col("new_interval").cast(pl.Int64).cum_sum())
            .group_by(["mmsi", "ship_name", "interval_id"])
            .agg(
                start_ts=pl.col("timestamp").min(),
                end_ts=pl.col("timestamp").max(),
                n_rows=pl.len(),
                lat=pl.col("latitude").mean(),
                lon=pl.col("longitude").mean(),
            )
            .filter((pl.col("n_rows") >= 5))
            .with_columns(
                duration=((pl.col("end_ts") - pl.col("start_ts")).dt.total_seconds()),
            )
            .sort(["start_ts"], descending=True)
            .collect()
        )
        return Response(intervals.write_json(), media_type="application/json")
    except Exception as e:
        log.error(f"Search failed: {e}")
        return ORJSONResponse(content=[])

get_history_slice async 

get_history_slice(
    mmsi: int,
    start_timestamp: int,
    end_timestamp: int,
    backend_state: InjectBackendState,
) -> Response
Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@router.get("/history/{mmsi}/{start_timestamp}/{end_timestamp}")
async def get_history_slice(
    mmsi: int,
    start_timestamp: int,
    end_timestamp: int,
    backend_state: tangram_core.InjectBackendState,
) -> Response:
    if not _HISTORY_AVAILABLE:
        return ORJSONResponse(content=[])

    redis_key = "tangram:history:table_uri:ship162"
    table_uri_bytes = await backend_state.redis_client.get(redis_key)
    if not table_uri_bytes:
        return ORJSONResponse(content=[])
    table_uri = table_uri_bytes.decode("utf-8")

    start_dt = pl.lit(start_timestamp).cast(pl.Datetime("ms"))
    end_dt = pl.lit(end_timestamp).cast(pl.Datetime("ms"))

    try:
        df = (
            pl.scan_delta(table_uri)
            .filter(
                (pl.col("mmsi") == mmsi)
                & (pl.col("timestamp") >= start_dt)
                & (pl.col("timestamp") <= end_dt)
            )
            .sort("timestamp")
            .collect()
        )
        return Response(df.write_json(), media_type="application/json")
    except Exception as e:
        log.error(f"History slice failed: {e}")
        return ORJSONResponse(content=[])

into_frontend 

into_frontend(config: ShipsConfig) -> ShipsFrontendConfig
Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
209
210
211
212
213
214
215
216
def into_frontend(config: ShipsConfig) -> ShipsFrontendConfig:
    return ShipsFrontendConfig(
        search_channel=config.search_channel,
        topbar_order=config.topbar_order,
        sidebar_order=config.sidebar_order,
        trail_color=config.trail_color,
        trail_alpha=config.trail_alpha,
    )

run_ships async 

run_ships(backend_state: BackendState) -> None
Source code in packages/tangram_ship162/src/tangram_ship162/__init__.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@plugin.register_service()
async def run_ships(backend_state: tangram_core.BackendState) -> None:
    from . import _ships

    plugin_config = backend_state.config.plugins.get("tangram_ship162", {})
    config_ships = TypeAdapter(ShipsConfig).validate_python(plugin_config)

    default_log_level = plugin_config.get(
        "log_level", backend_state.config.core.log_level
    )

    _ships.init_tracing_stderr(default_log_level)

    rust_config = _ships.ShipsConfig(
        redis_url=backend_state.config.core.redis_url,
        ship162_channel=config_ships.ship162_channel,
        history_control_channel=config_ships.history_control_channel,
        state_vector_expire=config_ships.state_vector_expire,
        stream_interval_secs=config_ships.stream_interval_secs,
        history_table_name=config_ships.history_table_name,
        history_buffer_size=config_ships.history_buffer_size,
        history_flush_interval_secs=config_ships.history_flush_interval_secs,
        history_optimize_interval_secs=config_ships.history_optimize_interval_secs,
        history_optimize_target_file_size=config_ships.history_optimize_target_file_size,
        history_vacuum_interval_secs=config_ships.history_vacuum_interval_secs,
        history_vacuum_retention_period_secs=config_ships.history_vacuum_retention_period_secs,
        history_checkpoint_interval=config_ships.history_checkpoint_interval,
        search_channel=config_ships.search_channel,
    )
    await _ships.run_ships(rust_config)

tangram_ship162._ships

ShipsConfig

redis_url property writable 

redis_url: str

ship162_channel property writable 

ship162_channel: str

history_control_channel property writable 

history_control_channel: str

state_vector_expire property writable 

state_vector_expire: int

stream_interval_secs property writable 

stream_interval_secs: float

history_table_name property writable 

history_table_name: str

history_buffer_size property writable 

history_buffer_size: int

history_flush_interval_secs property writable 

history_flush_interval_secs: int

history_optimize_interval_secs property writable 

history_optimize_interval_secs: int

history_optimize_target_file_size property writable 

history_optimize_target_file_size: int

history_vacuum_interval_secs property writable 

history_vacuum_interval_secs: int

history_vacuum_retention_period_secs property writable 

history_vacuum_retention_period_secs: Optional[int]

history_checkpoint_interval property writable 

history_checkpoint_interval: int

search_channel property writable 

search_channel: str

__new__ 

__new__(
    redis_url: str,
    ship162_channel: str,
    history_control_channel: str,
    state_vector_expire: int,
    stream_interval_secs: float,
    history_table_name: str,
    history_buffer_size: int,
    history_flush_interval_secs: int,
    history_optimize_interval_secs: int,
    history_optimize_target_file_size: int,
    history_vacuum_interval_secs: int,
    history_vacuum_retention_period_secs: Optional[int],
    history_checkpoint_interval: int,
    search_channel: str,
) -> ShipsConfig

init_tracing_stderr 

init_tracing_stderr(filter_str: str) -> None

run_ships 

run_ships(config: ShipsConfig) -> Any